Captures API | `<Client>.captures`

spectrumx.api.captures

API functions specific to captures.

Classes:

Name	Description
`CaptureAPI`

Classes

CaptureAPI

CaptureAPI(
    *,
    gateway: GatewayClient,
    dry_run: bool = True,
    verbose: bool = False,
)

Methods:

Name	Description
`advanced_search`	Advanced searches for RF captures in SDS.
`create`	Creates a new RF capture in SDS.
`delete`	Deletes a capture from SDS by its UUID.
`detach_from_datasets`	Remove this capture from all datasets (owner-only).
`listing`	Lists all RF captures in SDS under the current user.
`read`	Reads a specific capture from SDS.
`revoke_share_permissions`	Revoke all direct share permissions on this capture (owner-only).
`update`	Updates a capture in SDS by re-discovering and re-indexing the files.

Source code in spectrumx/api/captures.py

def __init__(
    self,
    *,
    gateway: GatewayClient,
    dry_run: bool = True,
    verbose: bool = False,
) -> None:
    """Initializes the CaptureAPI."""
    self.dry_run = dry_run
    self.gateway = gateway
    self.verbose = verbose

Methods:

advanced_search

advanced_search(
    *,
    field_path: str,
    query_type: str,
    filter_value: str | dict[str, Any],
) -> list[Capture]

Advanced searches for RF captures in SDS.

SDS querying examples: https://github.com/spectrumx/sds-code/tree/master/gateway#opensearch-query-tips

OpenSearch query DSL documentation: https://docs.opensearch.org/docs/latest/query-dsl/

Parameters:

Name	Type	Description	Default
`field_path`	`str`	The field to search captures by e.g.: `capture_props.center_freq`, `created_at`	required
`query_type`	`str`	The type of query to perform e.g.: `range`, `geo_bounding_box`, `term`, `match`, `wildcard`	required
`filter_value`	`str \| dict[str, Any]`	The value(s) to filter by e.g.: `{ "gte": 1990000000, "lte": 2010000000 }`	required

Returns: A list of captures matching the query.

Source code in spectrumx/api/captures.py

def advanced_search(
    self: CaptureAPI,
    *,
    field_path: str,
    query_type: str,
    filter_value: str | dict[str, Any],
) -> list[Capture]:
    """Advanced searches for RF captures in SDS.

    SDS querying examples:
    https://github.com/spectrumx/sds-code/tree/master/gateway#opensearch-query-tips

    OpenSearch query DSL documentation:
    https://docs.opensearch.org/docs/latest/query-dsl/

    Args:
        field_path:     The field to search captures by e.g.:
                        `capture_props.center_freq`, `created_at`
        query_type:     The type of query to perform e.g.:
                        `range`, `geo_bounding_box`, `term`, `match`, `wildcard`
        filter_value:   The value(s) to filter by e.g.:
                        ```
                        {
                            "gte": 1990000000,
                            "lte": 2010000000
                        }
                        ```
    Returns:
        A list of captures matching the query.
    """

    # TODO: adapt this function to return a Paginator[Capture] object

    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            "Dry run enabled: simulating search results"
        )
        rng = random.Random()  # noqa: S311
        return [
            _generate_capture(capture_type=rng.choice(list(CaptureType)))
            for _ in range(5)
        ]

    search_results_raw = self.gateway.captures_advanced_search(
        field_path=field_path,
        query_type=query_type,
        filter_value=filter_value,
    )
    search_results_obj = json.loads(search_results_raw)
    captures: list[Capture] = []
    if "results" not in search_results_obj:
        msg = "Unexpected search result format."
        raise CaptureError(msg)
    search_results = search_results_obj["results"]
    for result_raw in search_results:
        try:
            capture = Capture.model_validate(result_raw)
        except ValidationError as err:
            log_user_warning(
                f"Validation error loading search result: {result_raw}"
            )
            log.bind(cat=LogCategory.FILESYSTEM).exception(err)
            continue
        else:
            captures.append(capture)
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Search returned {len(captures)} captures"
        )
    return captures

create

create(
    *,
    top_level_dir: Path | PurePosixPath,
    capture_type: CaptureType,
    index_name: str = "",
    channel: str | None = None,
    scan_group: str | None = None,
    name: str | None = None,
) -> Capture

Creates a new RF capture in SDS.

An SDS capture is a collection of RF files that follow a predetermined structure which makes them suitable for indexing and searching. Example capture types include Digital-RF and RadioHound. The exact file structure and what files get indexed depend on the capture type.

Notes:

This method doesn't upload files, it is a lightweight method that only groups files already in SDS into a capture. To upload files, see upload() and upload_file() from your SDS client instance.
The top_level_dir is the path in SDS relative to your user directory where the files were uploaded to, not your local filesystem path.

Parameters:

Name	Type	Description	Default
`top_level_dir`	`Path \| PurePosixPath`	Virtual directory in SDS where capture files are stored.	required
`capture_type`	`CaptureType`	One of `spectrumx.models.captures.CaptureType`.	required
`index_name`	`str`	The SDS index name. Leave empty to automatically select.	`''`
`channel`	`str \| None`	(For Digital-RF) the DRF channel name to index.	`None`
`scan_group`	`str \| None`	(For RadioHound) UUIDv4 that groups RH files.	`None`
`name`	`str \| None`	Optional custom name for the capture.	`None`

Returns: The created capture object. Raises: CaptureError: If the capture couldn't be created e.g.: if it already exists, or the user doesn't have permission to create it.

Source code in spectrumx/api/captures.py

def create(
    self,
    *,
    top_level_dir: Path | PurePosixPath,
    capture_type: CaptureType,
    index_name: str = "",
    channel: str | None = None,
    scan_group: str | None = None,
    name: str | None = None,
) -> Capture:
    """Creates a new RF capture in SDS.

    An SDS capture is a collection of RF files that follow a predetermined structure
    which makes them suitable for indexing and searching. Example capture types
    include Digital-RF and RadioHound. The exact file structure and what files
    get indexed depend on the capture type.

    ### Notes:

    + This method doesn't upload files, it is a lightweight method that only groups
    files already in SDS into a capture. To upload files, see `upload()` and
    `upload_file()` from your SDS client instance.

    + The `top_level_dir` is the path in SDS relative to your user directory
    where the files were uploaded to, not your local filesystem path.

    Args:
        top_level_dir:  Virtual directory in SDS where capture files are stored.
        capture_type:   One of `spectrumx.models.captures.CaptureType`.
        index_name:     The SDS index name. Leave empty to automatically select.
        channel:        (For Digital-RF) the DRF channel name to index.
        scan_group:     (For RadioHound) UUIDv4 that groups RH files.
        name:           Optional custom name for the capture.
    Returns:
        The created capture object.
    Raises:
        CaptureError: If the capture couldn't be created e.g.: if it already exists,
            or the user doesn't have permission to create it.
    """
    if index_name:
        log.bind(cat=LogCategory.LOG).warning(
            "The 'index_name' parameter is deprecated and "
            "will be removed in future versions."
        )
    index_name = index_mapping.get(capture_type, index_name)
    if not index_name:
        log.bind(cat=LogCategory.LOG).warning(
            f"Could not find an index for {capture_type=}"
        )

    top_level_dir = PurePosixPath(top_level_dir)
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Creating capture with {top_level_dir=}, "
            f"{channel=}, {capture_type=}, {index_name=}, {scan_group=}, {name=}"
        )

    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            "Dry run enabled: simulating the capture creation"
        )
        return Capture(
            capture_props={},
            capture_type=capture_type,
            channel=channel,
            index_name=index_name,
            name=name or "",
            origin=CaptureOrigin.User,
            scan_group=uuid.UUID(scan_group) if scan_group else None,
            top_level_dir=top_level_dir,
            uuid=uuid.uuid4(),
            files=[],
            datasets=[],
            owner=User(name="Dry run", email="dry-run@local.invalid"),
            share_permissions=[],
            is_shared_with_me=False,
            is_shared=False,
            capture_start_iso_utc=None,
            capture_end_iso_utc=None,
            capture_start_display=None,
            capture_end_display=None,
            created_at=datetime.now(UTC),
        )
    capture_raw = self.gateway.create_capture(
        capture_type=capture_type,
        channel=channel,
        index_name=index_name,
        scan_group=scan_group,
        top_level_dir=top_level_dir,
        name=name,
    )
    capture = Capture.model_validate_json(capture_raw)
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Capture created with UUID {capture.uuid}"
        )
    return capture

delete

delete(capture_uuid: UUID) -> bool

Deletes a capture from SDS by its UUID.

Parameters:

Name	Type	Description	Default
`capture_uuid`	`UUID`	The UUID of the capture to delete.	required

Returns: True if the capture was deleted successfully, or if in dry run mode. Raises: CaptureError: If the capture couldn't be deleted e.g.: if it doesn't exist; if it has already been deleted; if this user doesn't own it.

Referenced by:

API Reference Captures API | <Client>.captures captures Classes CaptureAPI Methods:
- detach_from_datasets
- revoke_share_permissions

Source code in spectrumx/api/captures.py

def delete(
    self,
    capture_uuid: uuid.UUID,
) -> bool:
    """Deletes a capture from SDS by its UUID.

    Args:
        capture_uuid:   The UUID of the capture to delete.
    Returns:
        True if the capture was deleted successfully, or if in dry run mode.
    Raises:
        CaptureError: If the capture couldn't be deleted e.g.: if it doesn't exist;
            if it has already been deleted; if this user doesn't own it.
    """
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Deleting capture with UUID {capture_uuid}"
        )

    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Dry run enabled: would delete capture {capture_uuid}"
        )
        return True

    self.gateway.delete_capture(
        capture_uuid=capture_uuid,
    )
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Capture deleted with UUID {capture_uuid}"
        )
    return True

detach_from_datasets

detach_from_datasets(capture_uuid: UUID) -> bool

Remove this capture from all datasets (owner-only).

Use before delete when the capture is still linked to datasets.

Source code in spectrumx/api/captures.py

def detach_from_datasets(self, capture_uuid: uuid.UUID) -> bool:
    """Remove this capture from all datasets (owner-only).

    Use before [`delete`][spectrumx.api.captures.CaptureAPI.delete]
    when the capture is still linked to datasets.
    """
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Detaching capture {capture_uuid} from datasets"
        )
    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            "Dry run enabled: would detach capture from datasets"
        )
        return True
    self.gateway.detach_capture_from_datasets(capture_uuid=capture_uuid)
    return True

listing

listing(
    *,
    capture_type: CaptureType | None = None,
    page_size: int = 30,
) -> list[Capture]

Lists all RF captures in SDS under the current user.

Fetches all pages from the API, collecting captures across paginated results.

Note a capture must be manually ".create()-d" before it can be listed, even if all the files are uploaded to SDS.

Parameters:

Name	Type	Description	Default
`capture_type`	`CaptureType \| None`	The type of capture to list. If empty, it lists everything.	`None`
`page_size`	`int`	Number of results to request per API page (default: 30).	`30`

Returns: A list of the RF captures found owned by the requesting user.

Source code in spectrumx/api/captures.py

def listing(
    self,
    *,
    capture_type: CaptureType | None = None,
    page_size: int = 30,
) -> list[Capture]:
    """Lists all RF captures in SDS under the current user.

    Fetches all pages from the API, collecting captures across paginated results.

    Note a capture must be manually "`.create()`-d" before it can be listed, even
        if all the files are uploaded to SDS.

    Args:
        capture_type:   The type of capture to list. If empty, it lists everything.
        page_size:      Number of results to request per API page (default: 30).
    Returns:
        A list of the RF captures found owned by the requesting user.
    """
    if page_size < 1:
        msg = "page_size must be >= 1"
        raise ValueError(msg)

    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Listing captures of type {capture_type}"
        )
    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            "Dry run enabled: simulating capture listing"
        )
        num_captures: int = 3
        rng = random.Random()  # noqa: S311
        return [
            _generate_capture(
                capture_type=capture_type or rng.choice(list(CaptureType))
            )
            for _ in range(num_captures)
        ]

    max_pages = 1_000
    captures: list[Capture] = []
    page = 1
    while True:
        page_raw = self.gateway.list_captures(
            capture_type=capture_type,
            page=page,
            page_size=page_size,
        )
        captures_list_raw, has_more, _count = _extract_page_from_payload(page_raw)
        for capture_data in captures_list_raw:
            try:
                capture = Capture.model_validate(capture_data)
                captures.append(capture)
            except ValidationError as err:
                log_user_warning(
                    f"Validation error loading capture: {capture_data}"
                )
                log.bind(cat=LogCategory.FILESYSTEM).exception(err)
                continue
        if not has_more:
            break
        page += 1
        if page > max_pages:
            msg = (
                f"Pagination exceeded {max_pages} pages — "
                "possible API loop or misconfigured page_size"
            )
            log.bind(cat=LogCategory.FILESYSTEM).error(msg)
            raise CaptureError(msg)

    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Listing {len(captures)} captures"
        )
    return captures

read

read(capture_uuid: UUID) -> Capture

Reads a specific capture from SDS.

This differs from the .listing() method by retrieving more information about the capture, including the associated files' UUID, directory, and name; useful when working with a specific capture.

Parameters:

Name	Type	Description	Default
`capture_uuid`	`UUID`	The UUID of the capture to read.	required

Returns: The capture object.

Source code in spectrumx/api/captures.py

def read(
    self,
    capture_uuid: uuid.UUID,
) -> Capture:
    """Reads a specific capture from SDS.

    This differs from the `.listing()` method by retrieving more information about
        the capture, including the associated files' UUID, directory, and name;
        useful when working with a specific capture.

    Args:
        capture_uuid:   The UUID of the capture to read.
    Returns:
        The capture object.
    """
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Reading capture with UUID {capture_uuid}"
        )

    capture_raw = self.gateway.read_capture(capture_uuid=capture_uuid)
    capture = Capture.model_validate_json(capture_raw)
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Capture read with UUID {capture.uuid}"
        )
    return capture

revoke_share_permissions

revoke_share_permissions(capture_uuid: UUID) -> bool

Revoke all direct share permissions on this capture (owner-only).

Use this (or the web portal) before delete when the capture is shared.

Source code in spectrumx/api/captures.py

def revoke_share_permissions(self, capture_uuid: uuid.UUID) -> bool:
    """Revoke all direct share permissions on this capture (owner-only).

    Use this (or the web portal) before
    [`delete`][spectrumx.api.captures.CaptureAPI.delete]
    when the capture is shared.
    """
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Revoking share permissions for capture {capture_uuid}"
        )
    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            "Dry run enabled: would revoke capture share permissions"
        )
        return True
    self.gateway.revoke_capture_share_permissions(capture_uuid=capture_uuid)
    return True

update

update(capture_uuid: UUID) -> None

Updates a capture in SDS by re-discovering and re-indexing the files.

Note that, just like the .create(), this method doesn't upload files.

Parameters:

Name	Type	Description	Default
`capture_uuid`	`UUID`	The UUID of the capture to update.	required

Returns: None

Source code in spectrumx/api/captures.py

def update(
    self,
    capture_uuid: uuid.UUID,
) -> None:
    """Updates a capture in SDS by re-discovering and re-indexing the files.

    Note that, just like the `.create()`, this method doesn't upload files.

    Args:
        capture_uuid:   The UUID of the capture to update.
    Returns:
        None
    """
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Updating capture with UUID {capture_uuid}"
        )

    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Dry run enabled: simulating capture update {capture_uuid}"
        )
        return

    capture_raw = self.gateway.update_capture(
        capture_uuid=capture_uuid,
        verbose=True,
    )
    capture = Capture.model_validate_json(capture_raw)
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Capture updated with UUID {capture.uuid}"
        )

Captures API | <Client>.captures

spectrumx.api.captures

Classes

CaptureAPI

Methods:

advanced_search

create

Notes:

delete

detach_from_datasets

listing

read

revoke_share_permissions

update

Functions:

Captures API | `<Client>.captures`