SDS Client | spectrumx.client.Client

spectrumx.client.Client

Client(
    host: None | str,
    *,
    env_file: Path | None = None,
    env_config: Mapping[str, Any] | None = None,
    verbose: bool = False,
)

Instantiates an SDS client.

Used by:

• API Reference SDS Files API | <Client>.sds_files  sds_files Functions
  •  delete_file
  •  download_file
  •  get_file
  •  list_files
  •  upload_file

Methods:

Name	Description
authenticate	Authenticate the client.
delete_file	Deletes a file from SDS by its UUID.
download	Downloads files from SDS.
download_dataset	Downloads all files in a dataset using the existing download infrastructure.
download_file	Downloads a file from SDS: metadata and maybe contents.
get_file	Get a file instance by its ID. Only metadata is downloaded from SDS.
list_files	Lists files in a given SDS path.
upload	Uploads a file or directory to SDS.
upload_capture	Uploads a local directory and creates a capture using those files.
upload_file	Uploads a file to SDS.
upload_multichannel_drf_capture	Uploads a capture with multiple channels by running upload

Attributes:

Name	Type	Description
base_url	str	Base URL for the client.
base_url_no_port	str	Base URL without the port.
dry_run	bool	When in dry run mode, no SDS requests are made and files are not written.
verbose	bool	When True, shows verbose output.

Source code in spectrumx/client.py

def __init__(
    self,
    host: None | str,
    *,
    env_file: Path | None = None,
    env_config: Mapping[str, Any] | None = None,
    verbose: bool = False,
) -> None:
    # avoids circular imports
    from spectrumx.api import sds_files as _sds_files_api  # noqa: PLC0415
    from spectrumx.api import uploads as _uploads_api  # noqa: PLC0415

    self._sds_files = _sds_files_api
    self._uploads = _uploads_api

    self.host = host
    self.is_authenticated = False
    self._verbose = verbose
    self._config = SDSConfig(
        env_file=env_file,
        env_config=env_config,
        sds_host=host,
        verbose=self.verbose,
    )

    # initialize the gateway
    self._gateway = GatewayClient(
        host=self.host,
        api_key=self._config.api_key,
        timeout=self._config.timeout,
        verbose=self.verbose,
    )

    # create internal API instances
    self.captures = CaptureAPI(
        gateway=self._gateway,
        dry_run=self.dry_run,
    )
    self.datasets = DatasetAPI(
        gateway=self._gateway,
        dry_run=self.dry_run,
    )

    self._issue_user_alerts()

Attributes

base_url property

base_url: str

Base URL for the client.

base_url_no_port property

base_url_no_port: str

Base URL without the port.

dry_run property writable

dry_run: bool

When in dry run mode, no SDS requests are made and files are not written.

verbose property writable

verbose: bool

When True, shows verbose output.

Functions

authenticate

authenticate() -> None

Authenticate the client.

Source code in spectrumx/client.py

def authenticate(self) -> None:
    """Authenticate the client."""
    if self.dry_run:
        log_user("Dry run is enabled: assuming successful authentication")
        log_user("To authenticate against SDS, set Client.dry_run to False")
    else:
        log.warning(f"Dry run DISABLED: authenticating against '{self.host}'")
        self._gateway.authenticate()
    log.info("Authenticated successfully")
    self.is_authenticated = True

delete_file

delete_file(file_uuid: UUID4 | str) -> bool

Deletes a file from SDS by its UUID.

Parameters:

Name	Type	Description	Default
file_uuid	UUID4 | str	The UUID of the file to delete.	required

Returns: True if the file was deleted successfully, or if in dry run mode (simulating success). Raises: SDSError: If the file couldn't be deleted.

Source code in spectrumx/client.py

def delete_file(self, file_uuid: UUID4 | str) -> bool:
    """Deletes a file from SDS by its UUID.

    Args:
        file_uuid: The UUID of the file to delete.
    Returns:
        True if the file was deleted successfully,
        or if in dry run mode (simulating success).
    Raises:
        SDSError: If the file couldn't be deleted.
    """
    return self._sds_files.delete_file(client=self, file_uuid=file_uuid)

download

download(
    *,
    from_sds_path: PurePosixPath | Path | str | None = None,
    to_local_path: Path | str,
    files_to_download: list[File]
    | Paginator[File]
    | None = None,
    skip_contents: bool = False,
    overwrite: bool = False,
    verbose: bool = True,
) -> list[Result[File]]

Downloads files from SDS.

Parameters:

Name	Type	Description	Default
from_sds_path	PurePosixPath | Path | str | None	The virtual directory on SDS to download files from.	None
to_local_path	Path | str	The local path to save the downloaded files to.	required
files_to_download	list[File] | Paginator[File] | None	A paginator or list (in dry run mode) of files to download. If not provided, all files in the directory will be downloaded.	None
skip_contents	bool	When True, only the metadata is downloaded.	False
overwrite	bool	Whether to overwrite existing local files.	False
verbose	bool	Show a progress bar.	True

Returns: A list of results for each file discovered and downloaded.

Source code in spectrumx/client.py

def download(
    self,
    *,
    from_sds_path: PurePosixPath | Path | str | None = None,
    to_local_path: Path | str,
    files_to_download: list[File] | Paginator[File] | None = None,
    skip_contents: bool = False,
    overwrite: bool = False,
    verbose: bool = True,
) -> list[Result[File]]:
    """Downloads files from SDS.

    Args:
        from_sds_path:  The virtual directory on SDS to download files from.
        to_local_path:  The local path to save the downloaded files to.
        files_to_download:  A paginator or list (in dry run mode) of files to
            download. If not provided, all files in the directory will be
            downloaded.
        skip_contents:  When True, only the metadata is downloaded.
        overwrite:      Whether to overwrite existing local files.
        verbose:        Show a progress bar.
    Returns:
        A list of results for each file discovered and downloaded.
    """

    if from_sds_path is not None:
        if files_to_download is not None:
            error_msg = (
                "Both a path in the SDS and a list of files "
                "were provided: please provide only one."
            )
            raise ValueError(error_msg)
        from_sds_path = PurePosixPath(from_sds_path)
    to_local_path = Path(to_local_path)

    # Prepare the download environment
    self._prepare_download_directory(to_local_path)
    files_to_download = self._get_files_to_download(
        from_sds_path=from_sds_path,
        files_to_download=files_to_download,
        verbose=verbose,
    )

    # Download files
    return self._download_files(
        files_to_download=files_to_download,
        to_local_path=to_local_path,
        skip_contents=skip_contents,
        overwrite=overwrite,
        verbose=verbose,
    )

download_dataset

download_dataset(
    *,
    dataset_uuid: UUID4 | str,
    to_local_path: Path | str,
    skip_contents: bool = False,
    overwrite: bool = False,
    verbose: bool = True,
) -> list[Result[File]]

Downloads all files in a dataset using the existing download infrastructure.

This approach uses the get_dataset_files endpoint to get a paginated list of File objects and then uses the existing download() method with files_to_download parameter.

Parameters:

Name	Type	Description	Default
dataset_uuid	UUID4 | str	The UUID of the dataset to download.	required
to_local_path	Path | str	The local path to save the downloaded files to.	required
skip_contents	bool	When True, only the metadata is downloaded.	False
overwrite	bool	Whether to overwrite existing local files.	False
verbose	bool	Show progress bars and detailed output.	True

Returns: A list of results for each file downloaded.

Source code in spectrumx/client.py

def download_dataset(
    self,
    *,
    dataset_uuid: UUID4 | str,
    to_local_path: Path | str,
    skip_contents: bool = False,
    overwrite: bool = False,
    verbose: bool = True,
) -> list[Result[File]]:
    """Downloads all files in a dataset using the existing download infrastructure.

    This approach uses the get_dataset_files endpoint to get a paginated list of
    File objects and then uses the existing download() method with
    files_to_download parameter.

    Args:
        dataset_uuid: The UUID of the dataset to download.
        to_local_path: The local path to save the downloaded files to.
        skip_contents: When True, only the metadata is downloaded.
        overwrite: Whether to overwrite existing local files.
        verbose: Show progress bars and detailed output.
    Returns:
        A list of results for each file downloaded.
    """
    if isinstance(dataset_uuid, str):
        dataset_uuid = UUID(dataset_uuid)

    # Get all files in the dataset as a paginator
    files_to_download = self.datasets.get_files(dataset_uuid=dataset_uuid)

    if verbose:
        log_user(
            f"Downloading files from dataset "
            f"(total: {len(files_to_download)} files)"
        )

    # Create target directory
    to_local_path = Path(to_local_path)
    if not to_local_path.exists():
        if self.dry_run:
            log_user(f"Dry run: would create the directory '{to_local_path}'")
        else:
            to_local_path.mkdir(parents=True, exist_ok=True)

    # Use the existing download() method with the file list
    return self.download(
        to_local_path=to_local_path,
        files_to_download=files_to_download,
        skip_contents=skip_contents,
        overwrite=overwrite,
        verbose=verbose,
    )

download_file

download_file(
    *,
    file_instance: File | None = None,
    file_uuid: UUID4 | str | None = None,
    to_local_path: Path | str | None = None,
    skip_contents: bool = False,
    warn_missing_path: bool = True,
) -> File

Downloads a file from SDS: metadata and maybe contents.

Note this function always overwrites the local file, if it exists.

Either file_instance or file_uuid must be provided. When passing a file_instance, it's recommended to set its local_path attribute, otherwise the download will create a temporary file on disk.

Parameters:

Name	Type	Description	Default
file_instance	File | None	The file instance to download.	None
file_uuid	UUID4 | str | None	The UUID of the file to download.	None
to_local_path	Path | str | None	The local path to save the downloaded file to.	None
skip_contents	bool	When True, only the metadata is downloaded and no files are created on disk.	False
warn_missing_path	bool	Show a warning when the download location is undefined.	True

Returns: The file instance with updated attributes, or a sample when in dry run.

Source code in spectrumx/client.py

def download_file(
    self,
    *,
    file_instance: File | None = None,
    file_uuid: UUID4 | str | None = None,
    to_local_path: Path | str | None = None,
    skip_contents: bool = False,
    warn_missing_path: bool = True,
) -> File:
    """Downloads a file from SDS: metadata and maybe contents.

    > Note this function always overwrites the local file, if it exists.

    Either `file_instance` or `file_uuid` must be provided. When passing
    a `file_instance`, it's recommended to set its `local_path` attribute,
    otherwise the download will create a temporary file on disk.

    Args:
        file_instance:      The file instance to download.
        file_uuid:          The UUID of the file to download.
        to_local_path:      The local path to save the downloaded file to.
        skip_contents:      When True, only the metadata is downloaded
                                and no files are created on disk.
        warn_missing_path:  Show a warning when the download location is undefined.
    Returns:
        The file instance with updated attributes, or a sample when in dry run.
    """
    return self._sds_files.download_file(
        client=self,
        file_instance=file_instance,
        file_uuid=file_uuid,
        to_local_path=to_local_path,
        skip_contents=skip_contents,
        warn_missing_path=warn_missing_path,
    )

get_file

get_file(file_uuid: UUID4 | str) -> File

Get a file instance by its ID. Only metadata is downloaded from SDS.

Note this does not download the file contents from the server. File instances still need to have their contents downloaded to create a local copy - see Client.download_file().

Parameters:

Name	Type	Description	Default
file_uuid	UUID4 | str	The UUID of the file to retrieve.	required

Returns: The file instance, or a sample file if in dry run mode.

Source code in spectrumx/client.py

def get_file(self, file_uuid: UUID4 | str) -> File:
    """Get a file instance by its ID. Only metadata is downloaded from SDS.

    Note this does not download the file contents from the server. File
        instances still need to have their contents downloaded to create
        a local copy - see `Client.download_file()`.

    Args:
        file_uuid: The UUID of the file to retrieve.
    Returns:
        The file instance, or a sample file if in dry run mode.
    """

    return self._sds_files.get_file(client=self, file_uuid=file_uuid)

list_files

list_files(
    sds_path: PurePosixPath | Path | str,
    *,
    verbose: bool = False,
) -> Paginator[File]

Lists files in a given SDS path.

Parameters:

Name	Type	Description	Default
sds_path	PurePosixPath | Path | str	The virtual directory on SDS to list files from.	required
verbose	bool	Show network requests and other info.	False

Returns: A paginator for the files in the given SDS path.

Source code in spectrumx/client.py

def list_files(
    self, sds_path: PurePosixPath | Path | str, *, verbose: bool = False
) -> Paginator[File]:
    """Lists files in a given SDS path.

    Args:
        sds_path: The virtual directory on SDS to list files from.
        verbose:  Show network requests and other info.
    Returns:
        A paginator for the files in the given SDS path.
    """
    return self._sds_files.list_files(
        client=self, sds_path=sds_path, verbose=verbose
    )

upload

upload(
    *,
    local_path: Path | str,
    sds_path: PurePosixPath | Path | str = "/",
    verbose: bool = True,
    warn_skipped: bool = True,
    persist_state: bool = True,
) -> list[Result[File]]

Uploads a file or directory to SDS.

Parameters:

Name	Type	Description	Default
local_path	Path | str	The local path of the file or directory to upload.	required
sds_path	PurePosixPath | Path | str	The virtual directory on SDS to upload the file to, where '/' is the user root.	'/'
verbose	bool	Show a progress bar.	True
warn_skipped	bool	Display warnings for skipped files.	True
persist_state	bool	Whether to persist upload state for resumption.	True

Source code in spectrumx/client.py

def upload(
    self,
    *,
    local_path: Path | str,
    sds_path: PurePosixPath | Path | str = "/",
    verbose: bool = True,
    warn_skipped: bool = True,
    persist_state: bool = True,
) -> list[Result[File]]:
    """Uploads a file or directory to SDS.

    Args:
        local_path:     The local path of the file or directory to upload.
        sds_path:       The virtual directory on SDS to upload the file to, \
                            where '/' is the user root.
        verbose:        Show a progress bar.
        warn_skipped:   Display warnings for skipped files.
        persist_state:  Whether to persist upload state for resumption.
    """
    if self.ENABLE_NEW_UPLOAD_METHOD:
        return self._upload_resumable(
            local_path=local_path,
            sds_path=sds_path,
            verbose=verbose,
            warn_skipped=warn_skipped,
            persist_state=persist_state,
        )
    return self._upload_deprecated(
        local_path=local_path,
        sds_path=sds_path,
        verbose=verbose,
        warn_skipped=warn_skipped,
    )

upload_capture

upload_capture(
    *,
    local_path: Path | str,
    sds_path: PurePosixPath | Path | str = "/",
    capture_type: CaptureType,
    index_name: str = "",
    channel: str | None = None,
    scan_group: str | None = None,
    name: str | None = None,
    verbose: bool = True,
    warn_skipped: bool = False,
    raise_on_error: bool = True,
    persist_state: bool = True,
) -> Capture | None

Uploads a local directory and creates a capture using those files.

This method effectively combines Client.upload() and Client.captures.create_capture() into one call. For a more fine-grained control, call each method separately.

Parameters:

Name	Type	Description	Default
local_path	Path | str	The local path of the directory to upload.	required
sds_path	PurePosixPath | Path | str	The virtual directory on SDS to upload the file to.	'/'
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
verbose	bool	Show progress bar and failure messages, if any.	True
raise_on_error	bool	When True, raises an exception if any file upload fails. If False, the method will return None and log the errors.	True
persist_state	bool	Whether to persist upload state for resumption.	True

Returns: The created capture object when all operations succeed. Raises: When raise_on_error is True. FileError: If any file upload fails. CaptureError: If the capture creation fails.

Source code in spectrumx/client.py

def upload_capture(  # noqa: PLR0913
    self,
    *,
    local_path: Path | str,
    sds_path: PurePosixPath | Path | str = "/",
    capture_type: CaptureType,
    index_name: str = "",
    channel: str | None = None,
    scan_group: str | None = None,
    name: str | None = None,
    verbose: bool = True,
    warn_skipped: bool = False,
    raise_on_error: bool = True,
    persist_state: bool = True,
) -> Capture | None:
    """Uploads a local directory and creates a capture using those files.

    This method effectively combines `Client.upload()` and
        `Client.captures.create_capture()` into one call. For a more fine-grained
        control, call each method separately.

    Args:
        local_path:     The local path of the directory to upload.
        sds_path:       The virtual directory on SDS to upload the file to.
        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.
        verbose:        Show progress bar and failure messages, if any.
        raise_on_error: When True, raises an exception if any file upload fails.
                        If False, the method will return None and log the errors.
        persist_state:  Whether to persist upload state for resumption.
    Returns:
        The created capture object when all operations succeed.
    Raises:
        When `raise_on_error` is True.
        FileError:      If any file upload fails.
        CaptureError:   If the capture creation fails.
    """

    upload_results = self.upload(
        local_path=local_path,
        sds_path=sds_path,
        verbose=verbose,
        warn_skipped=warn_skipped,
        persist_state=persist_state,
    )

    if not process_upload_results(
        upload_results,
        raise_on_error=raise_on_error,
        verbose=verbose,
    ):
        return None

    try:
        capture = self.captures.create(
            top_level_dir=PurePosixPath(sds_path),
            capture_type=capture_type,
            index_name=index_name,
            channel=channel,
            scan_group=scan_group,
            name=name,
        )
    except SDSError:
        if raise_on_error:
            raise
        if verbose:
            log_user_error("Failed to create capture.")
        return None
    else:
        return capture

upload_file

upload_file(
    *,
    local_file: File | Path | str,
    sds_path: PurePosixPath | Path | str = "/",
) -> File

Uploads a file to SDS.

If the file instance passed already has a directory set, sds_path will be prepended. E.g. given: sds_path = 'baz'; and local_file.directory = 'foo/bar/', then: The file will be uploaded to baz/foo/bar/ (under the user root in SDS) and the returned file instance will have its directory attribute updated to match this new path.

Parameters:

Name	Type	Description	Default
local_file	File | Path | str	The local file to upload.	required
sds_path	PurePosixPath | Path | str	The virtual directory on SDS to upload the file to.	'/'

Returns: The file instance with updated attributes, or a sample when in dry run.

Source code in spectrumx/client.py

def upload_file(
    self,
    *,
    local_file: File | Path | str,
    sds_path: PurePosixPath | Path | str = "/",
) -> File:
    """Uploads a file to SDS.

    If the file instance passed already has a directory set, `sds_path` will
    be prepended. E.g. given:
        `sds_path = 'baz'`; and
        `local_file.directory = 'foo/bar/'`, then:
    The file will be uploaded to `baz/foo/bar/` (under the user root in SDS) and
        the returned file instance will have its `directory` attribute updated
        to match this new path.

    Args:
        local_file:     The local file to upload.
        sds_path:       The virtual directory on SDS to upload the file to.
    Returns:
        The file instance with updated attributes, or a sample when in dry run.
    """

    return self._sds_files.upload_file(
        client=self,
        local_file=local_file,
        sds_path=sds_path,
    )

upload_multichannel_drf_capture

upload_multichannel_drf_capture(
    *,
    local_path: Path | str,
    sds_path: PurePosixPath | Path | str = "/",
    channels: list[str],
    verbose: bool = True,
    warn_skipped: bool = False,
    raise_on_error: bool = True,
    persist_state: bool = True,
) -> list[Capture] | None

Uploads a capture with multiple channels by running upload once for the whole directory. Then, it creates a capture for each channel.

Note: This method is only for DigitalRF captures.

Parameters:

Name	Type	Description	Default
local_path	Path | str	The local path of the directory to upload.	required
sds_path	PurePosixPath | Path | str	The virtual directory on SDS to upload the file to.	'/'
channels	list[str]	The list of channels to create captures for.	required
verbose	bool	Show progress bar and failure messages, if any.	True
warn_skipped	bool	Display warnings for skipped files.	False
raise_on_error	bool	When True, raises an exception if any file upload fails. If False, the method will return an empty list or None and log the errors.	True
persist_state	bool	Whether to persist upload state for resumption.	True

Returns: A list of captures created for each channel, or an empty list if any capture creation fails or no channels are provided. Raises: When raise_on_error is True. FileError: If any file upload fails.

Source code in spectrumx/client.py

def upload_multichannel_drf_capture(
    self,
    *,
    local_path: Path | str,
    sds_path: PurePosixPath | Path | str = "/",
    channels: list[str],
    verbose: bool = True,
    warn_skipped: bool = False,
    raise_on_error: bool = True,
    persist_state: bool = True,
) -> list[Capture] | None:
    """Uploads a capture with multiple channels by running `upload`
    once for the whole directory. Then, it creates a capture for each channel.

    *Note: This method is only for DigitalRF captures.*

    Args:
        local_path: The local path of the directory to upload.
        sds_path: The virtual directory on SDS to upload the file to.
        channels: The list of channels to create captures for.
        verbose: Show progress bar and failure messages, if any.
        warn_skipped: Display warnings for skipped files.
        raise_on_error: When True, raises an exception if any file upload fails.
                        If False, the method will return an empty list
                        or None and log the errors.
        persist_state: Whether to persist upload state for resumption.
    Returns:
        A list of captures created for each channel,
        or an empty list if any capture creation fails or no channels are provided.
    Raises:
        When `raise_on_error` is True.
        FileError: If any file upload fails.
    """
    upload_results = self.upload(
        local_path=local_path,
        sds_path=sds_path,
        verbose=verbose,
        warn_skipped=warn_skipped,
        persist_state=persist_state,
    )

    if not process_upload_results(
        upload_results,
        raise_on_error=raise_on_error,
        verbose=verbose,
    ):
        return None

    captures: list[Capture] = []

    if len(channels) == 0:
        log_user_warning("No channels provided, skipping capture creation")
        return captures

    for channel in channels:
        try:
            capture = self.captures.create(
                top_level_dir=PurePosixPath(sds_path),
                capture_type=CaptureType.DigitalRF,
                channel=channel,
            )
        except SDSError as err:
            if verbose:
                log_user_error(
                    f"Failed to create multi-channel capture for channel: {channel}"
                )
            # If capture already exists, try to get it instead of
            # deleting all captures
            capture_found, existing_capture = self._handle_existing_capture_error(
                err
            )
            if capture_found:
                assert existing_capture is not None
                captures.append(existing_capture)
            else:
                # Cleanup any created captures
                for created_capture in captures:
                    if created_capture.uuid is not None:
                        self.captures.delete(capture_uuid=created_capture.uuid)

                return []
        else:
            captures.append(capture)
    return captures