`rfgen.storage`¶

Reference surface for the shipped Layer 7 storage backends. All shipped stores persist and read Record objects through the same factory-plus-handle contract.

Module summary¶

from rfgen.enums import StoreMode
from rfgen.storage import ZarrStore

store = ZarrStore(uri="file://./out.zarr")
with store.open("file://./out.zarr", StoreMode.WRITE) as handle:
    sample_id = handle.write(record)
    round_trip = handle.read(sample_id)

Class index¶

Class	Kind	Notes
BaseStore	abc	Configured factory for a backend.
StoreHandle	abc	Open read/write session returned by `BaseStore.open(...)`.
ZarrStore	concrete	Canonical random-access backend.
WebDatasetStore	concrete	Tar-shard backend with deterministic shard-then-member iteration.
HDF5Store	concrete	Local TorchSig interop backend.
SigMFStore	concrete	Local SigMF backend with rfgen sidecars.

`canonical_sample_id`¶

from rfgen.storage import canonical_sample_id

def canonical_sample_id(record: Record) -> str: ...

Returns the lowercase hexadecimal SHA-256 digest of rfgen’s canonical CBOR representation of the complete Record. Storage plugins call this function when write(record, sample_id=None) needs the same default ID as the shipped backends. Invalid record values or values that cannot be canonicalized raise StorageError.

`class rfgen.storage.BaseStore`¶

class BaseStore(ABC):
    name: ClassVar[str]

    @abstractmethod
    def open(self, uri: str, mode: StoreMode | str) -> StoreHandle: ...

    @abstractmethod
    def has_shard(self, shard_id: str) -> bool: ...

    def dataset_uri(self) -> str | None: ...

    @classmethod
    def from_config(cls, config: StorageConfig) -> BaseStore: ...

Notes¶

name is the registry key under rfgen.stores.
from_config(...) instantiates shipped backends directly and resolves custom backend names through the entry-point registry.
For custom backends, from_config(...) first honors a backend override of from_config(config). Otherwise it forwards any constructor parameters that match the StorageConfig surface, including path or uri, compression, chunk_samples, record_axis, assets_path, and the full config object when accepted.
Built-in backends are identified by StorageBackend. Custom plugin names stay as plain strings on the config surface.
has_shard(...) probes the dataset already bound to the store instance. Call it on a store created with a concrete uri=..., or on a custom backend whose dataset_uri implementation returns a real dataset URI.

open(..., StoreMode.APPEND) requires a real existing dataset URI. When a store is pre-bound, the normal call shape is:

dataset_uri = store.dataset_uri()
assert dataset_uri is not None
with store.open(dataset_uri, StoreMode.APPEND) as handle:
    ...

Method: `open`¶

def open(self, uri: str, mode: StoreMode | str) -> StoreHandle: ...

Returns a backend-specific StoreHandle for the requested URI and StoreMode.

Method: `dataset_uri`¶

def dataset_uri(self) -> str | None: ...

Returns the trimmed dataset URI carried by a pre-bound store instance. Returns None when the store was constructed without a default dataset binding. Returns the trimmed stored URI only, it does not canonicalize schemes, path spellings, or equivalent filesystem locations. Orchestration compares ShardSpec.output_uri against this value by exact string equality when a bound store exposes dataset_uri().

Method: `has_shard`¶

def has_shard(self, shard_id: str) -> bool: ...

Tests whether the dataset bound to this store contains a durable completion marker for shard_id. shard_id is the exact shard identifier to probe; the method does not open a handle or infer a dataset URI. It returns True only when the backend’s complete-shard representation is present, and returns False for an absent, partial, or still-writing shard. A store without a bound dataset URI raises StorageError.

Class method: `from_config`¶

@classmethod
def from_config(cls, config: StorageConfig) -> BaseStore: ...

Builds a configured store factory from config. When called on BaseStore, the method maps a built-in StorageBackend to its shipped store class or loads a plain-string backend name from the rfgen.stores entry-point group. When called on a concrete subclass, it constructs that subclass regardless of config.backend. The return value is an unopened BaseStore instance bound to config.path; callers still use open to create a session.

Custom classes may override from_config(config). Otherwise the factory forwards only supported StorageConfig-derived constructor arguments. It raises StorageError when a custom constructor requires unsupported parameters or cannot be called with the forwarded arguments. Entry-point discovery and loading errors propagate from the plugin registry.

`class rfgen.storage.StoreHandle`¶

Open session API shared by all stores.

class StoreHandle(ABC):
    def write(self, record: Record, *, sample_id: str | None = None) -> str: ...
    def write_shard(self, shard_id: str, records: Iterable[Record]) -> tuple[str, ...]: ...
    def read(self, sample_id: str) -> Record: ...
    def append_annotation_row(
        self,
        sample_id: str,
        *,
        annotation_type: str,
        template_id: str,
        run_id: str,
        row: Mapping[str, object],
    ) -> Record: ...
    def iter_sample_ids(self) -> Iterator[str]: ...
    def sample_ids(self) -> tuple[str, ...]: ...
    def read_siblings(self, record: Record) -> Iterable[Record]: ...
    def write_asset_bundle(self, ref: GeometryAssetRef, files: Mapping[str, bytes]) -> GeometryAssetRef: ...
    def resolve_asset(self, ref: GeometryAssetRef, *, worker_cache_dir: Path | None = None) -> Path | str: ...
    def prepare_worker_assets(self, scene_assets: SceneAssetsConfig, cache_root: Path) -> SceneAssetsConfig: ...
    def __iter__(self) -> Iterator[Record]: ...
    def __len__(self) -> int: ...
    def close(self) -> None: ...

Common behavior¶

write(...) returns the stable sample_id. Default IDs are sha256(canonical_cbor(record)).
write_shard(…) consumes records lazily and may persist records as they are yielded. It publishes the backend-specific shard-complete marker only after the iterable is exhausted normally. If iteration or a record write raises, the exception propagates and no completion marker is published; already-persisted content-addressed records may remain for an idempotent retry.
Re-writing the same sample_id with identical canonical bytes is idempotent.
append_annotation_row(...) appends Phase 2 text metadata to an existing sample_id without rewriting canonical Phase 1 sample bytes, IQ, scene, or labels. Re-appending the same row is idempotent; a different row for the same (sample_id, annotation_type, template_id, run_id) raises StorageError.
read(...) raises StorageError with a KeyError cause when the ID is missing.
iter_sample_ids() yields the deterministic sample-id order without materializing records.
sample_ids() is the eager tuple convenience wrapper over iter_sample_ids().
read_siblings(...) filters by record.scene.scene_id.
write_asset_bundle(...), resolve_asset(...), and prepare_worker_assets(...) implement the content-addressed geometry-asset contract.

Method: `write`¶

def write(self, record: Record, *, sample_id: str | None = None) -> str: ...

Persists one Record and returns its stable sample_id.

Method: `write_shard`¶

def write_shard(
    self,
    shard_id: str,
    records: Iterable[Record],
) -> tuple[str, ...]: ...

Consumes records lazily and returns their ordered stable sample IDs. The backend publishes its shard-complete marker only after the iterable is exhausted normally. If iteration or a record write raises, the exception propagates and the marker remains absent; records persisted before the failure may remain for an idempotent retry.

Method: `append_annotation_row`¶

def append_annotation_row(
    self,
    sample_id: str,
    *,
    annotation_type: str,
    template_id: str,
    run_id: str,
    row: Mapping[str, object],
) -> Record: ...

Persists one annotation row against an existing sample and returns the updated Record. The row is keyed by (annotation_type, template_id, run_id) under the sample’s text field. Backends store the row in an append-only annotation overlay so samples/<sample_id>/canonical_cbor and the IQ/label payloads remain unchanged.

Method: `read`¶

def read(self, sample_id: str) -> Record: ...

Loads one persisted Record by stable sample_id.

Method: `iter_sample_ids`¶

def iter_sample_ids(self) -> Iterator[str]: ...

Yields the backend’s deterministic sample-id order without reading full Record payloads.

Method: `sample_ids`¶

def sample_ids(self) -> tuple[str, ...]: ...

Returns tuple(handle.iter_sample_ids()) as an eager convenience helper.

Method: `read_siblings`¶

def read_siblings(self, record: Record) -> Iterable[Record]: ...

Accepts one Record as the scene anchor and returns an iterable of all stored records whose scene.scene_id equals record.scene.scene_id, including the anchor when it is stored. Records follow the backend’s deterministic sample order. Storage and decoding failures raised while locating or reading a sibling propagate as StorageError.

Method: `write_asset_bundle`¶

def write_asset_bundle(
    self,
    ref: GeometryAssetRef,
    files: Mapping[str, bytes],
) -> GeometryAssetRef: ...

Persists files, keyed by bundle-relative path, as the content-addressed bundle described by ref. The returned GeometryAssetRef contains the computed SHA-256 content hash, stored bundle URI, and normalized entrypoint. File-mapping order does not affect the hash. A Sionna built-in-scene ref is returned unchanged because it has no file bundle.

The method raises StorageError in a read-only handle, for an empty file mapping, unsafe or duplicate normalized paths, a missing or malformed XML entrypoint, or XML-relative references not present in the bundle. Filesystem write failures propagate from the configured asset filesystem.

Method: `resolve_asset`¶

def resolve_asset(
    self,
    ref: GeometryAssetRef,
    *,
    worker_cache_dir: Path | None = None,
) -> Path | str: ...

Resolves ref to its usable entrypoint. It returns the original URI string for a Sionna built-in scene and a local Path for a file-backed bundle. A local asset store may return its entrypoint in place. A non-local asset store copies and verifies the bundle below worker_cache_dir, reusing a cache entry only when its recorded hashes still match.

The method raises StorageError when a non-local bundle has no worker_cache_dir, the ref lacks required content-addressed metadata, the manifest or entrypoint is missing or invalid, or a stored or cached file fails its size or SHA-256 check. Filesystem read and copy failures propagate from the asset filesystem.

Method: `prepare_worker_assets`¶

def prepare_worker_assets(
    self,
    scene_assets: SceneAssetsConfig,
    cache_root: Path,
) -> SceneAssetsConfig: ...

Resolves the file-backed scene-asset refs in scene_assets into worker-local entrypoints under cache_root. Built-in Sionna scenes remain URI refs.

Method: `iter`¶

def __iter__(self) -> Iterator[Record]: ...

Returns an iterator over every stored Record in the backend’s deterministic sample order. The iterator is lazy; storage and decoding failures propagate as StorageError when the affected record is reached.

Method: `len`¶

def __len__(self) -> int: ...

Returns the non-negative number of stored records visible to this open handle. It does not materialize record payloads. Backend metadata or filesystem errors encountered while obtaining the count propagate.

Method: `close`¶

def close(self) -> None: ...

Flushes pending backend metadata and writes, then releases resources owned by the handle. It returns None. Exiting with store.open(...) as handle calls close() even when the body raises. Shipped handles permit repeated calls; flush or close failures from the backend propagate.

`class rfgen.storage.ZarrStore`¶

Canonical backend.

ZarrStore(
    *,
    uri: str | None = None,
    schema_version: int = 1,
    compression: str = "blosc",
    chunk_samples: int = 128,
    record_axis: RecordAxis = RecordAxis.PER_RX,
    assets_path: str | None = None,
)

Notes¶

Accepts local and object-store URIs through fsspec.
Persists IQ as complex64.
Persists segmentation masks in canonical storage dtype: int16 for single-label masks and uint8 for multi-label masks.
Iteration order is lexicographic sample_id.
has_shard(...) checks for the shard group plus _shard_complete.

`class rfgen.storage.WebDatasetStore`¶

Sequential tar-shard backend.

WebDatasetStore(
    *,
    uri: str | None = None,
    schema_version: int = 1,
    compression: str = "none",
    record_axis: RecordAxis = RecordAxis.PER_RX,
    assets_path: str | None = None,
)

Notes¶

Uses webdataset.TarWriter and webdataset.tarfile_samples when the optional package is installed.
Preserves the original URI scheme when calling tarfile_samples(...), so remote shard URIs continue to work.
Iteration order is shard-path lexicographic, then tar-member order within each shard.
has_shard(...) checks for the .tar plus .tar.SHA256.

`class rfgen.storage.HDF5Store`¶

Local HDF5 backend for TorchSig interop.

HDF5Store(
    *,
    uri: str | None = None,
    schema_version: int = 1,
    compression: str = "gzip",
    chunk_samples: int = 128,
    record_axis: RecordAxis = RecordAxis.PER_RX,
    assets_path: str | None = None,
)

Notes¶

Install rfgen[torchsig,hdf5] for HDF5/TorchSig interop. torchsig provides the wideband HDF5 helpers and hdf5 provides h5py.
Missing optional dependencies fail lazily on open(...) with BackendUnavailableError.
has_shard(...) checks for the HDF5 shard group with complete=True.

`class rfgen.storage.SigMFStore`¶

Local SigMF backend.

SigMFStore(
    *,
    uri: str | None = None,
    schema_version: int = 1,
    compression: str = "none",
    record_axis: RecordAxis = RecordAxis.PER_RX,
    assets_path: str | None = None,
)

Notes¶

Local-only in Layer 7. The runtime and config surface require file://....
Stores IQ through the sigmf package and writes rfgen metadata sidecars beside the SigMF pair.
Uses the same shared receiver-frame resolver as the label layer for capture center frequency and sample-rate fields.
has_shard(...) checks for the .sigmf-meta plus .sigmf-data pair.

Record-axis rules¶

RecordAxis.PER_RX requires record.scene.rx_index is not None and IQ shape (2, N).
RecordAxis.JOINT requires record.scene.rx_index is None and IQ shape (num_rx, 2, N).

rfgen.storage¶

Module summary¶

Class index¶

canonical_sample_id¶

class rfgen.storage.BaseStore¶

Notes¶

Method: open¶

Method: dataset_uri¶

Method: has_shard¶

Class method: from_config¶

class rfgen.storage.StoreHandle¶

Common behavior¶

Method: write¶

Method: write_shard¶

Method: append_annotation_row¶

Method: read¶

Method: iter_sample_ids¶

Method: sample_ids¶

Method: read_siblings¶

Method: write_asset_bundle¶

Method: resolve_asset¶

Method: prepare_worker_assets¶

Method: __iter__¶

Method: __len__¶

Method: close¶

class rfgen.storage.ZarrStore¶

Notes¶

class rfgen.storage.WebDatasetStore¶

Notes¶

class rfgen.storage.HDF5Store¶

Notes¶

class rfgen.storage.SigMFStore¶

Notes¶

Record-axis rules¶

See Also¶

`rfgen.storage`¶

`canonical_sample_id`¶

`class rfgen.storage.BaseStore`¶

Method: `open`¶

Method: `dataset_uri`¶

Method: `has_shard`¶

Class method: `from_config`¶

`class rfgen.storage.StoreHandle`¶

Method: `write`¶

Method: `write_shard`¶

Method: `append_annotation_row`¶

Method: `read`¶

Method: `iter_sample_ids`¶

Method: `sample_ids`¶

Method: `read_siblings`¶

Method: `write_asset_bundle`¶

Method: `resolve_asset`¶

Method: `prepare_worker_assets`¶

Method: `iter`¶

Method: `len`¶

Method: `close`¶

`class rfgen.storage.ZarrStore`¶

`class rfgen.storage.WebDatasetStore`¶

`class rfgen.storage.HDF5Store`¶

`class rfgen.storage.SigMFStore`¶