Skip to content

Contracts

Contracts define the shared column and metadata objects used by dataset configs, processing stages, and generated outputs.

MappingSpec

Bases: str

Canonical column name with dtype, aliases, parser, and optional values.

MappingSpec is a str subclass, so it can be passed directly to Polars selectors and aliases while carrying schema metadata for validation and raw column discovery. Common columns live in BaseColumns; dataset mapping modules derive dataset-specific variants with aliases or parsers.

Examples:

>>> curr = MappingSpec("Current [A]", dtype=pl.Float64)
>>> str(curr)
'Current [A]'
>>> curr.dtype
Float64

>>> raw_time = BaseColumns.time.with_alias("Test_Time(s)", "Test Time (s)")
>>> raw_time.matching_name(["Test Time (s)"])
'Test Time (s)'

>>> parsed_time = raw_time.with_parser(lambda col: pl.col(col).cast(pl.Float64))
>>> parsed_time.parser is not None
True

>>> git_status = BaseColumns.git_status
>>> git_status.values.clean
'clean'

Methods:

  • with_alias –

    Return a copy with dataset-specific source column aliases.

  • with_parser –

    Return a copy with a raw-column parser expression factory.

  • with_values –

    Return a copy with fixed named values attached.

  • matching_name –

    Return the first available column matching this spec or its aliases.

Attributes:

  • values (ValuesT) –

    Fixed named values declared for this column.

values property 

values: ValuesT

Fixed named values declared for this column.

Returns:

  • ValuesT –

    The values namespace attached to this spec.

Raises:

  • TypeError –

    If no values namespace was declared.

with_alias 

with_alias(*alias: str) -> MappingSpec[ValuesT]

Return a copy with dataset-specific source column aliases.

Parameters:

  • alias (str, default: () ) –

    Raw/source column names that should match this canonical spec.

Returns:

  • MappingSpec[ValuesT] –

    A spec with the same canonical name and metadata, but the supplied

  • MappingSpec[ValuesT] –

    source aliases.

Source code in batgrad/contracts/mapping.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def with_alias(self, *alias: str) -> MappingSpec[ValuesT]:
    """Return a copy with dataset-specific source column aliases.

    Args:
        alias: Raw/source column names that should match this canonical spec.

    Returns:
        A spec with the same canonical name and metadata, but the supplied
        source aliases.
    """
    return MappingSpec[ValuesT](
        str(self),
        dtype=self.dtype,
        alias=alias,
        description=self.description,
        parser=self.parser,
        values=self._values,
    )

with_parser 

with_parser(
    parser: Callable[[str], Expr],
) -> MappingSpec[ValuesT]

Return a copy with a raw-column parser expression factory.

Parameters:

  • parser (Callable[[str], Expr]) –

    Callable receiving the matched raw column name and returning a Polars expression for the parsed canonical column.

Returns:

  • MappingSpec[ValuesT] –

    A spec with the same canonical name, aliases, and values, but the

  • MappingSpec[ValuesT] –

    supplied parser.

Source code in batgrad/contracts/mapping.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def with_parser(self, parser: Callable[[str], pl.Expr]) -> MappingSpec[ValuesT]:
    """Return a copy with a raw-column parser expression factory.

    Args:
        parser: Callable receiving the matched raw column name and returning
            a Polars expression for the parsed canonical column.

    Returns:
        A spec with the same canonical name, aliases, and values, but the
        supplied parser.
    """
    return MappingSpec[ValuesT](
        str(self),
        dtype=self.dtype,
        alias=self.alias[1:],
        description=self.description,
        parser=parser,
        values=self._values,
    )

with_values 

with_values(values: NewValuesT) -> MappingSpec[NewValuesT]

Return a copy with fixed named values attached.

Parameters:

  • values (NewValuesT) –

    Namespace of allowed or conventional values for the column.

Returns:

  • MappingSpec[NewValuesT] –

    A spec with the same canonical name, aliases, parser, and dtype, but

  • MappingSpec[NewValuesT] –

    the supplied values namespace.

Source code in batgrad/contracts/mapping.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def with_values[NewValuesT](self, values: NewValuesT) -> MappingSpec[NewValuesT]:
    """Return a copy with fixed named values attached.

    Args:
        values: Namespace of allowed or conventional values for the column.

    Returns:
        A spec with the same canonical name, aliases, parser, and dtype, but
        the supplied values namespace.
    """
    return MappingSpec[NewValuesT](
        str(self),
        dtype=self.dtype,
        alias=self.alias[1:],
        description=self.description,
        parser=self.parser,
        values=values,
    )

matching_name 

matching_name(
    columns: set[str] | tuple[str, ...] | list[str],
) -> str | None

Return the first available column matching this spec or its aliases.

Parameters:

  • columns (set[str] | tuple[str, ...] | list[str]) –

    Available column names to search. Matching is case-insensitive and checks the canonical name first, followed by aliases in declaration order.

Returns:

  • str | None –

    The original available column name when a match is found, otherwise

  • str | None –

    None.

Source code in batgrad/contracts/mapping.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def matching_name(self, columns: set[str] | tuple[str, ...] | list[str]) -> str | None:
    """Return the first available column matching this spec or its aliases.

    Args:
        columns: Available column names to search. Matching is
            case-insensitive and checks the canonical name first, followed by
            aliases in declaration order.

    Returns:
        The original available column name when a match is found, otherwise
        `None`.
    """
    available = {column.casefold(): column for column in columns}
    for alias in self.alias:
        match = available.get(alias.casefold())
        if match is not None:
            return match
    return None

MetadataLayout dataclass 

MetadataLayout(
    required: tuple[MappingSpec, ...] = (),
    optional: Mapping[MappingSpec, object | None] = dict(),
)

Required columns and optional default values for stage metadata.

Layouts are used for both manifest rows and parquet footers. required columns must be present; optional entries provide default metadata values that can be added or overridden by dataset/protocol-specific configuration.

Attributes:

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

    Metadata columns that must be present.

  • optional (Mapping[MappingSpec, object | None]) –

    Metadata columns and default values that may be added by the stage, dataset, or protocol configuration.

Examples:

>>> layout = MetadataLayout(required=(BaseColumns.proto,))
>>> expanded = layout.with_optional({BaseColumns.cell_id: None})
>>> expanded.columns
('protocol', 'cell id')
>>> expanded.values
{'cell id': None}

Methods:

  • with_optional –

    Return a layout with additional or overridden optional defaults.

columns property 

columns: tuple[MappingSpec, ...]

Required columns followed by optional columns.

Returns:

values property 

values: dict[MappingSpec, object | None]

Optional metadata defaults as a mutable copy.

Returns:

with_optional 

with_optional(
    optional: Mapping[MappingSpec, object | None],
) -> MetadataLayout

Return a layout with additional or overridden optional defaults.

Parameters:

  • optional (Mapping[MappingSpec, object | None]) –

    Optional metadata columns and default values to merge into this layout.

Returns:

  • MetadataLayout –

    A layout with the same required columns and merged optional defaults.

Source code in batgrad/contracts/metadata.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def with_optional(self, optional: Mapping[MappingSpec, object | None]) -> MetadataLayout:
    """Return a layout with additional or overridden optional defaults.

    Args:
        optional: Optional metadata columns and default values to merge into
            this layout.

    Returns:
        A layout with the same required columns and merged optional defaults.
    """
    values = dict(self.optional)
    values.update(optional)
    return MetadataLayout(
        required=self.required,
        optional=values,
    )

StageLayout dataclass 

StageLayout(
    stage_id: DatasetStageId,
    manifest: MetadataLayout,
    footer: MetadataLayout,
)

Manifest and footer metadata contract for one processing stage.

Manifest metadata is validated against and written to manifest rows. Footer metadata is encoded as parquet key/value metadata when data files are written. Optional values set to None declare metadata that is expected from task metadata rather than a fixed stage-level default.

Attributes:

  • stage_id (DatasetStageId) –

    Pipeline stage this layout describes.

  • manifest (MetadataLayout) –

    Metadata required or defaulted in the stage manifest rows.

  • footer (MetadataLayout) –

    Metadata required or defaulted in generated parquet footers.

Examples:

>>> layout = INGEST_STAGE_METADATA.with_manifest(
...     {BaseColumns.cell_id: None, BaseColumns.cidx: None}
... ).with_footer({BaseColumns.time_conv: "start_of_interval"})
>>> BaseColumns.cell_id in layout.manifest.optional
True
>>> layout.footer.values[BaseColumns.time_conv]
'start_of_interval'

Methods:

  • with_manifest –

    Return a stage layout with extra optional manifest metadata.

  • with_footer –

    Return a stage layout with extra optional footer metadata.

with_manifest 

with_manifest(
    optional: Mapping[MappingSpec, object | None],
) -> StageLayout

Return a stage layout with extra optional manifest metadata.

Parameters:

Returns:

  • StageLayout –

    A stage layout with updated manifest metadata and unchanged footer

  • StageLayout –

    metadata.

Source code in batgrad/contracts/metadata.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def with_manifest(self, optional: Mapping[MappingSpec, object | None]) -> StageLayout:
    """Return a stage layout with extra optional manifest metadata.

    Args:
        optional: Manifest metadata columns and default values to add or
            override.

    Returns:
        A stage layout with updated manifest metadata and unchanged footer
        metadata.
    """
    return StageLayout(
        stage_id=self.stage_id,
        manifest=self.manifest.with_optional(optional),
        footer=self.footer,
    )

with_footer(
    optional: Mapping[MappingSpec, object | None],
) -> StageLayout

Return a stage layout with extra optional footer metadata.

Parameters:

Returns:

  • StageLayout –

    A stage layout with unchanged manifest metadata and updated footer

  • StageLayout –

    metadata.

Source code in batgrad/contracts/metadata.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def with_footer(self, optional: Mapping[MappingSpec, object | None]) -> StageLayout:
    """Return a stage layout with extra optional footer metadata.

    Args:
        optional: Footer metadata columns and default values to add or
            override.

    Returns:
        A stage layout with unchanged manifest metadata and updated footer
        metadata.
    """
    return StageLayout(
        stage_id=self.stage_id,
        manifest=self.manifest,
        footer=self.footer.with_optional(optional),
    )