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,
    log_file: Path | None = None,
)

Instantiates an SDS client.

Used by:

• API Reference SDS Files API | <Client>.sds_files  sds_files Functions:
  •  delete_file
  •  detach_file_from_datasets
  •  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.
detach_file_from_datasets	Detach a file from all datasets it is associated with (owner-only).
download	Downloads files from SDS.
download_dataset	Downloads files in a dataset using the existing download infrastructure.
download_file	Downloads a file from SDS: metadata and maybe contents.
download_single_file	Download a single file and return the result.
get_dataset	Fetch dataset metadata, captures, and artifact files from SDS.
get_file	Get a file instance by its ID. Only metadata is downloaded from SDS.
list_dataset_artifact_files	List dataset artifact files (not the full download manifest).
list_dataset_captures	List captures for a dataset (raw dicts; composite payloads supported).
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.
config	SDSConfig	The client configuration.
dry_run	bool	When in dry run mode, no SDS requests are made and files are not written.
gateway	GatewayClient	Underlying gateway client (for tests and advanced HTTP access).
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,
    log_file: Path | None = None,
) -> 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,
        log_file=log_file,
    )

    # Enable structured logging
    _log_path = self._config.log_file
    enable_structured_logging(log_path=_log_path)
    log_user(f"Structured log: {_log_path}")

    # Bind persistent context
    api_key_prefix = ""
    if self._config.api_key:
        api_key_prefix = self._config.api_key.split(".")[0]

    set_persistent_log_context(
        api_key_prefix=api_key_prefix,
        timeout=self._config.timeout,
    )

    # 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.

config property

config: SDSConfig

The client configuration.

dry_run property writable

dry_run: bool

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

gateway property

gateway: GatewayClient

Underlying gateway client (for tests and advanced HTTP access).

verbose property writable

verbose: bool

When True, shows verbose output.

Methods:

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.bind(cat=LogCategory.AUTH).warning(
            f"Dry run DISABLED: authenticating against '{self.host}'"
        )
        self._gateway.authenticate()
    log.bind(cat=LogCategory.AUTH).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,
    )

detach_file_from_datasets

detach_file_from_datasets(file_uuid: UUID4 | str) -> bool

Detach a file from all datasets it is associated with (owner-only).

Source code in spectrumx/client.py

def detach_file_from_datasets(self, file_uuid: UUID4 | str) -> bool:
    """Detach a file from all datasets it is associated with (owner-only)."""
    return self._sds_files.detach_file_from_datasets(
        client=self,
        file_uuid=file_uuid,
    )

download

download(
    *,
    from_sds_path: PurePosixPath | Path | str | None = None,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    to_local_path: Path | str,
    files_to_download: DownloadFileSource | 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
start_time	datetime | None	The start time to filter DRF capture files by	None
end_time	datetime | None	The end time to filter DRF capture files by	None
to_local_path	Path | str	The local path to save the downloaded files to.	required
files_to_download	DownloadFileSource | 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,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    to_local_path: Path | str,
    files_to_download: DownloadFileSource | 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.
        start_time:     The start time to filter DRF capture files by
        (optional, only applicable to DRF capture files).
        end_time:       The end time to filter DRF capture files by
        (optional, only applicable to DRF capture files).
        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,
        start_time=start_time,
        end_time=end_time,
        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,
    capture_uuids: Collection[UUID4 | str] | None = None,
    top_level_dirs: Collection[PurePosixPath | Path | str]
    | None = None,
    artifacts_only: bool = False,
    skip_contents: bool = False,
    overwrite: bool = False,
    verbose: bool = True,
) -> list[Result[File]]

Downloads 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
capture_uuids	Collection[UUID4 | str] | None	If set, only files linked to at least one of these capture UUIDs are downloaded (dataset artifacts with no capture are excluded).	None
top_level_dirs	Collection[PurePosixPath | Path | str] | None	If set, only files whose directory lies under one of these capture top_level_dir paths (as from :meth:list_dataset_captures) are included. Leading/trailing slashes are normalized.	None
artifacts_only	bool	If set, only return artifact files (not capture-linked files).	False
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

If both capture_uuids and top_level_dirs are set, a file is included when it matches either criterion (enforced on the gateway). In dry run mode, capture/path filters are ignored because simulated files are not capture-tagged.

Returns:

Type	Description
list[Result[File]]	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,
    capture_uuids: Collection[UUID4 | str] | None = None,
    top_level_dirs: Collection[PurePosixPath | Path | str] | None = None,
    artifacts_only: bool = False,
    skip_contents: bool = False,
    overwrite: bool = False,
    verbose: bool = True,
) -> list[Result[File]]:
    """Downloads 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.
        capture_uuids: If set, only files linked to at least one of these capture
            UUIDs are downloaded (dataset artifacts with no capture are excluded).
        top_level_dirs: If set, only files whose ``directory`` lies under one of
            these capture ``top_level_dir`` paths (as from
            :meth:`list_dataset_captures`) are included. Leading/trailing slashes
            are normalized.
        artifacts_only: If set, only return artifact files (not capture-linked
            files).
        skip_contents: When True, only the metadata is downloaded.
        overwrite: Whether to overwrite existing local files.
        verbose: Show progress bars and detailed output.

    If both ``capture_uuids`` and ``top_level_dirs`` are set, a file is included
    when it matches **either** criterion (enforced on the gateway). In dry run
    mode, capture/path filters are ignored because simulated files are not
    capture-tagged.

    Returns:
        A list of results for each file downloaded.
    """
    if isinstance(dataset_uuid, str):
        dataset_uuid = UUID(dataset_uuid)

    (
        filter_active,
        uuid_set,
        dir_prefixes_norm,
    ) = resolve_dataset_capture_filter_params(
        capture_uuids=capture_uuids,
        top_level_dirs=top_level_dirs,
        dry_run=self.dry_run,
    )

    if filter_active:
        uuid_nonempty = bool(uuid_set)
        dir_nonempty = bool(dir_prefixes_norm)
        if not uuid_nonempty and not dir_nonempty:
            files_to_download: DownloadFileSource = []
        elif uuid_nonempty and dir_nonempty:
            assert uuid_set is not None
            assert dir_prefixes_norm is not None
            files_to_download = self.datasets.get_files(
                dataset_uuid,
                capture_uuids=tuple(uuid_set),
                top_level_dirs=tuple(dir_prefixes_norm),
                artifacts_only=artifacts_only,
            )
        elif uuid_nonempty:
            assert uuid_set is not None
            files_to_download = self.datasets.get_files(
                dataset_uuid,
                capture_uuids=tuple(uuid_set),
                artifacts_only=artifacts_only,
            )
        else:
            assert dir_prefixes_norm is not None
            files_to_download = self.datasets.get_files(
                dataset_uuid,
                top_level_dirs=tuple(dir_prefixes_norm),
                artifacts_only=artifacts_only,
            )
        if verbose and (uuid_nonempty or dir_nonempty):
            log_user("Dataset capture filter active (applied on the gateway).")
    else:
        files_to_download = self.datasets.get_files(
            dataset_uuid,
            artifacts_only=artifacts_only,
        )

    if verbose and not filter_active:
        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,
    progress_callback: Callable[[int], None] | None = None,
) -> 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
progress_callback	Callable[[int], None] | None	Optional callable invoked with byte count as data is received from the stream (for progress tracking).	None

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,
    progress_callback: collections.abc.Callable[[int], None] | None = None,
) -> 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.
        progress_callback:      Optional callable invoked with byte count as
            data is received from the stream (for progress tracking).
    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,
        progress_callback=progress_callback,
    )

download_single_file

download_single_file(
    *,
    file_info: File,
    to_local_path: Path,
    skip_contents: bool,
    overwrite: bool,
    progress_callback: Callable[[int], None] | None = None,
) -> Result[File]

Download a single file and return the result.

Source code in spectrumx/client.py

def download_single_file(
    self,
    *,
    file_info: File,
    to_local_path: Path,
    skip_contents: bool,
    overwrite: bool,
    progress_callback: collections.abc.Callable[[int], None] | None = None,
) -> Result[File]:
    """Download a single file and return the result."""
    # Handle local files without UUID
    if file_info.uuid is None:
        msg = f"Skipping local file: {file_info.name}"
        log_user_warning(msg)
        return Result(
            exception=SDSError(msg),
            error_info={"file": file_info},
        )

    # Calculate local file path
    local_file_path = self._calculate_local_file_path(file_info, to_local_path)

    # Validate path is relative to target
    if not self._is_path_relative_to_target(local_file_path, to_local_path):
        msg = (
            f"Resolved path {local_file_path} is not relative to "
            f"{to_local_path}: skipping download."
        )
        log_user_warning(msg)
        return Result(
            exception=SDSError(msg),
            error_info={"file": file_info},
        )

    # skip file download when local exists and not overwriting
    if not skip_contents and local_file_path.exists() and not overwrite:
        log.bind(cat=LogCategory.DOWNLOAD).opt(depth=2).info(
            f"Download skipped: local file exists "
            f"(set overwrite=True to replace): '{local_file_path}'"
        )
        return Result(value=file_info)

    # never re-download identical files
    if not skip_contents and local_file_path.exists() and overwrite:
        if file_info.sum_blake3 == file_info.compute_sum_blake3():
            log.bind(cat=LogCategory.DOWNLOAD).opt(depth=2).info(
                f"Download skipped: local copy is identical to SDS version "
                f"(BLAKE3 match): '{local_file_path}'"
            )
            return Result(value=file_info)

    # download the file
    try:
        log.bind(cat=LogCategory.DOWNLOAD).debug(
            f"Downloading file contents to: {local_file_path}"
        )
        downloaded_file = self.download_file(
            file_uuid=file_info.uuid,
            to_local_path=local_file_path,
            skip_contents=skip_contents,
            progress_callback=progress_callback,
        )
        return Result(value=downloaded_file)
    except SDSError as err:
        log_user_error(f"Download failed: {err}")
        return Result(
            exception=err,
            error_info={"file": file_info},
        )

get_dataset

get_dataset(dataset_uuid: UUID4 | str) -> Dataset

Fetch dataset metadata, captures, and artifact files from SDS.

Source code in spectrumx/client.py

def get_dataset(self, dataset_uuid: UUID4 | str) -> Dataset:
    """Fetch dataset metadata, captures, and artifact files from SDS."""
    if isinstance(dataset_uuid, str):
        dataset_uuid = UUID(dataset_uuid)
    return self.datasets.get(dataset_uuid)

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_dataset_artifact_files

list_dataset_artifact_files(
    dataset_uuid: UUID4 | str,
) -> list[dict[str, Any]]

List dataset artifact files (not the full download manifest).

Source code in spectrumx/client.py

def list_dataset_artifact_files(
    self, dataset_uuid: UUID4 | str
) -> list[dict[str, Any]]:
    """List dataset artifact files (not the full download manifest)."""
    if isinstance(dataset_uuid, str):
        dataset_uuid = UUID(dataset_uuid)
    return self.datasets.list_artifact_files(dataset_uuid)

list_dataset_captures

list_dataset_captures(
    dataset_uuid: UUID4 | str,
) -> list[dict[str, Any]]

List captures for a dataset (raw dicts; composite payloads supported).

Source code in spectrumx/client.py

def list_dataset_captures(self, dataset_uuid: UUID4 | str) -> list[dict[str, Any]]:
    """List captures for a dataset (raw dicts; composite payloads supported)."""
    if isinstance(dataset_uuid, str):
        dataset_uuid = UUID(dataset_uuid)
    return self.datasets.list_captures(dataset_uuid)

list_files

list_files(
    sds_path: PurePosixPath | Path | str,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    *,
    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
start_time	datetime | None	Optional inclusive lower time bound for rf@*.h5 files (naive datetimes are treated as UTC). Requires end_time.	None
end_time	datetime | None	Optional inclusive upper time bound; requires start_time.	None
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,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    *,
    verbose: bool = False,
) -> Paginator[File]:
    """Lists files in a given SDS path.

    Args:
        sds_path: The virtual directory on SDS to list files from.
        start_time: Optional inclusive lower time bound for ``rf@*.h5`` files
            (naive datetimes are treated as UTC). Requires ``end_time``.
        end_time: Optional inclusive upper time bound; requires ``start_time``.
        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,
        start_time=start_time,
        end_time=end_time,
        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