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