data_canon.validation

Validation utilities for data quality checks.

data_canon.validation.column

Column-level validation functions for canonical survey data.

Two kinds of column are checked here. A declared field carries its constraints in the model, and pydantic enforces them row by row. A generated column cannot: its name comes from configuration, so there is no field to hang them on. Without something here a generated column is delivered unchecked -- the per-profile weight columns would accept a negative weight in silence.

So a generated column may declare the same constraints a field would, and they are checked over the column rather than per row.

GeneratedColumn dataclass

What a step promises about a column it named from configuration.

Attributes:

Name	Type	Description
description	str	What the column means, for the delivered codebook. A generated column has nowhere else to say it.
ge	float | None	Lower bound, inclusive, if the values are numeric and bounded. The equivalent of schema_field(ge=...) on a declared field.

description class-attribute instance-attribute

description: str = ''

ge class-attribute instance-attribute

ge: float | None = None

check_generated_constraints

check_generated_constraints(
    table_name: str,
    df: pl.DataFrame,
    specs: dict[str, GeneratedColumn],
) -> None

Check the constraints generated columns declared, or raise.

Only columns actually present are checked: a step registers what it can produce, and a run that produced fewer is not thereby invalid.

Parameters:

Name	Type	Description	Default
table_name	str	Name of the table being validated.	required
df	pl.DataFrame	DataFrame to validate.	required
specs	dict[str, GeneratedColumn]	Column name -> what the generating step promised.	required

Raises:

Type	Description
DataValidationError	If a column violates its declared bound.

get_unique_fields

get_unique_fields(model: type[BaseModel]) -> list[str]

Get list of fields marked as unique in the model.

Parameters:

Name	Type	Description	Default
model	type[BaseModel]	Pydantic model class	required

Returns:

Type	Description
list[str]	List of field names marked as unique

check_unique_constraints

check_unique_constraints(
    table_name: str, df: pl.DataFrame, unique_columns: list[str]
) -> None

Check uniqueness constraints on specified columns.

Parameters:

Name	Type	Description	Default
table_name	str	Name of the table being validated	required
df	pl.DataFrame	DataFrame to validate	required
unique_columns	list[str]	List of column names that must be unique	required

Raises:

Type	Description
DataValidationError	If uniqueness constraint is violated

data_canon.validation.custom

Custom validation checks for travel survey data.

This module contains DataFrame-level validation checks that run during the custom validator phase (after row-level validation). Users can add checks here and register them to tables using CUSTOM_VALIDATORS.

These checks are for more complex validation logic that requires multiple tables or spanning multiple rows. Most checks can be done using built-in validators, or at the row-level.

To add a new check: 1. Define a function that takes one or more DataFrames and returns list[str] 2. Add it to CUSTOM_VALIDATORS dict below, mapping table name to check functions 3. The check will automatically run when that table is validated

CUSTOM_VALIDATORS module-attribute

CUSTOM_VALIDATORS: dict[str, list[Callable]] = {
    "households": [],
    "persons": [],
    "days": [],
    "unlinked_trips": [],
    "linked_trips": [],
    "tours": [],
    "habitual_locations": [],
    "habitual_location_days": [],
}

check_for_teleports

check_for_teleports(unlinked_trips: pl.DataFrame) -> list[str]

Check for when trip destination is too far from next trip origin.

check_trip_count_matches_quality

check_trip_count_matches_quality(
    tours: pl.DataFrame, linked_trips: pl.DataFrame
) -> list[str]

Verify trip_count matches the trips, and NO_DESTINATION means what it says.

trip_count is checked against the linked trips rather than against another tour column: both are derived in the extractor, so they would agree even if the extractor were wrong. Only the trips themselves are independent.

NO_DESTINATION is checked in both directions against its actual definition: a closed tour that aggregation found no purpose for. A partial tour may also lack a purpose -- half of it went unobserved -- and is graded on its open end instead, so a null purpose alone does not imply the code. A closed tour without a purpose may also be OTHER_HOME: a move between two of the person's homes, where the move is the reason there is no stop.

Parameters:

Name	Type	Description	Default
tours	pl.DataFrame	Tour records with trip_count, tour_data_quality and tour_purpose	required
linked_trips	pl.DataFrame	Trip records to count per tour	required

Returns:

Type	Description
list[str]	List of error messages (empty if validation passes)

check_valid_tours_are_complete

check_valid_tours_are_complete(tours: pl.DataFrame) -> list[str]

Verify tours flagged VALID are genuinely complete.

A tour marked tour_data_quality == VALID must not be a single-trip tour and must have a non-null tour_purpose. This guards the downstream formatters, which keep only VALID tours: a tour mislabeled VALID but missing its purpose would survive that filter and leak into the model output (in CT-RAMP, silently coerced to the OTHDISCR catch-all purpose).

TourModel.validate_complete_tours enforces the converse (non-single-trip tours have a purpose); this check keys on tour_data_quality instead, so the two are complementary rather than redundant.

Parameters:

Name	Type	Description	Default
tours	pl.DataFrame	Tour records with tour_data_quality, trip_count, tour_purpose	required

Returns:

Type	Description
list[str]	List of error messages (empty if validation passes)

check_trip_spatial_continuity

check_trip_spatial_continuity(
    linked_trips: pl.DataFrame,
) -> list[str]

Report spatially discontinuous trip junctions (missing legs / teleports).

Within a person-day, one trip's destination should be at (or near) the next trip's origin. A larger jump means the diary skipped a connecting leg. The tour extractor already flags the affected tour SPATIAL_GAP so formatters drop it; this is the upstream tripwire.

Discontinuities are inherent survey noise (~1% of junctions), so this check does NOT fail on their mere presence - it logs the observed count and rate, and only raises when the rate exceeds _SPATIAL_GAP_MAX_RATE, which signals a systemic problem (e.g. a swapped/rescaled coordinate column) rather than ordinary respondent gaps.

Parameters:

Name	Type	Description	Default
linked_trips	pl.DataFrame	Linked trips with person_id, day_id, depart_time and o/d coordinates	required

Returns:

Type	Description
list[str]	List of error messages (empty unless the discontinuity rate is
list[str]	implausibly high)

check_reported_locations_match_coordinates

check_reported_locations_match_coordinates(
    habitual_locations: pl.DataFrame,
    households: pl.DataFrame,
    persons: pl.DataFrame,
) -> list[str]

Check the reported primary locations and the coordinate columns say the same.

Each reported primary home, workplace and school repeats a pair of coordinate columns (home_lat/lon on households, work_*/school_* on persons) that other steps still read. Every such pair must have its row, and every row its pair, within _SAME_PLACE_DEGREES, so the two can never drift apart.

Parameters:

Name	Type	Description	Default
habitual_locations	pl.DataFrame	Habitual locations.	required
households	pl.DataFrame	Households with hh_id and the home coordinates.	required
persons	pl.DataFrame	Persons with person_id, hh_id and work/school coordinates.	required

Returns:

Type	Description
list[str]	One message per location type that disagrees.

data_canon.validation.relational

Relational validation functions for canonical survey data.

This module provides validation for foreign key constraints and bidirectional relationships (required children).

get_foreign_key_fields

get_foreign_key_fields(
    model: type[BaseModel],
) -> dict[str, tuple[str, str]]

Extract foreign key relationships from model metadata.

Parameters:

Name	Type	Description	Default
model	type[BaseModel]	Pydantic model class	required

Returns:

Name	Type	Description
	dict[str, tuple[str, str]]	Dict mapping child field name to (parent_table, parent_column)
Example	dict[str, tuple[str, str]]	{"hh_id": ("households", "hh_id")}

get_required_children_fields

get_required_children_fields(
    model: type[BaseModel],
) -> dict[str, tuple[str, str, str | None]]

Extract required_child FK relationships from model metadata.

Parameters:

Name	Type	Description	Default
model	type[BaseModel]	Pydantic model class	required

Returns:

Type	Description
dict[str, tuple[str, str, str | None]]	Dict mapping child field name to (parent_table, parent_column,
dict[str, tuple[str, str, str | None]]	required_child_when) for fields that require bidirectional FK
dict[str, tuple[str, str, str | None]]	constraint. required_child_when is the name of a boolean column on
dict[str, tuple[str, str, str | None]]	the parent table -- only rows where it is true need a child -- or None
dict[str, tuple[str, str, str | None]]	when every parent row needs one.

foreign_key_edges

foreign_key_edges(
    models: dict[str, type[BaseModel]],
) -> tuple[tuple[str, str, str, str, bool], ...]

Every declared reference, with whether losing its target is survivable.

Returns (child_table, child_column, parent_table, parent_column, optional) per declared foreign key. optional comes from the field's own nullability, which already carries the distinction a cascade needs: a required reference means the child cannot exist without its parent, while an optional one means the child stands on its own and the reference is merely void.

So persons.hh_id is required -- a person with no household is not a record anyone can use -- while linked_trips.joint_trip_id is optional, because a trip that stops being joint is still a trip.

Deriving the list rather than writing it down is the point. There are more than twenty of these, they are denormalised across the trip tables, and any hand-maintained subset silently omits the one that matters next.

Parameters:

Name	Type	Description	Default
models	dict[str, type[BaseModel]]	Table name to model, as CanonicalData.models holds them.	required

Returns:

Type	Description
tuple[tuple[str, str, str, str, bool], ...]	One tuple per declared foreign key, in table then field order.

validate_fk_references

validate_fk_references(models: dict[str, type[BaseModel]]) -> None

Validate that all FK references point to unique fields.

Parameters:

Name	Type	Description	Default
models	dict[str, type[BaseModel]]	Dict mapping table names to Pydantic model classes	required

Raises:

Type	Description
ValueError	If FK references non-unique field or missing table/field

check_foreign_keys

check_foreign_keys(
    table_name: str,
    df: pl.DataFrame,
    fk_fields: dict[str, tuple[str, str]],
    get_table_func: Callable,
) -> None

Check foreign key constraints using FK metadata from models.

Parameters:

Name	Type	Description	Default
table_name	str	Name of the table being validated	required
df	pl.DataFrame	DataFrame to validate (child table)	required
fk_fields	dict[str, tuple[str, str]]	Dict mapping child FK field to (parent_table, parent_col) from get_foreign_key_fields()	required
get_table_func	Callable	Function to retrieve other tables by name	required

Raises:

Type	Description
DataValidationError	If foreign key constraint is violated

data_canon.validation.row

Row-level validation framework for canonical survey data.

This module provides step-aware row validation using Pydantic models. Step-aware validation allows fields to be required only in specific pipeline steps, enabling progressive data refinement throughout the pipeline.

Note: ValidationError class is defined in data_canon.core.dataclass to avoid circular imports.

validate_row_for_step

validate_row_for_step(
    row_dict: dict[str, Any],
    model: type[BaseModel],
    table_name: str,
    step_name: str | None = None,
) -> None

Validate a single row for a specific pipeline step.

This function validates that all fields required for the given step are present and valid. Fields not required for this step are still type-checked if present, but are not required.

Parameters:

Name	Type	Description	Default
row_dict	dict[str, Any]	Dictionary representing a single row	required
model	type[BaseModel]	Pydantic model class to validate against	required
table_name	str	Canonical table name (e.g. "unlinked_trips")	required
step_name	str | None	Name of the pipeline step. If None, validates all fields.	None

Raises:

Type	Description
PydanticValidationError	If validation fails
ValueError	If required fields are missing

validate_dataframe_rows

validate_dataframe_rows(
    table_name: str,
    df: pl.DataFrame,
    model: type[BaseModel],
    step: str | None = None,
) -> None

Validate all rows in a DataFrame using step-aware validation.

Parameters:

Name	Type	Description	Default
table_name	str	Name of the table being validated (for error messages)	required
df	pl.DataFrame	DataFrame to validate	required
model	type[BaseModel]	Pydantic model class for row validation	required
step	str | None	Pipeline step name for step-aware validation. If None, validates all fields strictly.	None

Raises:

Type	Description
DataValidationError	If any row fails validation