data_canon.core

Core data structures and utilities.

data_canon.core.dataclass

Data validation functions for travel survey data using Pydantic models.

CANONICAL_MODELS module-attribute

CANONICAL_MODELS: dict[str, type[BaseModel]] = {
    "households": survey_models.HouseholdModel,
    "persons": survey_models.PersonModel,
    "days": survey_models.PersonDayModel,
    "unlinked_trips": survey_models.UnlinkedTripModel,
    "linked_trips": survey_models.LinkedTripModel,
    "tours": survey_models.TourModel,
    "joint_trips": survey_models.JointTripModel,
    "joint_tours": survey_models.JointTourModel,
    "habitual_locations": survey_models.HabitualLocationModel,
    "habitual_location_days": survey_models.HabitualLocationDayModel,
    "household_weights": weighting_models.HouseholdWeightingModel,
}

CanonicalData dataclass

Canonical data structure for travel survey data with validation.

Use the validate() method to validate specific tables.

households class-attribute instance-attribute

households: pl.DataFrame | None = None

persons class-attribute instance-attribute

persons: pl.DataFrame | None = None

days class-attribute instance-attribute

days: pl.DataFrame | None = None

unlinked_trips class-attribute instance-attribute

unlinked_trips: pl.DataFrame | None = None

linked_trips class-attribute instance-attribute

linked_trips: pl.DataFrame | None = None

tours class-attribute instance-attribute

tours: pl.DataFrame | None = None

joint_trips class-attribute instance-attribute

joint_trips: pl.DataFrame | None = None

joint_tours class-attribute instance-attribute

joint_tours: pl.DataFrame | None = None

habitual_locations class-attribute instance-attribute

habitual_locations: pl.DataFrame | None = None

habitual_location_days class-attribute instance-attribute

habitual_location_days: pl.DataFrame | None = None

household_weights class-attribute instance-attribute

household_weights: pl.DataFrame | None = None

models class-attribute instance-attribute

models: dict[str, type[BaseModel]] = field(
    default_factory=lambda: dict(CANONICAL_MODELS)
)

generated_columns class-attribute instance-attribute

generated_columns: dict[str, dict[str, GeneratedColumn]] = field(
    default_factory=dict, repr=False
)

custom_validators class-attribute instance-attribute

custom_validators: dict[str, list[Callable]] = field(
    default_factory=lambda: {
        table: (list(validators))
        for table, validators in (CUSTOM_VALIDATORS.items())
    }
)

TABLE_NAMES class-attribute instance-attribute

TABLE_NAMES: list[str] = field(
    default_factory=lambda: [
        "households",
        "persons",
        "days",
        "unlinked_trips",
        "linked_trips",
        "joint_trips",
        "tours",
        "joint_tours",
        "habitual_locations",
        "habitual_location_days",
        "household_weights",
    ],
    repr=False,
)

register_generated_columns

register_generated_columns(
    table: str,
    columns: Iterable[str] | Mapping[str, str | GeneratedColumn],
) -> None

Record columns a step generated whose names come from configuration.

add_zone_ids emits {prefix}_{zone_name}, the imputation step emits {column}_preimputed, and cascade_completeness emits one column per usability profile, so the names differ per project and cannot be declared as model fields. They are still ours -- computed, named in config, and wanted in the delivered output -- so a step that creates them says so here rather than leaving a consumer to guess from the name.

Pass a mapping to describe them as well. A model field carries its description to the codebook; a generated column has nowhere else to say what it means, so a column whose name a project chose can still explain itself. Passing a bare iterable records the names with no description.

A mapping may give a GeneratedColumn instead of a string, to declare the constraints a model field would have carried. Without that a generated column is delivered unchecked.

Parameters:

Name	Type	Description	Default
table	str	Canonical table the columns were added to.	required
columns	Iterable[str] | Mapping[str, str | GeneratedColumn]	Column names, or column name -> description or GeneratedColumn.	required

public_columns

public_columns(table: str) -> set[str]

Columns of table the pipeline stands behind: declared or generated.

describe_generated

describe_generated(table: str) -> dict[str, str]

Descriptions for table's generated columns, empty string where none.

as_dict

as_dict() -> dict[str, pl.DataFrame | None]

Return all canonical tables as a dict (including None entries).

as_dict_non_null

as_dict_non_null() -> dict[str, pl.DataFrame]

Return only non-None canonical tables as a dict.

add_models

add_models(new_models: dict[str, type[BaseModel]]) -> None

Add or override data models for validation.

Also adds the new tables as canonical data attributes.

Parameters:

Name	Type	Description	Default
new_models	dict[str, type[BaseModel]]	Dictionary of table name to Pydantic model class	required

validate

validate(table_name: str, step: str | None = None) -> None

Validate a table through all validation layers.

Runs validation in this order: 1. Column constraints (uniqueness) 2. Foreign key constraints 3. Row-level Pydantic validation (step-aware if step provided) 4. Custom user-registered validators

Parameters:

Name	Type	Description	Default
table_name	str	Name of the table to validate	required
step	str | None	Pipeline step name for step-aware validation. If None, validates all fields strictly.	None

Raises:

Type	Description
DataValidationError	If any validation check fails

register_validator

register_validator(*table_names: str) -> Callable

Register a custom validator on one or more tables.

Parameters:

Name	Type	Description	Default
*table_names	str	One or more table names to register validator on	()

Returns:

Type	Description
Callable	Decorator function

Example

@data.register_validator("tours") def check_tours(tours: pl.DataFrame) -> list[str]: errors = [] # Check logic return errors

@data.register_validator("tours", "linked_trips") def check_consistency( tours: pl.DataFrame, linked_trips: pl.DataFrame ) -> list[str]: # Multi-table check return []

data_canon.core.exceptions

Validation exceptions for canonical survey data.

DataValidationError dataclass

Structured validation error with context.

Attributes:

Name	Type	Description
table	str	Name of the table being validated
rule	str	Name of the validation rule that failed
message	str	Human-readable error description
row_id	int | None	Optional row identifier for row-level errors
column	str | None	Optional column name for column-level errors

table instance-attribute

table: str

rule instance-attribute

rule: str

message instance-attribute

message: str

row_id class-attribute instance-attribute

row_id: int | None = None

column class-attribute instance-attribute

column: str | None = None

data_canon.core.labeled_enum

Base class for labeled enumerations with values and human-readable labels.

This module provides the LabeledEnum base class for creating enumerations that store both integer values and descriptive labels, with support for canonical field names for pydantic model mapping.

LabeledEnumMeta

Metaclass for LabeledEnum that reserves canonical_field_name.

canonical_field_name class-attribute instance-attribute

canonical_field_name: str | None = None

field_description class-attribute instance-attribute

field_description: str | None = None

LabeledEnum

Base class for enumerations with values and labels.

Each enum member is defined as a tuple of (value, label): MEMBER_NAME = (1, "Descriptive Label") or MEMBER_NAME = ("some_string", "Descriptive Label")

The enum provides: - value: The integer or string value - label: The human-readable label - field_name: The canonical field name - description: The field description

Class Methods: - from_value(val): Look up an enum member by its value - from_label(label): Look up an enum member by its label - get_field_name(): Get the canonical field name for the enum - get_description(): Get the field description for the enum

Example

class Gender(LabeledEnum): canonical_field_name = "gender" field_description = "Respondent's gender identity"

MALE = (1, "Male")
FEMALE = (2, "Female")

member = Gender.MALE print(member.value) # 1 print(member.label) # "Male" print(member.field_name) # "gender" print(member.description) # "Respondent's gender identity"

found = Gender.from_value(1) # Returns Gender.MALE found = Gender.from_label("Female") # Returns Gender.FEMALE

label property

label: str

Get the human-readable label for this enum member.

field_name property

field_name: str | None

Get the canonical field name for this enum.

Returns the canonical_field_name class attribute if defined, otherwise returns None.

description property

description: str | None

Get the field description for this enum.

Returns the field_description class attribute if defined, otherwise returns None.

from_value classmethod

from_value(value: int | str, strict: bool = True) -> LabeledEnum

from_value(
    value: int | str, strict: bool = False
) -> Optional[LabeledEnum]

from_value(
    value: int | str, strict: bool = True
) -> Optional[LabeledEnum]

Look up an enum member by its value.

Parameters:

Name	Type	Description	Default
value	int | str	The integer or string value to search for	required
strict	bool	If True (default), raise ValueError if not found. If False, return None if not found.	True

Returns:

Type	Description
Optional[LabeledEnum]	The enum member with the matching value, or None if not found and strict=False

Raises:

Type	Description
ValueError	If value not found and strict=True

from_label classmethod

from_label(
    label: str, strict: bool = True
) -> Optional[LabeledEnum]

Look up an enum member by its label.

Parameters:

Name	Type	Description	Default
label	str	The label to search for (case-sensitive)	required
strict	bool	If True (default), raise ValueError if not found. If False, return None if not found.	True

Returns:

Type	Description
Optional[LabeledEnum]	The enum member with the matching label, or None if not found

get_field_name classmethod

get_field_name() -> str | None

Get the canonical field name for this enum class.

Returns:

Type	Description
str | None	The canonical_field_name class attribute if defined, otherwise None

get_description classmethod

get_description() -> str | None

Get the field description for this enum class.

Returns:

Type	Description
str | None	The field_description class attribute if defined, otherwise None

to_dict classmethod

to_dict() -> dict[int | str, str]

Get a dictionary mapping enum values to labels.

Returns:

Type	Description
dict[int | str, str]	A dictionary where keys are enum values and values are labels

data_canon.core.schema_field

Module for creating Pydantic fields with relational schema metadata.

schema_field

schema_field(
    *,
    unique: bool = False,
    fk_to: str | None = None,
    required_child: bool = False,
    required_child_when: str | None = None,
    **field_kwargs: Any
) -> Any

Create a Pydantic Field with relational schema metadata.

This is a convenience wrapper that attaches relational metadata (uniqueness, foreign keys, required children) to Pydantic fields via json_schema_extra. Workflow metadata (which steps require which fields) is managed separately by the step registry.

Parameters:

Name	Type	Description	Default
unique	bool	Whether this field should be unique across records. Used to validate uniqueness constraints at the table level.	False
fk_to	str | None	Foreign key reference in format "parent_table.parent_column". The referenced parent column must be marked as unique.	None
required_child	bool	If True, marks this FK as requiring the parent record to have at least one child record (bidirectional FK constraint). Only valid when fk_to is specified.	False
required_child_when	str | None	Name of a boolean column on the parent table. Only parent rows where it is true need a child; rows where it is false are legitimately childless. A missing column or a null value still needs a child, so the constraint never weakens silently. Only valid with required_child=True.	None
**field_kwargs	Any	All other Field parameters (ge, le, default, etc.)	{}

Returns:

Type	Description
Any	Field instance with schema metadata attached

Example

Unique primary key

person_id: int = schema_field(ge=1, unique=True)

Foreign key with required child constraint

hh_id: int = schema_field( ... ge=1, ... fk_to="households.hh_id", ... required_child=True, ... )

Self-referential foreign key

parent_tour_id: int | None = schema_field( ... ge=1, ... fk_to="tours.tour_id", ... default=None, ... )

data_canon.core.validators

Re-exports from :mod:pipeline.step_registry for backward compatibility.

All step-aware validation metadata now lives in the step registry. Prefer importing directly from pipeline.step_registry.