Python API reference

Use import signal_dataset as sds for ordinary workflows. Advanced extension APIs live in the record, storage, dataset, and annotation namespaces. Classes returned by factories have internal constructors, so only the observable interfaces below are supported.

Every signature on this page is generated from the source, so it cannot drift from what the library actually accepts. Prose is written by hand.

All metadata is deeply frozen JSON-compatible data, and Field.data is a non-writeable NumPy array.

Logical values

class signal_dataset.Coordinate(values: 'tuple[Json, ...] | None' = None, start: 'int | float | None' = None, step: 'int | float | None' = None, reference: 'str | None' = None, unit: 'str | None' = None, metadata: 'Mapping[str, Json]' = <factory>)[source]

Bases: object

Parameters:
values: tuple[None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]], ...] | None
start: int | float | None
step: int | float | None
reference: str | None
unit: str | None
metadata: Mapping[str, None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]]]
class signal_dataset.Axis(name: 'str', length: 'int', role: 'str | None' = None, id: 'str | None' = None, coordinate: 'Coordinate | None' = None, metadata: 'Mapping[str, Json]' = <factory>)[source]

Bases: object

Parameters:
name: str
length: int
role: str | None
id: str | None
coordinate: Coordinate | None
metadata: Mapping[str, None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]]]
class signal_dataset.Field(data: 'np.ndarray', axes: 'tuple[Axis, ...]' = (), metadata: 'Mapping[str, Json]' = <factory>)[source]

Bases: object

Parameters:
data: ndarray
axes: tuple[Axis, ...]
metadata: Mapping[str, None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]]]
class signal_dataset.Record(id: 'str', fields: 'Mapping[str, Field]', scene_id: 'str | None' = None, metadata: 'Mapping[str, Json]' = <factory>)[source]

Bases: Mapping[str, Field]

Parameters:
id: str
fields: Mapping[str, Field]
scene_id: str | None
metadata: Mapping[str, None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]]]
class signal_dataset.FieldMetadata(dtype: 'str', shape: 'tuple[int, ...]', axes: 'tuple[Axis, ...]' = (), metadata: 'Mapping[str, Json]' = <factory>)[source]

Bases: object

Parameters:
dtype: str
shape: tuple[int, ...]
axes: tuple[Axis, ...]
metadata: Mapping[str, None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]]]
class signal_dataset.RecordMetadata(id: 'str', fields: 'Mapping[str, FieldMetadata]', scene_id: 'str | None' = None, metadata: 'Mapping[str, Json]' = <factory>)[source]

Bases: Mapping[str, FieldMetadata]

Parameters:
id: str
fields: Mapping[str, FieldMetadata]
scene_id: str | None
metadata: Mapping[str, None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]]]
classmethod from_record(record)[source]
Parameters:

record (Record)

Return type:

RecordMetadata

class signal_dataset.PublishedShard(data_uri: 'str', metadata_uri: 'str', work_id: 'str', attempt: 'int', record_count: 'int', data_bytes: 'int', metadata_bytes: 'int', data_generation: 'int | None' = None, metadata_generation: 'int | None' = None)[source]

Bases: object

Parameters:

  • data_uri (str)

  • metadata_uri (str)

  • work_id (str)

  • attempt (int)

  • record_count (int)

  • data_bytes (int)

  • metadata_bytes (int)

  • data_generation (int | None)

  • metadata_generation (int | None)

data_uri: str
metadata_uri: str
work_id: str
attempt: int
record_count: int
data_bytes: int
metadata_bytes: int
data_generation: int | None
metadata_generation: int | None
class signal_dataset.PublishedAnnotationShard(name: 'str', publication_id: 'str', source_snapshot_id: 'str', logical_shard_id: 'str', first_ordinal: 'int', record_ids: 'tuple[str, ...]', uri: 'str', stored_bytes: 'int', generation: 'int | None', statuses: 'tuple[str, ...]', terminal_statuses: 'tuple[str, ...]')[source]

Bases: object

Parameters:

  • name (str)

  • publication_id (str)

  • source_snapshot_id (str)

  • logical_shard_id (str)

  • first_ordinal (int)

  • record_ids (tuple[str, ...])

  • uri (str)

  • stored_bytes (int)

  • generation (int | None)

  • statuses (tuple[str, ...])

  • terminal_statuses (tuple[str, ...])

name: str
publication_id: str
source_snapshot_id: str
logical_shard_id: str
first_ordinal: int
record_ids: tuple[str, ...]
uri: str
stored_bytes: int
generation: int | None
statuses: tuple[str, ...]
terminal_statuses: tuple[str, ...]

Dataset operations

signal_dataset.open(root, *, object_store=None, options=None, shard_store=None)[source]
Parameters:
Return type:

Dataset

signal_dataset.write_shard(records, root, *, work_id='0', attempt=0, object_store=None, shard_store=None, options=None, retry=None)[source]
Parameters:
Return type:

PublishedShard

signal_dataset.publish(root, shards, *, dataset_id, snapshot_id, expected_work_ids=None, metadata=None, object_store=None, shard_store=None, storage_options=None, publication_options=None, retry=None)[source]

Publish shards as an immutable snapshot, writing root.json last.

Safe to re-run. Every write is create-only and resumes on an identical object, so an attempt interrupted anywhere completes when repeated – which is also what makes retry safe: a transient failure re-enters the same body and finds its own earlier writes rather than colliding with them.

Parameters:
Return type:

Dataset

signal_dataset.validate(record, *, policy=None)[source]

Validate structural, reference, JSON, dtype, and resource invariants.

Parameters:
Return type:

None

class signal_dataset.Dataset(root, dataset_root, snapshot, index)[source]

Bases: Sequence[Record]

Parameters:

  • root (str)

  • dataset_root (DatasetRoot)

  • snapshot (Snapshot)

  • index (_Index)

read_instructions(start=0, stop=None, *, metadata=False)[source]

Where a range of records lives, as shard URIs and offsets.

Performs no I/O beyond manifest pages already loaded, so it is safe to call in a DataLoader.__init__ or inside a worker.

This is the seam for a caller who wants their own fetching. The library owns addressing and decoding because those are the format; ordering, concurrency, prefetch and caching belong to whatever is feeding the trainer. Pair it with signal_dataset.decode():

for instruction in dataset.read_instructions(0, 1024):
    payloads = my_reader(instruction.filename,
                         instruction.skip, instruction.take)
    records = [sds.decode(payload) for payload in payloads]

metadata=True addresses the aligned metadata shards instead, which carry every record’s identity and shape without its samples.

Parameters:
Return type:

list[ReadInstruction]

property object_store: ObjectStore
property shard_store: ShardStore
property storage_options: StorageOptions
iter_record_metadata()[source]

Yield all record metadata with bounded, shard-batched I/O.

Return type:

Iterator[RecordMetadata]

iter_records()[source]

Yield all full records with bounded, shard-batched I/O.

Return type:

Iterator[Record]

select(indices)[source]
Parameters:

indices (Sequence[int])

Return type:

DatasetView

class signal_dataset.DatasetView(dataset, indices)[source]

Bases: Sequence[Record]

Parameters:
publish(destination, *, dataset_id, snapshot_id, publication_options=None, retry=None)[source]
Parameters:
Return type:

Dataset

materialize(destination, *, dataset_id, snapshot_id, records_per_shard=10000, retry=None)[source]

Copy the view into a new dataset.

retry is passed down to each shard write and to the publication rather than wrapping the whole loop: recovering from one failure on the last shard should not re-encode every shard before it.

Parameters:
Return type:

Dataset

class signal_dataset.ReadInstruction(filename, skip, take, examples_in_shard, first_ordinal)[source]

Bases: object

A contiguous run of records within one shard object.

Inert by design: it names bytes and does not fetch them. Iceberg’s FileScanTask and TFDS’s FileInstruction, the two closest analogues, are likewise plain records that the caller reads for itself.

Parameters:

  • filename (str)

  • skip (int)

  • take (int)

  • examples_in_shard (int)

  • first_ordinal (int)

filename: str

Absolute URI of the shard object holding these records.

skip: int

Index of the first record to take, within that shard.

take: int

How many records to take.

examples_in_shard: int

Total records the shard holds, which a reader may use to size a buffer.

first_ordinal: int

Ordinal of the first record in the dataset’s own numbering.

Annotation operations

signal_dataset.publish_annotations(dataset, name, records, *, metadata=None, options=None, retry=None)[source]

Publish a dense annotation set for every record in the dataset.

Safe to re-run. With publication_id left unset it is derived from the snapshot and set name, so a repeat writes to the same places and converges rather than orphaning a new set of objects under a fresh identity.

Parameters:
Return type:

AnnotationSet

class signal_dataset.AnnotationRecord(source_record_id: 'str', source_index: 'int', status: 'AnnotationStatus', values: 'Mapping[str, Json]' = <factory>, fields: 'Mapping[str, Field]' = <factory>, provenance: 'Mapping[str, Json]' = <factory>, detail_status: 'str | None' = None)[source]

Bases: object

Parameters:
source_record_id: str
source_index: int
status: AnnotationStatus
values: Mapping[str, None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]]]
fields: Mapping[str, Field]
provenance: Mapping[str, None | bool | int | float | str | list[None | bool | int | float | str | list[Json] | dict[str, Json]] | dict[str, None | bool | int | float | str | list[Json] | dict[str, Json]]]
detail_status: str | None
class signal_dataset.AnnotationStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: StrEnum

SUCCESS = 'success'
SKIPPED = 'skipped'
FAILED = 'failed'
class signal_dataset.AnnotationSet(root, specification, objects, shards, options, source_metadata)[source]

Bases: Sequence[AnnotationRecord]

Parameters:
class signal_dataset.AnnotationCatalog(values, revision)[source]

Bases: Mapping[str, AnnotationSet]

Parameters:

Options and policies

class signal_dataset.StorageOptions(writer_options: str = 'group_size:1', reader_options: str = 'readahead_buffer_size:0, max_parallelism:0', file_reader_buffer_size: int | None = None, resources: signal_dataset._internal.policy.ResourcePolicy = <factory>, control: signal_dataset._internal.policy.ControlPolicy = <factory>, manifest_cache_pages: int = 16, records_per_read_batch: int = 1024, max_records_per_shard: int | None = None, max_encoded_shard_bytes: int | None = None, max_open_shards: int = 4)[source]

Bases: object

Parameters:

  • writer_options (str)

  • reader_options (str)

  • file_reader_buffer_size (int | None)

  • resources (ResourcePolicy)

  • control (ControlPolicy)

  • manifest_cache_pages (int)

  • records_per_read_batch (int)

  • max_records_per_shard (int | None)

  • max_encoded_shard_bytes (int | None)

  • max_open_shards (int)

writer_options: str
reader_options: str
file_reader_buffer_size: int | None
resources: ResourcePolicy
control: ControlPolicy
manifest_cache_pages: int
records_per_read_batch: int
max_records_per_shard: int | None
max_encoded_shard_bytes: int | None
max_open_shards: int

How many staged shards to keep open at once, on a backend that stages.

Only s3:// stages; local and gs:// hand the URI straight to the reader. A staging miss re-downloads a whole object, so under an access order that alternates between more shards than are retained, every read costs a full shard. One is the right answer for a single sequential scan and the wrong one for a shuffled reader, which is the shape a training loop usually has – so this is a choice the caller has to be able to make, and until now could not.

class signal_dataset.PublicationOptions(shards_per_manifest=10000, verify_record_counts=False)[source]

Bases: object

How a publication is laid out, and how hard it checks its inputs.

verify_record_counts re-opens every shard a producer reported and counts its records. That is exact, and on object storage it costs a full read of everything being published – for a terabyte dataset, a terabyte downloaded, twice, for data and metadata. It is off by default. The byte size of each shard is always compared, which costs one HEAD and catches the failure that actually happens: a truncated or half-written upload.

Parameters:

  • shards_per_manifest (int)

  • verify_record_counts (bool)

shards_per_manifest: int
verify_record_counts: bool
class signal_dataset.AnnotationOptions(records_per_shard: int = 10000, catalog_retry_limit: int = 8, publication_id: str | None = None, verify_record_counts: bool = False)[source]

Bases: object

Parameters:

  • records_per_shard (int)

  • catalog_retry_limit (int)

  • publication_id (str | None)

  • verify_record_counts (bool)

records_per_shard: int
catalog_retry_limit: int
publication_id: str | None
verify_record_counts: bool

See PublicationOptions.verify_record_counts; same trade, same default.

class signal_dataset.RetryPolicy(attempts=5, base_seconds=1.0, max_seconds=60.0, jitter=True)[source]

Bases: object

How often, and how patiently, to retry a transient failure.

The defaults are tuned for object storage rather than for local contention. A bucket that just returned 503 is unlikely to be healthy a tenth of a second later, so the first pause is a full second; five attempts span roughly half a minute of trouble. Waiting costs only time, because the operation underneath is idempotent.

Parameters:
attempts: int
base_seconds: float
max_seconds: float
jitter: bool
delay(attempt)[source]

Capped exponential delay before the attempt after attempt.

Jitter matters more than the exponent: several workers that fail at the same moment and retry in lockstep reproduce whatever they collided with.

Parameters:

attempt (int)

Return type:

float

classmethod coerce(value)[source]

Accept retry=3, retry=RetryPolicy(…), or retry=None.

The integer shorthand is the spelling callers reach for first; the dataclass is there when the delays need tuning too.

Parameters:

value (int | RetryPolicy | None)

Return type:

RetryPolicy

class signal_dataset.ControlPolicy(max_document_bytes=67108864, max_json_depth=128, max_json_nodes=1000000)[source]

Bases: object

Bounds untrusted dataset roots, snapshots, manifests, and shard references.

Parameters:

  • max_document_bytes (int)

  • max_json_depth (int)

  • max_json_nodes (int)

max_document_bytes: int
max_json_depth: int
max_json_nodes: int
class signal_dataset.ResourcePolicy(max_encoded_record_bytes=68719476736, max_descriptor_bytes=16777216, max_fields=10000, max_axes_per_field=128, max_coordinate_values=10000000, max_tensor_bytes=17179869184, max_total_tensor_bytes=68719476736, max_metadata_bytes=16777216, max_json_depth=128, max_json_nodes=1000000)[source]

Bases: object

Bounds untrusted tensor-bearing records and their descriptors.

Parameters:

  • max_encoded_record_bytes (int)

  • max_descriptor_bytes (int)

  • max_fields (int)

  • max_axes_per_field (int)

  • max_coordinate_values (int)

  • max_tensor_bytes (int)

  • max_total_tensor_bytes (int)

  • max_metadata_bytes (int)

  • max_json_depth (int)

  • max_json_nodes (int)

max_encoded_record_bytes: int
max_descriptor_bytes: int
max_fields: int
max_axes_per_field: int
max_coordinate_values: int
max_tensor_bytes: int
max_total_tensor_bytes: int
max_metadata_bytes: int
max_json_depth: int
max_json_nodes: int

Record codecs

signal_dataset.encode(record, *, policy=None)
Parameters:
Return type:

bytes

signal_dataset.decode(data, *, policy=None)
Parameters:
Return type:

Record

Storage extension contracts

Implement these to add a transport or a shard container. See architecture for the dependency direction they must respect.

class signal_dataset.storage.ObjectStore[source]

Bases: ABC

Immutable-object transport with version-aware control updates.

supports_compare_and_swap: bool = True

Whether this store can modify an object in place without losing a concurrent writer’s update.

Almost everything in the format is written once, so a store that cannot do this is still fully usable for publishing datasets. The exception is the annotation catalog, which is swung from one revision to the next under a compare-and-swap; a store answering False refuses that, and callers check here before doing any work rather than failing at the swap.

A plain class attribute rather than an abstract member, so that every ObjectStore written against an earlier release keeps working unchanged.

abstract read(uri, *, generation=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

Return type:

bytes

abstract read_version(uri)[source]
Parameters:

uri (str)

Return type:

ObjectVersion

abstract info(uri, *, generation=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

Return type:

ObjectInfo

abstract create(uri, data, *, content_type)[source]
Parameters:
Return type:

int | None

abstract create_file(uri, path, *, content_type)[source]
Parameters:
Return type:

int | None

abstract compare_and_swap(uri, data, *, expected_generation, content_type)[source]
Parameters:

  • uri (str)

  • data (bytes)

  • expected_generation (int | None)

  • content_type (str)

Return type:

int | None

abstract generation(uri)[source]
Parameters:

uri (str)

Return type:

int | None

class signal_dataset.storage.ShardStore[source]

Bases: ABC

Indexed byte-record container independent of logical codecs.

abstract write(path, records, *, options)[source]
Parameters:
Return type:

int

abstract read(uri, index, *, generation=None, options, file_reader_buffer_size=None)[source]
Parameters:

  • uri (str)

  • index (int)

  • generation (int | None)

  • options (str)

  • file_reader_buffer_size (int | None)

Return type:

bytes

read_many(uri, indices, *, generation=None, options, file_reader_buffer_size=None)[source]

Read several entries while preserving the requested index order.

Implementations may override this method to reuse one underlying shard reader. The default preserves compatibility for custom shard stores.

Parameters:
Return type:

tuple[bytes, …]

abstract count(uri, *, generation=None, options, file_reader_buffer_size=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

  • options (str)

  • file_reader_buffer_size (int | None)

Return type:

int

class signal_dataset.storage.ShardStagingArea[source]

Bases: ABC

Makes a stored shard openable by a path-only indexed reader.

abstract stage(uri, *, generation=None)[source]

Return a context manager yielding a path the reader may open.

The path is valid only while the context is active. On exit, including on an exception, the implementation releases everything it acquired for this call.

Parameters:

  • uri (str)

  • generation (int | None)

Return type:

AbstractContextManager[str]

class signal_dataset.storage.ObjectStoreRegistry[source]

Bases: object

Maps URI schemes to backends.

Registration is additive and guarded: re-registering a scheme raises unless replace is passed, so an import cycle or a duplicated plugin cannot silently redirect an existing scheme. There is deliberately no way to unregister.

register(backend, *, replace=False)[source]
Parameters:
Return type:

None

backend_for(uri)[source]
Parameters:

uri (str)

Return type:

ObjectStoreBackend

schemes()[source]
Return type:

frozenset[str]

class signal_dataset.storage.ObjectStoreBackend(*, scheme, filesystem, factory, validate_root)[source]

Bases: object

One storage service, keyed by the URI scheme it claims.

Parameters:

  • scheme (str)

  • filesystem (bool)

  • factory (ObjectStoreFactory)

  • validate_root (Callable[[str], None])

scheme: str
filesystem: bool

Whether this backend’s URIs are operating-system paths.

factory: ObjectStoreFactory
validate_root: Callable[[str], None]
class signal_dataset.storage.ObjectInfo(size: 'int', generation: 'int | None' = None)[source]

Bases: object

Parameters:

  • size (int)

  • generation (int | None)

size: int
generation: int | None
class signal_dataset.storage.ObjectVersion(data: 'bytes', generation: 'int | None' = None)[source]

Bases: object

Parameters:
data: bytes
generation: int | None
class signal_dataset.storage.FilesystemCapabilities(*, hard_links, file_locks, directory_fsync, atomic_rename, stable_inodes)[source]

Bases: object

Which primitives the filesystem beneath a local root provides.

hard_links

link(2). The backend publishes by writing a temporary file and linking it to its final name, which is atomic and fails if the name is taken – create-only semantics for free. gcsfuse and Mountpoint implement no hard links at all.

file_locks

flock(2). The only compare-and-swap in the format guards annotations/catalog.json, the one mutable object, and it needs a critical section.

directory_fsync

fsync on a directory descriptor, which is what makes a newly linked name durable. A mount rejects it, and commits on close regardless.

atomic_rename

rename(2) replacing an existing name in one step. On a mounted bucket this is copy-then-delete, so a reader can observe the target missing.

stable_inodes

(st_dev, st_ino) identifying a directory across opens. A mount synthesizes inode numbers, so the identity check neither detects a replaced root nor is safe to rely on.

Parameters:

  • hard_links (bool)

  • file_locks (bool)

  • directory_fsync (bool)

  • atomic_rename (bool)

  • stable_inodes (bool)

file_locks: bool
directory_fsync: bool
atomic_rename: bool
stable_inodes: bool
signal_dataset.storage.object_store_for(uri, *, root_is_directory=False, registry=None)[source]

Return a store able to serve uri.

Parameters:
Return type:

ObjectStore

Built-in implementations:

class signal_dataset.storage.LocalObjectStore(*, root=None, file_mode=416, capabilities=None)[source]

Bases: ObjectStore

Local storage, optionally confined with descriptor-relative operations.

A configured root is bound to the directory identity observed at construction. Operations reject symlinks below it and cannot be redirected by replacing the root or one of its ancestors.

Parameters:
read(uri, *, generation=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

Return type:

bytes

read_version(uri)[source]
Parameters:

uri (str)

Return type:

ObjectVersion

info(uri, *, generation=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

Return type:

ObjectInfo

create(uri, data, *, content_type)[source]
Parameters:
Return type:

None

create_file(uri, path, *, content_type)[source]
Parameters:
Return type:

None

compare_and_swap(uri, data, *, expected_generation, content_type)[source]
Parameters:

  • uri (str)

  • data (bytes)

  • expected_generation (int | None)

  • content_type (str)

Return type:

None

list(prefix_uri)[source]
Parameters:

prefix_uri (str)

Return type:

list[str]

generation(uri)[source]
Parameters:

uri (str)

Return type:

int

indexed_uri(uri)[source]

Yield a stable descriptor path suitable for a native indexed reader.

Parameters:

uri (str)

Return type:

Iterator[str]

class signal_dataset.storage.GCSObjectStore(client=None, *, precondition_failed=None, api_error=None)[source]

Bases: ObjectStore

Parameters:
read(uri, *, generation=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

Return type:

bytes

read_version(uri)[source]
Parameters:

uri (str)

Return type:

ObjectVersion

info(uri, *, generation=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

Return type:

ObjectInfo

create(uri, data, *, content_type)[source]
Parameters:
Return type:

int

create_file(uri, path, *, content_type)[source]
Parameters:
Return type:

int

compare_and_swap(uri, data, *, expected_generation, content_type)[source]
Parameters:

  • uri (str)

  • data (bytes)

  • expected_generation (int | None)

  • content_type (str)

Return type:

int

list(prefix_uri)[source]
Parameters:

prefix_uri (str)

Return type:

list[str]

generation(uri)[source]
Parameters:

uri (str)

Return type:

int

class signal_dataset.storage.S3ObjectStore(client=None, *, options=None, client_error=None)[source]

Bases: ObjectStore

Create-only object transport over S3 conditional writes.

Parameters:
property options: S3Options
read(uri, *, generation=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

Return type:

bytes

read_version(uri)[source]
Parameters:

uri (str)

Return type:

ObjectVersion

info(uri, *, generation=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

Return type:

ObjectInfo

generation(uri)[source]
Parameters:

uri (str)

Return type:

int

create(uri, data, *, content_type)[source]
Parameters:
Return type:

None

create_file(uri, path, *, content_type)[source]
Parameters:
Return type:

None

compare_and_swap(uri, data, *, expected_generation, content_type)[source]
Parameters:

  • uri (str)

  • data (bytes)

  • expected_generation (int | None)

  • content_type (str)

Return type:

None

download_to_fileobj(uri, stream, *, generation=None)[source]

Stream an object into a file object without buffering it whole.

Parameters:
Return type:

None

list(prefix_uri)[source]
Parameters:

prefix_uri (str)

Return type:

list[str]

head(uri)[source]

Return an object’s size and ETag in one request.

Parameters:

uri (str)

Return type:

tuple[int, str]

put(uri, body, *, content_type, condition, content_length)[source]

Write conditionally, retrying only the racing-writer conflict.

Parameters:
Return type:

None

backoff(attempt)[source]

Capped exponential delay with jitter.

A 409 means writers collided, so retrying them in lockstep reproduces the collision. The cap keeps a generous attempt limit from turning into an unbounded sleep.

The schedule itself is RetryPolicy’s, so this backend and the operation-level retry share one implementation. What differs is the tuning: this retries a conflict on a single key, which clears in milliseconds, so its defaults are far shorter.

Parameters:

attempt (int)

Return type:

float

remember(etag)[source]

Mint a token and record the ETag it came from.

Parameters:

etag (str)

Return type:

int

static rewind(body)[source]
Parameters:

body (bytes | IO[bytes])

Return type:

None

require_etag(response)[source]
Parameters:

response (dict[str, Any])

Return type:

str

reject_generation(generation)[source]
Parameters:

generation (int | None)

Return type:

None

class signal_dataset.storage.S3Options(region_name=None, endpoint_url=None, profile_name=None, addressing_style=None, max_attempts=5, retry_mode='standard', conflict_attempt_limit=5, conflict_retry_base_seconds=0.05, conflict_retry_max_seconds=1.0, etag_memo_size=128, max_single_part_bytes=5368709120)[source]

Bases: object

Transport settings for S3ObjectStore.

Parameters:

  • region_name (str | None)

  • endpoint_url (str | None)

  • profile_name (str | None)

  • addressing_style (str | None)

  • max_attempts (int)

  • retry_mode (str)

  • conflict_attempt_limit (int)

  • conflict_retry_base_seconds (float)

  • conflict_retry_max_seconds (float)

  • etag_memo_size (int)

  • max_single_part_bytes (int)

region_name: str | None
endpoint_url: str | None
profile_name: str | None
addressing_style: str | None
max_attempts: int
retry_mode: str
conflict_attempt_limit: int

1 means no retry.

Type:

Total conditional-write attempts, not retries

conflict_retry_base_seconds: float
conflict_retry_max_seconds: float
etag_memo_size: int
max_single_part_bytes: int
class signal_dataset.storage.ArrayRecordShardStore(*, staging=None)[source]

Bases: ShardStore

Reads indexed records through a staging area.

The binding opens a path and knows one remote scheme, so a shard stored anywhere else has to be materialized first. Staging is injected rather than wrapped around this class: ShardStore.read_many falls back to calling read per index, so a decorator that staged in both would download the same object once per record.

Parameters:

staging (ShardStagingArea | None)

property staging: ShardStagingArea
write(path, records, *, options)[source]
Parameters:
Return type:

int

read(uri, index, *, generation=None, options, file_reader_buffer_size=None)[source]
Parameters:

  • uri (str)

  • index (int)

  • generation (int | None)

  • options (str)

  • file_reader_buffer_size (int | None)

Return type:

bytes

read_many(uri, indices, *, generation=None, options, file_reader_buffer_size=None)[source]

Read ordered entries with one staging and one reader lifetime.

Parameters:
Return type:

tuple[bytes, …]

count(uri, *, generation=None, options, file_reader_buffer_size=None)[source]
Parameters:

  • uri (str)

  • generation (int | None)

  • options (str)

  • file_reader_buffer_size (int | None)

Return type:

int

class signal_dataset.storage.EphemeralFileStaging(*, objects=None, directory=None, max_object_bytes=None, max_open_shards=4)[source]

Bases: ShardStagingArea

Materializes a shard as an unnamed local file, retaining it for reuse.

Parameters:
stage(uri, *, generation=None)[source]

Return a context manager yielding a path the reader may open.

The path is valid only while the context is active. On exit, including on an exception, the implementation releases everything it acquired for this call.

Parameters:

  • uri (str)

  • generation (int | None)

Return type:

Iterator[str]

acquire(uri, generation)[source]

Return a descriptor for uri, counting this caller as a reader.

Every acquire must be matched by a release. Eviction alone must never close a descriptor: another thread may be reading through it right now, and closing it underneath would hand that reader a descriptor number the kernel is free to reuse for something else.

Parameters:

  • uri (str)

  • generation (int | None)

Return type:

int

release(descriptor)[source]

Mark one reader as finished, closing the descriptor if it was evicted.

Parameters:

descriptor (int)

Return type:

None

materialize(uri, generation)[source]

Fetch an object into a fresh unnamed file and return its descriptor.

Parameters:

  • uri (str)

  • generation (int | None)

Return type:

int

close()[source]

Release every retained descriptor.

Descriptors still being read are retired rather than closed, so a concurrent stage() block finishes against a descriptor that is still open and the last reader out closes it.

Return type:

None

check_size(store, uri, generation)[source]
Parameters:
Return type:

None

static warn_if_still_named(descriptor, name)[source]

Detect a filesystem that renames rather than unlinks an open file.

Parameters:
Return type:

None

static fetch(store, uri, stream, generation)[source]
Parameters:
Return type:

None

class signal_dataset.storage.PassthroughStaging[source]

Bases: ShardStagingArea

Yields the URI itself, pinning a generation when one is given.

stage(uri, *, generation=None)[source]

Return a context manager yielding a path the reader may open.

The path is valid only while the context is active. On exit, including on an exception, the implementation releases everything it acquired for this call.

Parameters:

  • uri (str)

  • generation (int | None)

Return type:

Iterator[str]

Exceptions

See errors for what raises each one and what to do about it.

exception signal_dataset.SignalDatasetError[source]

Bases: Exception

Base class for package-specific failures.

exception signal_dataset.ValidationError[source]

Bases: SignalDatasetError, ValueError

A logical structure or resource policy is invalid.

exception signal_dataset.PublicationCollisionError[source]

Bases: SignalDatasetError, FileExistsError

A create-only publication target already exists.

exception signal_dataset.CorruptDatasetError[source]

Bases: SignalDatasetError, ValueError

A dataset control document or shard is inconsistent.

exception signal_dataset.StorageError[source]

Bases: SignalDatasetError, OSError

A storage operation failed.

Bring your own loader

SDS defines a dataset format. It ships a reader because reading one requires knowing the format – root.json, snapshot and manifest paging, ArrayRecord shards holding SafeTensors payloads – and without one, every consumer would reimplement exactly the contract this library exists to define.

It does not ship a data loader. Order, concurrency, prefetch depth and caching belong to whatever feeds the trainer, and the two frameworks in use here, Grain and PyTorch, disagree about all of them.

The seam is that addressing and decoding are public and separate from fetching, following the split Apache Iceberg draws between planning a scan and executing it.

who owns it

where ordinal i lives

SDS, because it is the format

fetching those bytes

you

bytes to a typed Record

SDS, because it is the format

order, workers, prefetch, caching

you

 
from array_record.python.array_record_module import ArrayRecordReader

import signal_dataset as sds


def my_reader(filename: str, skip: int, take: int) -> list[bytes]:
    """Your loader owns this fetch: any client, any concurrency, any caching."""
    reader = ArrayRecordReader(filename)
    try:
        return reader.read(list(range(skip, skip + take)))
    finally:
        reader.close()


dataset = sds.open("captures.sds")

for instruction in dataset.read_instructions(0, len(dataset)):
    payloads = my_reader(instruction.filename, instruction.skip, instruction.take)
    records = [sds.decode(payload) for payload in payloads]

assert len(records) > 0

read_instructions(start, stop, *, metadata=False) returns ReadInstruction(filename, skip, take, examples_in_shard, first_ordinal) and performs no I/O beyond manifest pages already loaded, so it is safe inside a DataLoader.__init__ or a worker. The first four field names are the ones TFDS and ArrayRecord already agreed on, so readers built against that protocol accept these unchanged.

metadata=True addresses the aligned metadata shards instead. Those carry every record’s identity and shape without its samples, which makes shuffling, filtering and sampling cheap before any tensor is fetched.

The source contract

A source satisfying Grain’s contract satisfies PyTorch’s, because Grain’s is the superset. Writing to this list once is cheaper than maintaining two adapters:

  • __len__, and __getitem__ taking an integer

  • thread-safe. Grain calls it from a thread pool, 16 threads by default. Dataset qualifies: its manifest cache and shard-verification set are both guarded, and test_a_dataset_is_safe_to_read_from_many_threads exercises one shared reader from sixteen threads with fewer cached pages than shards, so eviction runs while those threads are reading.

  • deterministic. A given index returns the same value.

  • picklable, holding no live reader: Grain spawns and cloudpickles the source, PyTorch forks today and is moving to forkserver. Keep the URI, reopen lazily.

  • a stable __repr__. Grain’s DataLoader writes it into a checkpoint and compares it on restore, so a default repr carrying a memory address fails every restore. Dataset provides one.

sds.Dataset satisfies all of it, so it can be used as a source directly. The worked examples are Grain and PyTorch; neither is privileged, and the library imports neither.