Source code for signal_dataset.dataset.instructions

"""Where records live, without reading any of them.

The library ships a reader, and a training pipeline may not want it: the loader
already owns prefetch depth, worker count, shuffling and caching, and those are
its decisions to make. What it cannot own is knowing where ordinal 4,821,003
lives -- that is the format, and reimplementing it means parsing ``root.json``,
snapshot and manifest paging, and the ArrayRecord and SafeTensors layout, which
is the whole thing this library exists to define.

So the addressing is public, and separate from the fetching. A caller asks
where a range of records is, fetches those bytes however it likes, and turns
them back into records with :func:`signal_dataset.decode`.

The shape is deliberately the one TFDS and ArrayRecord already agreed on.
``array_record_data_source.py`` defines a ``FileInstruction`` protocol -- the
fields below, structurally checked -- specifically so that a library which
*plans* can hand descriptors to a reader which *executes*, with no dependency
between them. Emitting the same shape means those readers accept these
instructions unchanged.
"""

from __future__ import annotations

from dataclasses import dataclass




@dataclass(frozen=True, slots=True)
class ReadInstruction:
    """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.
    """

    #: Absolute URI of the shard object holding these records.
    filename: str
    #: Index of the first record to take, within that shard.
    skip: int
    #: How many records to take.
    take: int
    #: Total records the shard holds, which a reader may use to size a buffer.
    examples_in_shard: int
    #: Ordinal of the first record in the dataset's own numbering.
    first_ordinal: int

    def __post_init__(self) -> None:
        for name in ("skip", "take", "examples_in_shard", "first_ordinal"):
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int) or value < 0:
                raise ValueError(f"{name} must be a nonnegative integer")
        if self.skip + self.take > self.examples_in_shard:
            raise ValueError("instruction runs past the end of its shard")