Configuration

class ireiat.config.data_pipeline.DataPipelineConfig(*, ops: dict = None)

Follows Dagster’s specific configuration requirements for asset-specific configs

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'ops': FieldInfo(annotation=dict, required=False, default_factory=default_asset_mapping, description='List of assets and associated configuration')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.FAF5CountyConfig(*, faf_demand_field: str = 'tons_2022', county_to_county_ktons_threshold: float = 0.1)

Excludes county->county flows below a threshold

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'county_to_county_ktons_threshold': FieldInfo(annotation=float, required=False, default=0.1, description='Annual kton threshold to be included in county->county flows. Anything smaller is excluded.'), 'faf_demand_field': FieldInfo(annotation=str, required=False, default='tons_2022', description='The field in FAF used to aggregate demand ( in tons)')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.FAF5DemandConfig(*, faf_demand_field: str = 'tons_2022', unknown_mode_percent: float = 0.3)

Used to group and allocate unknown modes to this mode

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'faf_demand_field': FieldInfo(annotation=str, required=False, default='tons_2022', description='The field in FAF used to aggregate demand ( in tons)'), 'unknown_mode_percent': FieldInfo(annotation=float, required=False, default=0.3, description='Percentage of mixed mode FAF demand to allocate to the given modal network')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.FAF5FilterConfig(*, faf_demand_field: str = 'tons_2022', faf_commodities: list[FAFCommodity] = None)

Allocates whether FAF commodities are considered as part of the demand. See FAF SCTG2 codes. https://www.bts.gov/sites/bts.dot.gov/files/2021-02/FAF5-User-Guide.pdf

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'faf_commodities': FieldInfo(annotation=list[FAFCommodity], required=False, default_factory=default_faf_commodity_factory), 'faf_demand_field': FieldInfo(annotation=str, required=False, default='tons_2022', description='The field in FAF used to aggregate demand ( in tons)')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.FAF5MasterConfig(*, faf_demand_field: str = 'tons_2022')

Parent class to allow us to inject the FAF demand field into child classes

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'faf_demand_field': FieldInfo(annotation=str, required=False, default='tons_2022', description='The field in FAF used to aggregate demand ( in tons)')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.FAFCommodity(*, name: str, sctg2: str, containerizable: bool = False, percentage_containerizable: float = 0.0)

Single FAF Commodity

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'containerizable': FieldInfo(annotation=bool, required=False, default=False), 'name': FieldInfo(annotation=str, required=True), 'percentage_containerizable': FieldInfo(annotation=float, required=False, default=0.0, metadata=[Ge(ge=0), Le(le=1)]), 'sctg2': FieldInfo(annotation=str, required=True, metadata=[MinLen(min_length=2), MaxLen(max_length=2)])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.GeographicImpedance(*, jr260: str, latitude: float, longitude: float, class_1_to_class_1_impedance: int, default_impedance: int = 275, radius_miles: int = 20, overrides: list[RailToRailImpedanceOverride] = None)

An override for impedance values within a geographic area. All impedance junctions with nodes that fall within radius_miles of a given lat/long will use this object’s class_1_to_class_1_impedance and this configuration’s default impedance for anything else. If overrides is specified, the specific rail-to-rail impedances will be used

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'class_1_to_class_1_impedance': FieldInfo(annotation=int, required=True), 'default_impedance': FieldInfo(annotation=int, required=False, default=275, description='Default impedance in this geographic area'), 'jr260': FieldInfo(annotation=str, required=True), 'latitude': FieldInfo(annotation=float, required=True), 'longitude': FieldInfo(annotation=float, required=True), 'overrides': FieldInfo(annotation=list[RailToRailImpedanceOverride], required=False, default_factory=list), 'radius_miles': FieldInfo(annotation=int, required=False, default=20)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.RailImpedanceConfig(*, class_1_rr_codes: list = None, default_impedance: int = 275, class_1_to_class_1_impedance: int = 750, geographic_overrides: list[GeographicImpedance] = None)

Used to specify impedance parameters on the rail graph in equivalent miles

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'class_1_rr_codes': FieldInfo(annotation=list, required=False, default_factory=_generate_class_1_rr_codes), 'class_1_to_class_1_impedance': FieldInfo(annotation=int, required=False, default=750, description='Default impedance to use when switching between Class 1 RRs'), 'default_impedance': FieldInfo(annotation=int, required=False, default=275, description='Default impedance to use when no match is found'), 'geographic_overrides': FieldInfo(annotation=list[GeographicImpedance], required=False, default_factory=list)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.RailToRailImpedanceOverride(*, from_owner: str, to_owner: str, impedance: int)

A class to specify individual rail owner impedances, currently only used within the GeographicImpedance class

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'from_owner': FieldInfo(annotation=str, required=True), 'impedance': FieldInfo(annotation=int, required=True), 'to_owner': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.TAPFilterTonsConfig(*, quantile_threshold: float | None = 0.99)

Used to specify an optional quantile threshold to filter county|county tons for the mode

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'quantile_threshold': FieldInfo(annotation=Union[float, NoneType], required=False, default=0.99, description='If specified, only includes a county|county highway tonnage above the quantile value')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.TAPNetworkConfig(*, default_speed_mph: int = 20, default_capacity_ktons: int = 100000, default_network_alpha: float = 0.1, default_network_beta: float = 2.0)

Used to specify default configuration parameters

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'default_capacity_ktons': FieldInfo(annotation=int, required=False, default=100000, description='Default capacity (in kilotons) to use for network edges without a custom capacity'), 'default_network_alpha': FieldInfo(annotation=float, required=False, default=0.1, description='Scalar, alpha, in TAP congestion term (alpha * (flow/capacity) ^ beta)'), 'default_network_beta': FieldInfo(annotation=float, required=False, default=2.0, description='Exponent, beta, in TAP congestion term (alpha * (flow/capacity) ^ beta)'), 'default_speed_mph': FieldInfo(annotation=int, required=False, default=20, description='Default speed to use for (non-dray) rail links in the rail TAP network')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class ireiat.config.data_pipeline.TAPRailConfig(*, default_speed_mph: int = 20, default_capacity_ktons: int = 100000, default_network_alpha: float = 0.1, default_network_beta: float = 2.0, dray_default_speed_mph: int = 50, dray_penalty_factor: int = 100, capacity_ktons_per_track: int = 1000, capacity_ktons_default: int = 100000, intermodal_search_radius_miles: int = 250, intermodal_facility_capacity_ktons: int = 7500, intermodal_edge_alpha: float = 1, intermodal_edge_beta: float = 1)

Adjustable configuration parameters for running the rail network

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True, 'ignored_types': (<class 'functools.cached_property'>,)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'capacity_ktons_default': FieldInfo(annotation=int, required=False, default=100000, description='Default capacity (in kilotons) to use for non-IM capacity and non-rail network (i.e. dray, impedance, etc.)'), 'capacity_ktons_per_track': FieldInfo(annotation=int, required=False, default=1000, description='Per track capacity (in kilotons) used in the rail network for NARN edges'), 'default_capacity_ktons': FieldInfo(annotation=int, required=False, default=100000, description='Default capacity (in kilotons) to use for network edges without a custom capacity'), 'default_network_alpha': FieldInfo(annotation=float, required=False, default=0.1, description='Scalar, alpha, in TAP congestion term (alpha * (flow/capacity) ^ beta)'), 'default_network_beta': FieldInfo(annotation=float, required=False, default=2.0, description='Exponent, beta, in TAP congestion term (alpha * (flow/capacity) ^ beta)'), 'default_speed_mph': FieldInfo(annotation=int, required=False, default=20, description='Default speed to use for (non-dray) rail links in the rail TAP network'), 'dray_default_speed_mph': FieldInfo(annotation=int, required=False, default=50, description='Default speed for drayage legs (for road segments) in the rail TAP'), 'dray_penalty_factor': FieldInfo(annotation=int, required=False, default=100, description='Factor to increase FFT on drayage legs to avoid exclusive use of the road network for the rail TAP'), 'intermodal_edge_alpha': FieldInfo(annotation=float, required=False, default=1, description='Scalar, alpha, in TAP congestion term (alpha * (flow/capacity) ^ beta) for IM capacity links'), 'intermodal_edge_beta': FieldInfo(annotation=float, required=False, default=1, description='Exponent, beta, in TAP congestion term (alpha * (flow/capacity) ^ beta) for IM capacity links'), 'intermodal_facility_capacity_ktons': FieldInfo(annotation=int, required=False, default=7500, description='Default capacity (in kiltons) to use for IM facility inlet/outlet edges'), 'intermodal_search_radius_miles': FieldInfo(annotation=int, required=False, default=250, description='Distance to search from an origin/destination centroid to an intermodal facility')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.