Normalized¶

Runtime, validation, and parquet-writing settings for normalization.

Tasks are collected in memory when max_batch_rows is None or the task row count fits within that limit. Larger tasks use bounded chunk processing; if a protocol also has resampling, its resampling spec must support bounded execution.

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) –

  Output chunk size written to final shards.

• 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.

• max_batch_rows (int | None) –

  Maximum rows processed in memory per bounded batch. Set to None to collect each task fully.

• apply_resampling (bool) –

  Whether protocol resampling specs are applied.

• apply_physics_compensation (bool) –

  Whether bounded MinMaxLTTB recomputes dt, rebuilt time, and averaged current/C-rate when possible.

Examples:

>>> NormalizeStageConfig(n_jobs=-1, max_batch_rows=500_000)
NormalizeStageConfig(...)

Dataset-level normalization configuration.

The stage expands manifest metadata with raw paths, ingested parquet segment references, resampling method/arguments, time convention, and protocol group keys. time_convention is also written to generated parquet footers.

Attributes:

• metadata (StageLayout) –

  Base normalized-stage metadata layout.

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

  Protocol normalization recipes.

• time_convention (str) –

  Label written to manifest/footer metadata describing how task time values are interpreted.

Examples:

>>> from batgrad.contracts.protocols import BatteryProtocols
>>> NormalizeStageSpec(
...     protocol_specs=(NormalizeProtocolSpec(protocol=BatteryProtocols.cyc),),
...     time_convention="start_of_interval",
... )
NormalizeStageSpec(...)

Methods:

• protocol_spec –

  Return the protocol spec matching a protocol id or value.

• output_columns –

  Canonical output columns for a protocol.

• required_input_columns –

  Ingested columns required for a protocol's normalize tasks.

• task_metadata –

  Metadata written for one normalized task's manifest rows and footers.

• output_spec –

  Build the output writer configuration for normalized shards.

protocol_spec(protocol: object) -> NormalizeProtocolSpec

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

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

task_metadata(
    dataset_id: str, task: NormalizeTask
) -> dict[MappingSpec, object]

output_spec(
    dataset_spec: DatasetSpec,
    *,
    output_root: str | None = None,
    manifest_path: str | None = None,
) -> StageOutputSpec

Normalization recipe for one protocol.

Required input columns are inferred from order columns, transform inputs, output columns not produced by transforms/constants/checks, check inputs, and resampling inputs. When multiple aliases for a requested column are available in ingested shards, normalization coalesces them into the canonical output column.

Resampling runs after transforms and checks. Large tasks only work with a resampler that implements bounded execution; otherwise keep max_batch_rows unset or high enough for full-task processing.

Attributes:

• protocol (BatteryProtocolSpec) –

  Shared protocol definition, including task grouping metadata.

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

  Columns used to sort each task. Defaults to protocol.axis_col.

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

  Canonical output columns to write.

• constant_columns (dict[MappingSpec, object]) –

  Fixed columns added to every output row, such as ambient temperature when it is known from dataset metadata instead of raw measurements.

• transforms (tuple[NormalizeTransformSpec, ...]) –

  Column derivations run before checks.

• checks (tuple[CheckSpec, ...]) –

  Validation or annotation checks run after transforms.

• resampling (ResamplingSpec | None) –

  Optional row reduction or interpolation step.

Examples:

>>> from batgrad.contracts.protocols import BatteryProtocols
>>> from batgrad.data.transforms.checks import MissingCheckSpec, TimeCheckSpec
>>> NormalizeProtocolSpec(
...     protocol=BatteryProtocols.cyc,
...     columns=(
...         BaseColumns.time,
...         BaseColumns.volt,
...         BaseColumns.curr,
...         BaseColumns.amb_temp,
...     ),
...     constant_columns={BaseColumns.amb_temp: 20.0},
...     checks=(MissingCheckSpec(), TimeCheckSpec(BaseColumns.time, BaseColumns.dt)),
... )
NormalizeProtocolSpec(...)

protocol_id: DatasetProtocolId

group_by: tuple[MappingSpec, ...]

output_columns: tuple[MappingSpec, ...]

required_input_columns: tuple[MappingSpec, ...]

One protocol/group normalization unit planned from the ingested manifest.

A task contains the ingested parquet segments for one protocol task key, such as one cell/cycle pair, plus source-path and row-count metadata used for the normalized manifest.

Attributes:

• task_id (str) –

  Stable id used in logs and temporary output paths.

• protocol_id (str) –

  Protocol represented by the task.

• group_values (dict[MappingSpec, object]) –

  Task-key metadata values, such as cell and cycle.

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

  Raw source paths represented by the task.

• parquet_segments (tuple[ParquetSegment, ...]) –

  Ingested parquet segments to read.

• row_count (int) –

  Total input row count across segments.