Ingested¶

Runtime and parquet-writing settings for the ingest stage.

Attributes:

• n_jobs (int) –

  Worker count. Use 1 for sequential execution, -1 for available CPUs minus one, or a positive count capped by task count.

• worker_polars_max_threads (int | None) –

  Polars threads per worker. -1 divides CPUs across workers, None leaves Polars unrestricted, and a positive value sets an exact thread count.

• chunk_rows (int) –

  Rows read from temporary task outputs at a time.

• compression (str) –

  Parquet compression codec.

• use_content_defined_chunking (bool) –

  Whether table writers may use content defined chunking.

• row_group_size (int) –

  Parquet row group size for written files.

• max_shard_size_bytes (int) –

  Roll a protocol shard after this approximate size; 0 disables size-based rolling.

Examples:

>>> IngestStageConfig(n_jobs=-1, worker_polars_max_threads=-1)
IngestStageConfig(...)

Dataset-level ingest configuration.

Include/exclude patterns are used by raw adapters when planning source-file tasks. Stage metadata is expanded with each protocol's task keys and manifest extras before the output manifest is written.

Attributes:

• metadata (StageLayout) –

  Base stage metadata layout.

• included_file_patterns (tuple[str, ...]) –

  Raw files an adapter should consider.

• excluded_file_patterns (tuple[str, ...]) –

  Raw files an adapter should ignore.

• protocol_specs (tuple[IngestProtocolSpec, ...]) –

  Protocol-specific raw mappings.

Examples:

>>> from batgrad.contracts.metadata import INGEST_STAGE_METADATA
>>> from batgrad.contracts.protocols import BatteryProtocols
>>> protocol_spec = IngestProtocolSpec(
...     protocol=BatteryProtocols.cyc,
...     columns=(BaseColumns.time, BaseColumns.curr, BaseColumns.volt),
... )
>>> spec = IngestStageSpec(
...     metadata=INGEST_STAGE_METADATA,
...     included_file_patterns=("*.xlsx",),
...     protocol_specs=(protocol_spec,),
... )
>>> spec.is_included_file("raw/cell.xlsx")
True

Methods:

• protocol_spec –

  Return the protocol spec matching a protocol id or value.

• output_columns –

  Canonical output columns for a protocol.

• required_metadata –

  Batch metadata required for a protocol.

• manifest_columns –

  Protocol metadata columns written to the ingested manifest.

• is_included_file –

  Return whether a raw file path passes include/exclude patterns.

• output_spec –

  Build the output writer configuration for ingested shards.

protocol_spec(protocol: object) -> IngestProtocolSpec

output_columns(protocol: object) -> tuple[MappingSpec, ...]

required_metadata(
    protocol: object,
) -> tuple[MappingSpec, ...]

manifest_columns(
    protocol: object,
) -> tuple[MappingSpec, ...]

is_included_file(path: str) -> bool

output_spec(dataset_spec: DatasetSpec) -> StageOutputSpec

Raw-to-ingested mapping for one protocol.

Dataset adapters yield raw batches for a protocol. The ingest stage aligns each batch to columns using MappingSpec aliases and parsers, validates protocol metadata, optionally flips current sign, and writes ingested parquet shards grouped by protocol.

Attributes:

• protocol (BatteryProtocolSpec) –

  Shared protocol definition.

• columns (tuple[MappingSpec, ...]) –

  Canonical output columns expected in adapter output.

• metadata (ProtocolMetadata | None) –

  Optional protocol metadata override for this dataset.

• dropped_columns (tuple[MappingSpec, ...]) –

  Declared raw columns that may appear but are omitted.

• flip_current_sign (bool) –

  Whether to multiply canonical current by -1.

Examples:

>>> from batgrad.contracts.protocols import BatteryProtocols
>>> IngestProtocolSpec(
...     protocol=BatteryProtocols.cyc,
...     columns=(BaseColumns.time, BaseColumns.curr, BaseColumns.volt),
...     flip_current_sign=True,
... )
IngestProtocolSpec(...)

protocol_id: DatasetProtocolId

protocol_metadata: ProtocolMetadata

output_columns: tuple[MappingSpec, ...]

manifest_columns: tuple[MappingSpec, ...]

required_metadata: tuple[MappingSpec, ...]

Bases: Protocol

Interface implemented by dataset-specific raw loaders.

Implementations discover raw source files in plan_raw_tasks and load them in load_raw_task. The loader is responsible for inferring protocol and task metadata; the ingest stage handles column alignment, validation, and parquet writing.

Methods:

• load_raw_task –

  Load a planned raw task and yield protocol batches.

• plan_raw_tasks –

  Discover raw source work units for the ingest stage.

load_raw_task(
    task: IngestTask,
    input_store: DataProcessingStore,
    raw_spec: IngestStageSpec,
) -> Iterator[IngestBatch]

plan_raw_tasks(
    input_store: DataProcessingStore,
    raw_spec: IngestStageSpec,
) -> tuple[IngestTask, ...]

One adapter-planned raw ingest unit, usually one source file.

Attributes:

• task_id (str) –

  Stable id used in logs and temporary output paths.

• source_paths (tuple[str, ...]) –

  Raw source paths consumed by this task.

Raw data and metadata yielded by a raw dataset adapter.

metadata must include BaseColumns.proto and all columns required by the selected protocol metadata. The ingest stage writes these values to manifest rows and parquet footer metadata where declared by the stage layout.

Attributes:

• data (DataFrame | LazyFrame) –

  Raw frame yielded by the adapter.

• protocol_id (DatasetProtocolId) –

  Protocol represented by this batch.

• source_paths (tuple[str, ...]) –

  Raw source paths that produced the batch.

• metadata (dict[MappingSpec, object]) –

  Task and protocol metadata for manifest/footer writing.