SpectrumX Python SDK

https://sds.crc.nd.edu/sdk
Admin
SpectrumX Python SDK is a Python package that provides a simple interface for interacting with the...
Tokens:33,342
Snippets:181
Trust Score:10
Update:7 months ago
Show doc for...
Tokens:
Raw
### Install SpectrumX SDK

Source: https://sds.crc.nd.edu/sdk/getting-started

Instructions for installing the SpectrumX SDK using various package managers like uv, poetry, or pip. This is the first step to using the SDK in your project.

```shell
uv add spectrumx

# or one of:
#   poetry add spectrumx
#   pip install spectrumx
#   ...
```

--------------------------------

### Basic SpectrumX SDK Usage: Upload and Download

Source: https://sds.crc.nd.edu/sdk/getting-started

A Python example demonstrating the basic usage of the SpectrumX SDK. It covers client initialization, authentication, uploading files from a local directory to SDS, and downloading files from SDS to a local directory. It also includes notes on dry-run mode and potential issues with concurrent access.

```python
from pathlib import Path
from random import randint, random
from spectrumx.client import Client

# Example of files upload, listing, and download from SDS.

# NOTE: the SDS client-server interaction is stateless, so it is
#   not recommended to have multiple clients writing to the same
#   locations simultaneously, as they may overrule each other
#   and cause loss of data. See "Concurrent Access" in the
#   usage guide to learn more.
sds = Client(
    host="sds.crc.nd.edu",
    # env_file=Path(".env"),  # default, recommended to keep tokens out of version control
    # env_config={"SDS_SECRET_TOKEN": "my-custom-token"},  # alternative way to pass the access token
)

# when in dry-run (default), no changes are made to the SDS or the local filesystem
# to enable the changes, set dry_run to False, as in:
# sds.dry_run = False

# authenticate using either the token from
# the .env file or in the config passed in
sds.authenticate()

# local_dir has your own local files that will be uploaded to the SDS
reference_name: str = "my_spectrum_capture"
local_dir: Path = Path(reference_name)

# or, if the directory doesn't exist, let's create some fake data
if not local_dir.exists():
    local_dir.mkdir(exist_ok=True)
    num_files = 10
    for file_idx in range(num_files):
        num_lines = randint(10, 100)  # noqa: S311
        file_name = f"capture_{file_idx}.csv"
        with (local_dir / file_name).open(mode="w", encoding="utf-8") as file_ptr:
            fake_nums = [random() for _ in range(num_lines)]  # noqa: S311
            file_ptr.write("\n".join(map(str, fake_nums)))

# upload all files in a directory to the SDS
# sds.dry_run = False   # uncomment to actually upload the files
upload_results = sds.upload(
    local_path=local_dir,  # may be a single file or a directory
    sds_path=reference_name,  # files will be created under this virtual directory
    verbose=True,  # shows a progress bar (default)
)
success_results = [success for success in upload_results if success]
failed_results = [success for success in upload_results if not success]
assert len(failed_results) == 0, (
    f"No failed uploads should be present: {failed_results}"
)
log.debug(f"Uploaded {len(success_results)} assets.")

# download the files from an SDS directory
# sds.dry_run = False
local_downloads = Path("sds-downloads") / "files" / reference_name
sds.download(
    from_sds_path=reference_name,  # files will be downloaded from this virtual dir
    to_local_path=local_downloads,  # download to this location (it may be created)
    overwrite=False,  # do not overwrite local existing files (default)
    verbose=True,  # shows a progress bar (default)
)

if not sds.dry_run:
    print("Downloaded files:")
    for file_path in local_downloads.iterdir():
        print(file_path)
else:
    print("Turn off dry-run to download and write files.")

```

--------------------------------

### Configure SDS Authentication Token

Source: https://sds.crc.nd.edu/sdk/getting-started

Demonstrates how to set the SDS secret token for authentication. This can be done by creating a .env file or by setting an environment variable. The environment variable takes precedence.

```dotenv
SDS_SECRET_TOKEN=your-secret-token-no-quotes
```

```shell
# the env var takes precedence over the .env file
export SDS_SECRET_TOKEN=your-secret-token
```

--------------------------------

### Install SDS SDK using Package Managers

Source: https://sds.crc.nd.edu/sdk/faq

Instructions on how to install the SpectrumX SDK using popular Python package managers like uv, poetry, and pip. The SDK is available on the Python Package Index (PyPI).

```bash
uv add spectrumx
poetry add spectrumx
pip install spectrumx
```

--------------------------------

### Python Paginator Usage Example

Source: https://sds.crc.nd.edu/sdk/api/ops/pagination

This snippet demonstrates how to instantiate and use the Paginator class for file operations. It includes setting up a mock gateway, defining a list method with a mock response, and iterating through the paginated results. The example also shows how a paginator behaves after being fully consumed.

```python
import logging

from spectrumx import files
from spectrumx.gateway import GatewayClient
from spectrumx.ops.pagination import Paginator

log = logging.getLogger(__name__)

def _process_file_fake(my_file: files.File) -> None:
    """A fake function to simulate file processing."""
    pass

def main() -> None:
    """Usage example for paginator."""
    log.info("Running the main script.")
    file_paginator = Paginator[files.File](
        Entry=files.File,
        gateway=GatewayClient(
            host="localhost",
            api_key="does-not-matter-in-dry-run",
        ),
        list_method=lambda **kwargs: b'{"count": 25, "results": []}',  # Mock response
        list_kwargs={"sds_path": "/path/to/files"},
        page_size=10,
        dry_run=True,  # in dry-run this should always generate 2.5 pages
        verbose=True,
    )
    log.info(f"Total files matched: {len(file_paginator)}")
    processed_count: int = 0
    for my_file in file_paginator:
        log.info(f"Processing file: {my_file.name}")
        _process_file_fake(my_file)
        processed_count += 1
        # new pages are fetched automatically

    log.info(f"Processed {processed_count} / {len(file_paginator)} files.")

    log.info("Trying another loop:")
    for _my_file in file_paginator:
        msg = "This will not run, as the paginator was consumed."
        raise AssertionError(msg)
    log.info("No more files to process.")

    log.info("Paginator demo finished.")

```

--------------------------------

### SpectrumX SDK Error Handling: Authentication

Source: https://sds.crc.nd.edu/sdk/getting-started

This Python code snippet demonstrates how to handle authentication errors when using the SpectrumX SDK. It shows how to catch `NetworkError` for connection issues and `AuthError` for authentication failures during the `sds.authenticate()` call.

```python
# ======== Authentication ========
from pathlib import Path
from spectrumx.client import Client
from spectrumx.errors import AuthError, NetworkError

sds = Client(host="sds.crc.nd.edu")
try:
    sds.authenticate()
except NetworkError as err:
    print(f"Failed to connect to the SDS: {err}")
    # check your host= parameter and network connection
    # if you're hosting the SDS Gateway, make sure it is accessible
except AuthError as err:
    print(f"Failed to authenticate: {err}")
    # TODO: take action

```

--------------------------------

### Upload Files with Retry Logic (Python)

Source: https://sds.crc.nd.edu/sdk/getting-started

Uploads files from a local directory to SDS, with retry logic for network and service errors. It handles partial uploads, logs successes and failures, and implements exponential backoff for retries.

```python
from pathlib import Path
from time import sleep

# Assuming sds, log, File, Result, NetworkError, ServiceError, SDSError are defined elsewhere

local_dir: Path = Path("my_spectrum_files")
reference_name: str = "my_spectrum_files"
retries_left: int = 5
is_success: bool = False
uploaded_files: list[File] = []

while not is_success and retries_left > 0:
    try:
        retries_left -= 1

        # `sds.upload()` will restart a partial file transfer from zero,
        # but it won't re-upload already finished files.
        upload_results: list[Result[File]] = sds.upload(
            local_path=local_dir,
            sds_path=reference_name,
            verbose=True,
        )

        # Since `upload()` is a batch operation, some files may succeed and some
        #   may fail. The return value of `sds.upload` stored in `upload_results`
        #   is a list of `Result` objects:
        # A `Result` wraps either the value of a variable (in this case the File
        #   object that was uploaded) or an exception. Here's how we can check if
        #   there were any failed uploads:
        success_results = [success for success in upload_results if success]
        failed_results = [success for success in upload_results if not success]

        log.debug(f"Uploaded {len(success_results)} assets.")
        log.warning(f"Failed to upload {len(failed_results)} assets")

        # calling a successful result will return the value it holds
        uploaded_files = [result() for result in success_results]

        # And calling a failed result will raise the exception it holds.
        # Here we re-raise it to handle retries with the except blocks below,
        #   based on the exception raised:
        for result in failed_results:
            result()  # will raise

    except (NetworkError, ServiceError) as err:
        # NetworkError refers to connection issues between client and SDS Gateway
        # ServiceError refers to issues with the SDS Gateway itself (e.g. HTTP 500)
        # sleep longer with each retry, at least 5s, up to 5min
        sleep_time = max(5, 5 / (retries_left**2) * 60)
        log.error(f"Error: {err}")
        log.warning(f"Failed to reach the gateway; sleeping {sleep_time}s")
        if retries_left > 0:
            sleep(sleep_time)
        continue
    except SDSError as err:
        log.error(f"Another SDS error occurred: {err}")
        # other errors might include e.g. OSError
        #   if listed files cannot be found.
        # TODO: take action or break
        break

log.debug(f"Uploaded files: {uploaded_files}")

```

--------------------------------

### GET /captures/listing

Source: https://sds.crc.nd.edu/sdk/api/captures

Retrieves a list of all RF captures associated with the current user.

```APIDOC
## GET /captures/listing

### Description
Lists all RF captures in SDS under the current user.

### Method
GET

### Endpoint
/captures/listing

### Response
#### Success Response (200)
- **captures** (array) - A list of capture objects.

#### Response Example
[
  {
    "id": "550e8400-e29b-41d4-a716-446655440000",
    "name": "capture_01"
  }
]
```

--------------------------------

### GET /captures

Source: https://sds.crc.nd.edu/sdk/api/captures

Retrieves a list of all RF captures in SDS associated with the current user. Optionally filters by capture type.

```APIDOC
## GET /captures

### Description
Lists all RF captures in SDS under the current user. A capture must be manually created before it can be listed, even if all associated files have been uploaded.

### Method
GET

### Endpoint
`/captures`

### Parameters
#### Query Parameters
- **capture_type** (CaptureType | None) - Optional - The type of capture to list. If not provided, all capture types are listed.

### Request Example
```json
{
  "message": "GET request to list captures"
}
```

### Response
#### Success Response (200)
- **captures** (list[Capture]) - A list of Capture objects, each representing an RF capture owned by the user.
  - **uuid** (UUID) - The unique identifier for the capture.
  - **type** (CaptureType) - The type of the capture.
  - **created_at** (datetime) - The timestamp when the capture was created.
  - **updated_at** (datetime) - The timestamp when the capture was last updated.

#### Response Example
```json
{
  "captures": [
    {
      "uuid": "a1b2c3d4-e5f6-7890-1234-567890abcdef",
      "type": "RF",
      "created_at": "2023-10-27T10:00:00Z",
      "updated_at": "2023-10-27T10:05:00Z"
    },
    {
      "uuid": "f0e9d8c7-b6a5-4321-fedc-ba9876543210",
      "type": "RF",
      "created_at": "2023-10-26T15:30:00Z",
      "updated_at": "2023-10-26T15:30:00Z"
    }
  ]
}
```
```

--------------------------------

### SpectrumX Python SDK - Valid Result Constructors

Source: https://sds.crc.nd.edu/sdk/api/errors

Illustrates the valid ways to instantiate the `Result` class in Python. These examples cover successful outcomes, simple failures, and failures with additional error information.

```python
from typing import Any, Dict, Generic, TypeVar, Union

_unset = object()
T = TypeVar("T")

class KnownError(Exception):
    pass

class Result(Generic[T]):
    def __init__(
        self,
        *,
        value: T | object = _unset,
        exception: Exception | object = _unset,
        error_info: Dict[str, Any] | None = None,
    ):
        if value is not _unset and exception is not _unset:
            raise ValueError("Cannot provide both value and exception.")
        if value is _unset and exception is _unset:
            raise ValueError("Must provide either value or exception.")
        if isinstance(value, Exception):
            raise ValueError("Value cannot be an Exception.")
        if not isinstance(exception, Exception) and exception is not _unset:
            raise ValueError("Exception must be an instance of Exception.")
        if error_info is not None and exception is _unset:
            raise ValueError("error_info requires an exception.")

        self.value = value
        self.exception = exception
        self.error_info = error_info

# Valid examples:
result_success = Result(value=123)                               # success
result_success_none = Result(value=None)                              # success with None
result_failure = Result(exception=RuntimeError())                # simple failure
result_failure_info = Result(exception=RuntimeError(), error_info={}) # failure with extra info
result_failure_info_true = Result(exception=RuntimeError(), error_info=True)

print(f"Success result: {result_success.value}")
print(f"Failure result exception: {result_failure.exception}")

```

--------------------------------

### SpectrumX SDK Error Handling: File Operations

Source: https://sds.crc.nd.edu/sdk/getting-started

This Python code snippet illustrates error handling for file operations within the SpectrumX SDK. It imports necessary error types like `NetworkError`, `Result`, `SDSError`, and `ServiceError` for robust management of potential issues during file interactions.

```python
# ======== File operations ========

from time import sleep
from spectrumx.errors import NetworkError
from spectrumx.errors import Result
from spectrumx.errors import SDSError
from spectrumx.errors import ServiceError
from loguru import logger as log

```

--------------------------------

### Create Multi-Channel Captures with SDS Client

Source: https://sds.crc.nd.edu/sdk/common-workflows/capture-uploads

Creates a Digital-RF capture for each specified channel. It handles cases where captures might already exist, logging them and retrieving existing capture objects. If creation fails for other reasons, it logs an error and provides guidance for resolution.

```python
def create_multichannel_captures(
    *, 
    client: Client,
    sds_path: PurePosixPath,
    channels: list[str],
) -> list[Capture]:
    """Create a Digital-RF capture for each channel.

    Args:
        client:   Authenticated SDS client.
        sds_path: Virtual path in SDS where files were uploaded.
        channels: List of Digital-RF channel names.

    Returns:
        List of successfully created Capture objects (may be shorter than channels).
    """
    captures: list[Capture] = []
    for ch in channels:
        try:
            capture = client.captures.create(
                top_level_dir=sds_path,
                capture_type=CaptureType.DigitalRF,
                channel=ch,
            )
        except CaptureError as err:
            existing_uuid_str = err.extract_existing_capture_uuid()
            if existing_uuid_str:
                log.info(
                    f"Capture for channel '{ch}' already exists: {existing_uuid_str}"
                )
                captures.append(
                    client.captures.read(capture_uuid=UUID(existing_uuid_str))
                )
            else:
                _report_capture_failure(err, channel=ch)
        else:
            log.success(f"Capture for channel '{ch}' created: {capture.uuid}")
            captures.append(capture)

    return captures
```

--------------------------------

### Initialize the Paginator class

Source: https://sds.crc.nd.edu/sdk/api/ops/pagination

The constructor for the paginator, requiring an SDSModel subclass, a gateway client, and specific configuration parameters for fetching pages.

```python
def __init__(
    self,
    *,
    Entry: type[SDSModel],
    gateway: GatewayClient,
    list_method: Callable[..., bytes],
    list_kwargs: dict[str, Any],
    dry_run: bool = False,
    page_size: int = 30,
    start_page: int = 1,
    total_matches: int | None = None,
    verbose: bool = False,
) -> None:
    self.dry_run = dry_run
    self._Entry = Entry
    self._gateway = gateway
    self._list_method = list_method
    self._list_kwargs = copy.deepcopy(list_kwargs)
    self._next_page = start_page
    self._page_size = page_size
    self._total_matches = total_matches or 1
    self._verbose: bool = verbose
```

--------------------------------

### Main Function for SDS CRC ND EDU SDK Usage

Source: https://sds.crc.nd.edu/sdk/common-workflows/capture-uploads

Sets up the SDS client, defines local and SDS paths, discovers Digital-RF channels, uploads files, and then creates single or multi-channel captures based on the discovered channels. It includes error handling for missing local directories and provides feedback on the upload and capture creation process.

```python
def main() -> None:
    # ── client setup ──────────────────────────────────────────────
    client = Client(
        host="sds.crc.nd.edu",
        # TODO: create a .env file with your SDS API token for authentication
        # env_file=Path(".env"),  # default
    )
    client.dry_run = False  # ⚠️ required to actually upload files
    client.authenticate()

    # ── paths ─────────────────────────────────────────────────────
    # TODO: set these to match your local Digital-RF directory and
    #       chosen SDS destination
    drf_local_path = Path("./rf_data")
    sds_destination = PurePosixPath(f"drf_capture_{int(time.time())}")

    if not drf_local_path.is_dir():
        log.error(f"Directory not found: {drf_local_path}")
        log.info(
            "Please update drf_local_path to point to your Digital-RF directory and try again."
        )
        sys.exit(1)

    # ── channel discovery ─────────────────────────────────────────
    # Auto-discover channels from subdirs that contain drf_properties.h5.
    # To override, set drf_channel (single) or drf_channels (multi) manually.
    drf_channel: str | None = None
    drf_channels: list[str] = discover_drf_channels(drf_local_path)
    log.info(f"Discovered {len(drf_channels)} channel(s): {drf_channels}")

    # ── step 1: upload ────────────────────────────────────────────
    log.info(f"Uploading '{drf_local_path}' → SDS:/{sds_destination}")
    uploaded_files = upload_with_retries(
        client=client,
        local_path=drf_local_path,
        sds_path=sds_destination,
    )
    log.info(f"{len(uploaded_files)} file(s) now stored in SDS.")

    # ── step 2: create capture(s) ─────────────────────────────────
    if len(drf_channels) > 1:
        captures = create_multichannel_captures(
            client=client,
            sds_path=sds_destination,
            channels=drf_channels,
        )
        log.info(f"Created {len(captures)} of {len(drf_channels)} channel captures.")
    elif len(drf_channels) == 1 or drf_channel:
        channel = drf_channels[0] if drf_channels else drf_channel
        assert channel is not None
        capture = create_single_channel_capture(
            client=client,
            sds_path=sds_destination,
            channel=channel,
            name="My Digital-RF Capture",
        )
        if capture:
            log.info(f"Capture UUID: {capture.uuid}")
            log.info("Capture details:")
            log.info(f"  Channel: {capture.channel}")
            log.info(f"  Created at: {capture.created_at}")
            log.info(f"  Number of files: {len(capture.files)}")
```

--------------------------------

### Main Execution Block for SDS CRC ND EDU SDK

Source: https://sds.crc.nd.edu/sdk/common-workflows/capture-uploads

This snippet defines the standard Python entry point for the script. It ensures that the `main()` function is called only when the script is executed directly, not when imported as a module. This is a common Python best practice for organizing script execution.

```python
if __name__ == "__main__":
    main()
```

--------------------------------

### POST /captures/create

Source: https://sds.crc.nd.edu/sdk/api/captures

Creates a new RF capture record in the SpectrumX SDS system.

```APIDOC
## POST /captures/create

### Description
Creates a new RF capture in the SDS environment.

### Method
POST

### Endpoint
/captures/create

### Request Body
- **capture_data** (object) - Required - The capture configuration details.

### Request Example
{
  "name": "my_rf_capture",
  "metadata": {}
}

### Response
#### Success Response (200)
- **id** (string) - The UUID of the created capture.

#### Response Example
{
  "id": "550e8400-e29b-41d4-a716-446655440000"
}
```

--------------------------------

### GET Dataset Files

Source: https://sds.crc.nd.edu/sdk/api/datasets

Retrieves a paginator for all files associated with a specific dataset using its UUID.

```APIDOC
## GET /datasets/{dataset_uuid}/files

### Description
Retrieves a paginator object containing all files associated with the specified dataset UUID. This method handles pagination automatically via the SDK's Paginator class.

### Method
GET

### Endpoint
/datasets/{dataset_uuid}/files

### Parameters
#### Path Parameters
- **dataset_uuid** (UUID) - Required - The unique identifier of the dataset to retrieve files for.

### Request Example
```python
# Using the DatasetAPI instance
dataset_api.get_files(dataset_uuid="550e8400-e29b-41d4-a716-446655440000")
```

### Response
#### Success Response (200)
- **Paginator[File]** (Object) - A paginator object that iterates over File instances.

#### Response Example
```json
{
  "files": [
    { "id": "file_1", "name": "data.csv" },
    { "id": "file_2", "name": "metadata.json" }
  ],
  "next_page": "/datasets/uuid/files?page=2"
}
```
```

--------------------------------

### GET /get_file

Source: https://sds.crc.nd.edu/sdk/api/client

Retrieves the metadata for a specific file instance by its UUID without downloading the actual file contents.

```APIDOC
## GET /get_file

### Description
Get a file instance by its ID. Only metadata is downloaded from SDS. This does not download the file contents; use `download_file` for that purpose.

### Method
GET

### Endpoint
/get_file/{file_uuid}

### Parameters
#### Path Parameters
- **file_uuid** (UUID4 | str) - Required - The UUID of the file to retrieve.

### Request Example
GET /get_file/550e8400-e29b-41d4-a716-446655440000

### Response
#### Success Response (200)
- **file** (File) - The file instance object containing metadata.

#### Response Example
{
  "file_uuid": "550e8400-e29b-41d4-a716-446655440000",
  "filename": "example.txt",
  "size": 1024
}
```

--------------------------------

### Download SDS Dataset with Retry Logic in Python

Source: https://sds.crc.nd.edu/sdk/common-workflows/dataset-downloads

This script initializes an SDS client, authenticates, and performs a bulk download of a dataset using its UUID. It includes a loop with exponential backoff to manage retries upon encountering SDSError or other exceptions.

```python
import time
from pathlib import Path
from uuid import UUID
from loguru import logger as log
from spectrumx import Client
from spectrumx.errors import SDSError

def main() -> None:
    client = Client(host="sds.crc.nd.edu")
    client.dry_run = False
    client.authenticate()

    dataset_uuid = UUID("xxxxxxxx-xxxx-4xxx-xxxx-xxxxxxxxxxxx")
    download_path = Path("./downloaded_dataset")

    is_success = False
    retries_left = 50
    sleep_time_sec = 1
    max_sleep_time_sec = 300
    is_first_run = True

    while not is_success and retries_left > 0:
        if not is_first_run:
            sleep_time_sec *= 2
            time.sleep(min(sleep_time_sec, max_sleep_time_sec))
        is_first_run = False
        retries_left -= 1
        try:
            results = client.download_dataset(
                dataset_uuid=dataset_uuid,
                to_local_path=download_path,
                overwrite=False,
                verbose=True
            )
        except (SDSError, Exception) as err:
            log.exception(f"Error: {err}")
            continue

        sleep_time_sec = 1
        if not [r for r in results if not r]:
            is_success = True
            break
```

--------------------------------

### Initialize CaptureAPI class

Source: https://sds.crc.nd.edu/sdk/api/captures

The CaptureAPI class is initialized with a GatewayClient instance. It supports optional dry_run and verbose flags to control execution behavior and logging output.

```python
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
```

--------------------------------

### Initialize SDS Client | spectrumx.client.Client

Source: https://sds.crc.nd.edu/sdk/api/client

Instantiates an SDS client. This client is used for various operations such as file management, uploads, and capture creation. It requires host information and can be configured with environment files or dictionaries, with options for verbose output.

```python
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()

```

--------------------------------

### Get Base URL Property | spectrumx.client.Client

Source: https://sds.crc.nd.edu/sdk/api/client

Retrieves the base URL for the SDS client. This property provides the primary network address used for communication with the SDS service.

```python
base_url: str

```

--------------------------------

### Compare file contents using BLAKE3 hashing

Source: https://sds.crc.nd.edu/sdk/api/models

Checks if two files have identical contents by comparing their BLAKE3 hash sums. Optionally logs the hash and path information for debugging.

```python
def is_same_contents(self, other: "File", *, verbose: bool = False) -> bool:
    """Checks if the file has the same contents as another local file."""
    this_sum = self.compute_sum_blake3()
    other_sum = other.compute_sum_blake3()
    if verbose:
        utils.log_user(f"{this_sum}  {self.local_path}")
        utils.log_user(f"{other_sum}  {other.local_path}")
    return this_sum == other_sum
```

--------------------------------

### SpectrumX Python SDK - Invalid Result Constructors

Source: https://sds.crc.nd.edu/sdk/api/errors

Demonstrates invalid ways to instantiate the `Result` class in Python. These examples highlight constructor constraints, such as providing both value and exception, or providing neither.

```python
from typing import Any, Dict, Generic, TypeVar, Union

_unset = object()
T = TypeVar("T")

class KnownError(Exception):
    pass

class Result(Generic[T]):
    def __init__(
        self,
        *,
        value: T | object = _unset,
        exception: Exception | object = _unset,
        error_info: Dict[str, Any] | None = None,
    ):
        if value is not _unset and exception is not _unset:
            raise ValueError("Cannot provide both value and exception.")
        if value is _unset and exception is _unset:
            raise ValueError("Must provide either value or exception.")
        if isinstance(value, Exception):
            raise ValueError("Value cannot be an Exception.")
        if not isinstance(exception, Exception) and exception is not _unset:
            raise ValueError("Exception must be an instance of Exception.")
        if error_info is not None and exception is _unset:
            raise ValueError("error_info requires an exception.")

# Invalid examples (these will raise ValueError):
# Result()                                    # empty result
# Result(value=123, exception=RuntimeError()) # both value and exception
# Result(value=123, error_info={})            # error_info without exception
# Result(exception=123)                       # exception is not an Exception
# Result(value=RuntimeError())                # value is an Exception

# To demonstrate, we'll wrap these in try-except blocks:

try:
    Result()
except ValueError as e:
    print(f"Caught expected error: {e}")

try:
    Result(value=123, exception=RuntimeError())
except ValueError as e:
    print(f"Caught expected error: {e}")

try:
    Result(value=123, error_info={{}})
except ValueError as e:
    print(f"Caught expected error: {e}")

try:
    Result(exception=123)
except ValueError as e:
    print(f"Caught expected error: {e}")

try:
    Result(value=RuntimeError())
except ValueError as e:
    print(f"Caught expected error: {e}")

```

--------------------------------

### File Content Comparison

Source: https://sds.crc.nd.edu/sdk/api/models

Compares the content of a file with another local file using Blake3 hashing. Optionally logs the hash values and file paths.

```APIDOC
## POST /files/compare_contents

### Description
Checks if the file has the same contents as another local file using Blake3 hashing. Optionally logs the hash values and file paths for debugging.

### Method
POST

### Endpoint
`/files/compare_contents`

### Parameters
#### Query Parameters
- **other** (File) - Required - The other local file to compare contents with.
- **verbose** (bool) - Optional - If true, logs the computed hash values and file paths.

### Request Body
This endpoint does not require a request body, as the comparison is based on the file object itself and the provided `other` file.

### Response
#### Success Response (200)
- **result** (bool) - True if the file contents are the same, False otherwise.

#### Response Example
```json
{
  "result": true
}
```
```

--------------------------------

### Get Base URL Without Port Property | spectrumx.client.Client

Source: https://sds.crc.nd.edu/sdk/api/client

Retrieves the base URL for the SDS client, excluding the port number. This is useful for constructing URLs or when the port is implicitly handled.

```python
base_url_no_port: str

```

--------------------------------

### POST /download_dataset

Source: https://sds.crc.nd.edu/sdk/api/client

Downloads all files in a dataset using the existing download infrastructure by fetching a paginated list of files and processing them.

```APIDOC
## POST /download_dataset

### Description
Downloads all files in a dataset using the existing download infrastructure. This method fetches a paginated list of file objects and saves them to the specified local directory.

### Method
POST

### Endpoint
download_dataset

### Parameters
#### Request Body
- **dataset_uuid** (UUID4 | str) - Required - The UUID of the dataset to download.
- **to_local_path** (Path | str) - Required - The local path to save the downloaded files to.
- **skip_contents** (bool) - Optional - When True, only the metadata is downloaded. Default: False.
- **overwrite** (bool) - Optional - Whether to overwrite existing local files. Default: False.
- **verbose** (bool) - Optional - Show progress bars and detailed output. Default: True.

### Request Example
{
  "dataset_uuid": "550e8400-e29b-41d4-a716-446655440000",
  "to_local_path": "/data/my_dataset",
  "skip_contents": false,
  "overwrite": false,
  "verbose": true
}

### Response
#### Success Response (200)
- **results** (list[Result[File]]) - A list of results for each file processed during the download operation.

#### Response Example
{
  "results": [
    {"status": "success", "file": "data_01.csv"},
    {"status": "success", "file": "data_02.csv"}
  ]
}
```

--------------------------------

### File Method: compute_sum_blake3

Source: https://sds.crc.nd.edu/sdk/api/models

Calculates the BLAKE3 checksum of the file. This method can block if the file is currently being downloaded. It returns the checksum or None if the file is not available locally. The checksum is not cached as the file might change.

```python
def compute_sum_blake3(self) -> str | None:
    """Calculates the BLAKE3 checksum of the file.

    If the file is currently being downloaded, this method will
        block until the download is complete or terminated.

    Args:
        block: when True, waits until the file contents are unlocked.
    Returns:
        The BLAKE3 checksum of the file,
            OR None if the file is not available locally.
    """
    # we deliberately don't cache the checksum because the file
    # might be changed externally by a different process or the user.
    with self.contents_lock:
        # content downloads cannot start within this block
        if not self.is_local:
            return None
        # this should not happen anyway, so let's just assert
        assert self.local_path is not None, "Local path is not set"
        return utils.sum_blake3(self.local_path)
```

--------------------------------

### Get File Metadata from SDS using Python

Source: https://sds.crc.nd.edu/sdk/api/client

Retrieves a file instance by its UUID from SDS, downloading only its metadata. This function does not download the file's content. To obtain a local copy of the file, `Client.download_file()` must be called subsequently.

```python
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)
```

--------------------------------

### Permission Conversion Utilities

Source: https://sds.crc.nd.edu/sdk/api/models

Provides utilities for converting between octal and string representations of Unix file permissions.

```APIDOC
## Permission Conversion Utilities

### Description
This section details functions for converting Unix file permissions between octal integer and string formats.

### Functions

#### `octal_to_unix_perm_string`

##### Description
Converts an octal Unix file permission integer to its string representation (e.g., 'rwxr-xr-x').

##### Method
POST

##### Endpoint
`/permissions/octal_to_string`

##### Parameters
*   **value** (int) - Required - An integer representing the file permission in octal format (e.g., 0o755).

##### Request Example
```json
{
  "value": 448 
}
```

##### Response
*   **permission_string** (str) - A 9-character string representing Unix permissions (e.g., 'rwxr-xr-x').

##### Response Example
```json
{
  "permission_string": "rwx------"
}
```

*   **Raises**: `ValueError` - If the input is not a valid octal Unix permission.

#### `unix_perm_string_to_octal`

##### Description
Converts a Unix permission string (e.g., 'rwxr-xr-x') to its octal integer representation.

##### Method
POST

##### Endpoint
`/permissions/string_to_octal`

##### Parameters
*   **value** (str) - Required - A 9-character Unix permission string (e.g., 'rw-r--r--').

##### Request Example
```json
{
  "value": "rwxr-xr-x"
}
```

##### Response
*   **permission_octal** (int) - An integer representing the octal value of the permission (e.g., 0o755).

##### Response Example
```json
{
  "permission_octal": 493
}
```

*   **Raises**: `ValueError` - If the input string is not a valid Unix permission string.
```

--------------------------------

### Get Files from Dataset using Python

Source: https://sds.crc.nd.edu/sdk/api/datasets

Retrieves files associated with a specific dataset using the get_files method of the DatasetAPI. This method takes a dataset UUID as input and returns a Paginator object that yields File objects. It supports dry run mode for simulated operations.

```python
def get_files(
    self,
    dataset_uuid: uuid.UUID,
) -> Paginator[File]:
    """Get files in the dataset as a paginator.

    Args:
        dataset_uuid: The UUID of the dataset to get files for.
    Returns:
        A paginator for the files in the dataset.
    """
    if self.dry_run:
        log_user("Dry run enabled: files will be simulated")

    # Create a paginator that fetches from the dataset files endpoint
    pagination: Paginator[File] = Paginator(
        Entry=File,
        gateway=self.gateway,
        list_method=self.gateway.get_dataset_files,
        list_kwargs={"dataset_uuid": dataset_uuid},
        dry_run=self.dry_run,
        verbose=self.verbose,
    )

    return pagination

```

--------------------------------

### Log Information and Warnings in SDS CRC ND EDU SDK

Source: https://sds.crc.nd.edu/sdk/common-workflows/capture-uploads

This snippet demonstrates how to log informational messages, including file paths and URLs, and warning messages when specific conditions are not met. It utilizes Python's logging module to provide feedback during the SDK's operation. No external dependencies are required beyond the standard Python logging library.

```python
log.info(f"  Path in SDS: {capture.top_level_dir}")
if not client.dry_run:
    log.info(
        f"  Capture page: https://sds.crc.nd.edu/api/v1/assets/captures/{capture.uuid}/"
    )
else:
    log.warning(
        "No channels discovered or specified — files were uploaded but no "
        "capture was created. Verify your Digital-RF directory contains "
        f"subdirectories with '{DRF_PROPERTIES_FILENAME}' files, or set "
        "drf_channel / drf_channels manually."
    )
```

--------------------------------

### Get File Metadata from SDS (Python)

Source: https://sds.crc.nd.edu/sdk/api/sds_files

Retrieves a file's metadata from SDS using its unique identifier (UUID). This function does not download the file's content, only its information. A Client object and the file's UUID are required inputs. If dry run mode is enabled, a sample file object is returned.

```python
def get_file(client: Client, 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 `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.
    """

    uuid_to_set: UUID4 = (
        uuid.UUID(file_uuid) if isinstance(file_uuid, str) else file_uuid
    )

    if not client.dry_run:
        file_bytes = client._gateway.get_file_by_id(uuid=uuid_to_set.hex)
        return File.model_validate_json(file_bytes)

    log_user("Dry run enabled: a sample file is being returned instead")
    return files.generate_sample_file(uuid_to_set)

```

--------------------------------

### Compare File Contents using Blake3 Checksums (Python)

Source: https://sds.crc.nd.edu/sdk/api/models

Compares the content of the current file object with another file object by computing and comparing their Blake3 checksums. If verbose mode is enabled, it logs the checksums and file paths of both files. This function relies on the `compute_sum_blake3` method and a `utils.log_user` function for output.

```python
def compare_contents(self, other, verbose=False):
    """Checks if the file has the same contents as another local file."""
    this_sum = self.compute_sum_blake3()
    other_sum = other.compute_sum_blake3()
    if verbose:
        utils.log_user(f"{this_sum}  {self.local_path}")
        utils.log_user(f"{other_sum}  {other.local_path}")
    return this_sum == other_sum
```

--------------------------------

### Configure Authentication with .env File

Source: https://sds.crc.nd.edu/sdk/faq

Sets up token-based authentication by reading the SDS secret token from a .env file. This method is recommended for security to prevent tokens from being committed to version control. The SDK automatically loads the .env file upon initialization.

```env
SDS_SECRET_TOKEN=your-secret-token-no-quotes

```

--------------------------------

### POST upload_file

Source: https://sds.crc.nd.edu/sdk/api/client

Uploads a local file to the SDS virtual file system, optionally prepending the target directory path.

```APIDOC
## POST upload_file

### Description
Uploads a local file to the SDS environment. If the file object contains directory information, it is prepended to the target SDS path.

### Method
POST

### Endpoint
upload_file

### Parameters
#### Request Body
- **local_file** (File | Path | str) - Required - The local file to upload.
- **sds_path** (PurePosixPath | Path | str) - Optional - The virtual directory on SDS to upload the file to. Defaults to '/'.

### Request Example
{
  "local_file": "/path/to/local/file.txt",
  "sds_path": "/data/uploads"
}

### Response
#### Success Response (200)
- **File** (Object) - The file instance with updated attributes reflecting the new SDS path.

#### Response Example
{
  "name": "file.txt",
  "directory": "/data/uploads/",
  "size": 1024
}
```

--------------------------------

### POST /download_file

Source: https://sds.crc.nd.edu/sdk/api/client

Downloads a file from SDS, including metadata and optionally the file contents. This method will overwrite any existing local file at the destination path.

```APIDOC
## POST /download_file

### Description
Downloads a file from SDS. Either `file_instance` or `file_uuid` must be provided. Note that this function always overwrites the local file if it already exists.

### Method
POST

### Endpoint
/download_file

### Parameters
#### Request Body
- **file_instance** (File | None) - Optional - The file instance to download.
- **file_uuid** (UUID4 | str | None) - Optional - The UUID of the file to download.
- **to_local_path** (Path | str | None) - Optional - The local path to save the downloaded file to.
- **skip_contents** (bool) - Optional - When True, only metadata is downloaded. Default: False.
- **warn_missing_path** (bool) - Optional - Show a warning when the download location is undefined. Default: True.

### Request Example
{
  "file_uuid": "550e8400-e29b-41d4-a716-446655440000",
  "to_local_path": "/data/downloads/file.txt",
  "skip_contents": false
}

### Response
#### Success Response (200)
- **file** (File) - The file instance with updated attributes.

#### Response Example
{
  "file_uuid": "550e8400-e29b-41d4-a716-446655440000",
  "status": "downloaded",
  "local_path": "/data/downloads/file.txt"
}
```

--------------------------------

### Initialize DatasetAPI in Python

Source: https://sds.crc.nd.edu/sdk/api/datasets

Initializes the DatasetAPI class, which manages dataset-related operations. It requires a GatewayClient instance and accepts optional dry_run and verbose flags for controlling execution behavior. The constructor sets up internal states for these parameters.

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

```
Tokens:
Raw