Contracts#

Data structure contracts enforced at module boundaries. Every pipeline stage validates its inputs and outputs against these contracts.

Grid stage contract.

Enforces that after regridding, the dataset is a valid 2D Cartesian grid suitable for downstream segmentation and projection.

adapt.contracts.grid.assert_gridded(ds, reflectivity_var)#

Enforce grid stage contract.

Parameters:

  • ds (Dataset) – Dataset from loader.load_and_regrid()

  • reflectivity_var (str) – Name of reflectivity variable (from config)

Raises:

ContractViolation – If any invariant is violated

Return type:

None

adapt.contracts.grid.check_grid_ds_2d(ds)#

Bound contract for the standard 2D grid output.

Structural only: x/y present, dataset non-empty. Which FIELD must be present depends on config (global_.tracking_field) and is validated by the detection module at its own boundary — bound contracts have no config access by design.

Return type:

None

Parameters:

ds (Dataset)

Segmentation stage contract.

Enforces that after cell detection, labels are present, integer-typed, non-negative, and 2D.

adapt.contracts.segmentation.assert_segmented(ds, labels_name)#

Enforce segmentation stage contract.

Parameters:

  • ds (Dataset) – Dataset from segmenter.segment()

  • labels_name (str) – Name of cell labels variable (from config)

Raises:

ContractViolation – If any invariant is violated

Return type:

None

adapt.contracts.segmentation.check_seed_carry(ages)#

Carry age per cell label: a tuple of ints >= 0, index i for label i + 1.

0 means the cell was seeded independently this scan; k means it has been carried k consecutive scans from the previous frame’s cores.

Return type:

None

Parameters:

ages (tuple)

adapt.contracts.segmentation.check_segmented_ds(ds)#

Bound contract for the standard segmented dataset (cell_labels variable name fixed).

Return type:

None

Parameters:

ds (Dataset)

Projection stage contract.

Enforces that after optical flow computation, motion vectors and optional projection arrays are present and structurally valid.

adapt.contracts.projection.assert_projected(ds, max_steps=5)#

Enforce projection stage contract.

Parameters:

  • ds (Dataset) – Dataset from projector.project()

  • max_steps (int) – Maximum number of projection steps (default 5). If dataset has ‘max_projection_steps’ in attrs, that value is used instead.

Raises:

ContractViolation – If any invariant is violated

Return type:

None

adapt.contracts.projection.check_projected_ds(ds)#

Bound contract for the standard projected dataset.

Return type:

None

Parameters:

ds (Dataset)

Analysis stage contracts.

Enforces structural requirements on cell statistics and adjacency DataFrames. Scientific correctness of the values is the module’s responsibility, not checked here.

adapt.contracts.analysis.assert_analysis_output(df, min_expected_rows=0)#

Enforce analysis stage contract.

Parameters:

  • df (DataFrame) – Output from analyzer.extract()

  • min_expected_rows (int) – Minimum number of rows expected (default 0, allows no-cell frames)

Raises:

ContractViolation – If structural requirements are violated

Return type:

None

adapt.contracts.analysis.assert_cell_adjacency(df)#

Enforce cell adjacency contract.

Raises:

ContractViolation – If structural requirements are violated

Return type:

None

Parameters:

df (DataFrame)

adapt.contracts.analysis.check_cell_adjacency(df)#

Bound contract for cell adjacency DataFrame.

Return type:

None

Parameters:

df (DataFrame)

adapt.contracts.analysis.check_cell_stats(df)#

Bound contract for cell statistics DataFrame.

Return type:

None

Parameters:

df (DataFrame)

Tracking stage contracts.

Enforces structural requirements on tracked_cells and cell_events DataFrames.

adapt.contracts.tracking.assert_cell_events(df)#

Enforce cell events contract.

Raises:

ContractViolation – If structural requirements are violated

Return type:

None

Parameters:

df (DataFrame)

adapt.contracts.tracking.assert_tracked_cells(df)#

Enforce tracked cells contract.

Raises:

ContractViolation – If structural requirements are violated

Return type:

None

Parameters:

df (DataFrame)

adapt.contracts.tracking.check_cell_events(df)#

Bound contract for cell events DataFrame (skips validation on empty frame).

Return type:

None

Parameters:

df (DataFrame)

adapt.contracts.tracking.check_tracked_cells(df)#

Bound contract for tracked cells DataFrame (skips validation on empty frame).

Return type:

None

Parameters:

df (DataFrame)

Core contract enforcement primitives.

ContractViolation and require are the only two names every contract function depends on. This module has zero adapt imports so contracts can be imported from anywhere without creating import cycles.

exception adapt.contracts.pipeline.ContractViolation#

Bases: RuntimeError

Raised when a pipeline contract is violated.

This indicates a bug in pipeline logic, not bad user input or recoverable science edge cases. It means a pipeline stage did not produce the invariants it promised.

Key distinction: - ValueError: User/config error (handled by Pydantic) - ContractViolation: Pipeline bug (programmer error) - Exception: Recoverable science issues (try/except in algorithms)

adapt.contracts.pipeline.require(condition, message)#

Enforce a pipeline contract.

Fail-fast: no recovery, no fallback, no silence.

Raises:

ContractViolation – If condition is False.

Return type:

None

Parameters:

  • condition (bool)

  • message (str)

Time coordinate contract.

Enforces that datasets crossing module boundaries carry a normalized time coordinate — numpy datetime64, not cftime — so no module needs its own conversion logic. Modules that read raw radar data (e.g. ingest) are responsible for calling adapt.utils.time.normalize_time_scalar before returning a dataset to the context.

adapt.contracts.time.assert_time_normalized(ds)#

Enforce that the dataset time coordinate is numpy-compatible, not cftime.

Parameters:

ds (Dataset) – Dataset crossing a module boundary.

Raises:

ContractViolation – If the dataset has no time coordinate, or if the time coordinate uses cftime objects instead of numpy datetime64.

Return type:

None

adapt.contracts.time.check_time_normalized(ds)#

Bound contract for time normalization — use in module input_contracts.

Return type:

None

Parameters:

ds (Dataset)