Models Module

The models module provides the core infrastructure modeling components including dynamics, costs, budgets, hierarchy, and metadata models.

Base Models

Unified base model architecture for InfraLib.

class infralib.models.base.ModelContext(states: ndarray, actions: ndarray | None = None, time_step: int = 0, dynamics: Any = None, cost: Any = None, budget: Any = None, hierarchy: Any = None, metadata: Any = None, component_ids: ndarray | None = None, history: dict[str, list] | None = None, custom_data: dict[str, Any] | None = None)

Bases: object

Context object passed to all models containing state and dependencies.

states: ndarray

actions: ndarray | None = None

time_step: int = 0

dynamics: Any = None

cost: Any = None

budget: Any = None

hierarchy: Any = None

metadata: Any = None

component_ids: ndarray | None = None

history: dict[str, list] | None = None

custom_data: dict[str, Any] | None = None

get_model(model_type: str) → Any

Get a model by type name.

__init__(states: ndarray, actions: ndarray | None = None, time_step: int = 0, dynamics: Any = None, cost: Any = None, budget: Any = None, hierarchy: Any = None, metadata: Any = None, component_ids: ndarray | None = None, history: dict[str, list] | None = None, custom_data: dict[str, Any] | None = None) → None

class infralib.models.base.BaseModel(**params)

Bases: ABC

Unified base class for all infrastructure models.

__init__(**params)

Initialize with parameters and validate them.

abstractmethod compute(context: ModelContext) → Any

Main computation method - unified across all models.

Parameters:

context – ModelContext with current state and dependencies

Returns:

Model-specific output

abstractmethod reset(context: ModelContext | None = None)

Reset model to initial state.

Parameters:

context – Optional context for state-dependent resets

abstractmethod classmethod get_parameter_spec() → dict[str, tuple[type, tuple[float, float], str]]

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

classmethod get_required_models() → list[str]

List of required model dependencies.

Override to specify which other models this one needs. Returns list of model names: [‘dynamics’, ‘cost’, ‘hierarchy’, etc.]

validate_context(context: ModelContext)

Validate that context has required dependencies.

Dynamics Models

Deterioration dynamics models with unified interface.

class infralib.models.dynamics.DynamicsModel(**params)

Bases: BaseModel

Base class for deterioration dynamics with unified interface.

compute(context: ModelContext) → ndarray

Compute next states using context.

Parameters:

context – Must contain states and actions

Returns:

Array of next states

Return type:

next_states

step(states: ndarray, actions: ndarray) → ndarray

Legacy interface for backward compatibility.

class infralib.models.dynamics.WeibullDynamics(n_states: int = 10, shape: float = None, scale: float = None, shapes: list = None, scales: list = None, type_indices: ndarray = None, repair_effectiveness: float = 0.7, seed: int | None = None)

Bases: DynamicsModel

Weibull deterioration dynamics with validation.

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

__init__(n_states: int = 10, shape: float = None, scale: float = None, shapes: list = None, scales: list = None, type_indices: ndarray = None, repair_effectiveness: float = 0.7, seed: int | None = None)

Initialize with parameters and validate them.

reset(context: ModelContext | None = None)

Reset dynamics model (rebuild matrices if needed).

class infralib.models.dynamics.MarkovDynamics(n_states: int = 10, base_deterioration_rate: float = 0.1, repair_effectiveness: float = 0.7, seed: int | None = None)

Bases: DynamicsModel

Markov chain dynamics with custom transition matrices.

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

__init__(n_states: int = 10, base_deterioration_rate: float = 0.1, repair_effectiveness: float = 0.7, seed: int | None = None)

Initialize with parameters and validate them.

reset(context: ModelContext | None = None)

Reset dynamics model (rebuild matrices if needed).

class infralib.models.dynamics.FastWeibullDynamics(n_states: int = 10, shape: float = None, scale: float = None, shapes: list = None, scales: list = None, type_indices: ndarray = None, repair_effectiveness: float = 0.7, seed: int | None = None)

Bases: WeibullDynamics

Numba-optimized version of Weibull dynamics for large-scale simulation.

Cost Models

Cost models with unified interface.

class infralib.models.cost.CostModel(**params)

Bases: BaseModel

Base class for cost models with unified interface.

compute(context: ModelContext) → ndarray

Compute costs using context.

Parameters:

context – Must contain states and actions

Returns:

Array of costs per component

Return type:

costs

compute_legacy(states: ndarray, actions: ndarray, next_states: ndarray) → ndarray

Legacy interface for backward compatibility.

class infralib.models.cost.SimpleCost(inspect_cost: float = 10.0, repair_cost: float = 100.0, replace_cost: float = 1000.0, failure_penalty: float = 5000.0)

Bases: CostModel

Fixed costs per action type with validation.

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

__init__(inspect_cost: float = 10.0, repair_cost: float = 100.0, replace_cost: float = 1000.0, failure_penalty: float = 5000.0)

Initialize with parameters and validate them.

reset(context: ModelContext | None = None)

Reset cost model (nothing to reset for simple costs).

class infralib.models.cost.LengthBasedCost(inspect_cost_per_km: float = 50.0, repair_cost_per_km: float = 500.0, replace_cost_per_km: float = 5000.0, failure_penalty_per_km: float = 10000.0, component_lengths: ndarray = None)

Bases: CostModel

Costs scaled by component length (for road networks).

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

__init__(inspect_cost_per_km: float = 50.0, repair_cost_per_km: float = 500.0, replace_cost_per_km: float = 5000.0, failure_penalty_per_km: float = 10000.0, component_lengths: ndarray = None)

Initialize with parameters and validate them.

set_component_lengths(lengths: ndarray)

Set component lengths for cost calculation.

reset(context: ModelContext | None = None)

Reset cost model.

class infralib.models.cost.NonlinearCost(inspect_cost: float = 10.0, replacement_cost: float = 1000.0, cost_sensitivity: float = 2.0, min_repair_fraction: float = 0.2, failure_threshold: int = 1, failure_penalty: float = 5000.0, n_states: int = 10)

Bases: CostModel

Nonlinear cost model reflecting deterioration-dependent repair costs.

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

__init__(inspect_cost: float = 10.0, replacement_cost: float = 1000.0, cost_sensitivity: float = 2.0, min_repair_fraction: float = 0.2, failure_threshold: int = 1, failure_penalty: float = 5000.0, n_states: int = 10)

Initialize with parameters and validate them.

reset(context: ModelContext | None = None)

Reset cost model.

Budget Models

Budget models with unified interface.

class infralib.models.budget.BudgetModel(**params)

Bases: BaseModel

Base class for budget models with unified interface.

compute(context: ModelContext) → dict[str, Any]

Compute budget state and constraints.

Parameters:

context – Contains costs and state information

Returns:

Dict with ‘available’, ‘consumed’, ‘sufficient’ keys

apply_constraint(costs: ndarray) → tuple[ndarray, float]

Apply budget constraints to proposed costs.

Parameters:

costs – Proposed costs per component

Returns:

(allowed_mask, remaining_budget)

update(cost: float) → bool

Legacy interface for backward compatibility.

available() → float

Legacy interface for backward compatibility.

class infralib.models.budget.FixedBudget(initial_budget: float = 100000.0)

Bases: BudgetModel

Fixed budget model with validation.

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

__init__(initial_budget: float = 100000.0)

Initialize with parameters and validate them.

reset(context: ModelContext | None = None)

Reset budget to initial amount.

apply_constraint(costs: ndarray) → tuple[ndarray, float]

Apply budget constraints to costs.

class infralib.models.budget.CyclicBudget(cycle_budget: float = 50000.0, cycle_length: int = 30)

Bases: BudgetModel

Cyclic budget model with periodic allocations.

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

__init__(cycle_budget: float = 50000.0, cycle_length: int = 30)

Initialize with parameters and validate them.

reset(context: ModelContext | None = None)

Reset budget to initial state.

step_time()

Advance time and refresh budget if cycle complete.

apply_constraint(costs: ndarray) → tuple[ndarray, float]

Apply budget constraints to costs.

class infralib.models.budget.VariableCyclicBudget(cycle_budgets: list[float], cycle_lengths: list[int] = None)

Bases: BudgetModel

Variable cyclic budget with different amounts per cycle.

__init__(cycle_budgets: list[float], cycle_lengths: list[int] = None)

Initialize with parameters and validate them.

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

reset(context: ModelContext | None = None)

Reset budget to initial state.

step_time()

Advance time and refresh budget if cycle complete.

apply_constraint(costs: ndarray) → tuple[ndarray, float]

Apply budget constraints to costs.

class infralib.models.budget.EmergencyReserveBudget(normal_budget: float = 100000.0, emergency_reserve: float = 50000.0, emergency_threshold: int = 5)

Bases: BudgetModel

Budget with emergency reserve that can be accessed under certain conditions.

classmethod get_parameter_spec()

Get parameter specifications.

Returns:

Dict of param_name -> (type, (min, max), description)

__init__(normal_budget: float = 100000.0, emergency_reserve: float = 50000.0, emergency_threshold: int = 5)

Initialize with parameters and validate them.

reset(context: ModelContext | None = None)

Reset budget to initial state.

activate_emergency(failure_count: int)

Activate emergency reserve if failure threshold exceeded.

apply_constraint(costs: ndarray) → tuple[ndarray, float]

Apply budget constraints with emergency reserve.

Hierarchy Models

Hierarchy system with unified interface.

class infralib.models.hierarchy.HierarchyLevel(name: str, parent_level: str | None = None, properties: dict[str, ~typing.Any] = <factory>, aggregation_rules: dict[str, str] = <factory>)

Bases: object

Definition of a hierarchy level.

name: str

parent_level: str | None = None

properties: dict[str, Any]

aggregation_rules: dict[str, str]

__init__(name: str, parent_level: str | None = None, properties: dict[str, ~typing.Any] = <factory>, aggregation_rules: dict[str, str] = <factory>) → None

class infralib.models.hierarchy.HierarchyModel(**params)

Bases: BaseModel

Base class for hierarchy models with unified interface.

compute(context: ModelContext) → dict[str, Any]

Compute hierarchy-based metrics.

Parameters:

context – Contains states and component information

Returns:

Dict of hierarchy metrics and aggregations

get_hierarchy_levels() → list[HierarchyLevel]

Return ordered hierarchy levels from bottom to top.

get_component_group(component_id: int, level: str) → str | None

Get the group a component belongs to at a level.

get_group_components(group_id: str, level: str) → list[int]

Get all components in a group.

get_all_groups(level: str) → list[str]

Get all group IDs at a hierarchy level.

get_group_property(group_id: str, level: str, property: str) → Any

Get a property of a hierarchy group.

reset(context: ModelContext | None = None)

Reset hierarchy model.

classmethod get_parameter_spec() → dict[str, tuple[type, tuple[float, float], str]]

Hierarchy models typically don’t have numeric parameters.

class infralib.models.hierarchy.GeneralHierarchy(level_definitions: list[HierarchyLevel] | None = None)

Bases: HierarchyModel

General-purpose hierarchy for any domain.

__init__(level_definitions: list[HierarchyLevel] | None = None)

Create hierarchy with user-defined levels.

get_hierarchy_levels() → list[HierarchyLevel]

Return ordered hierarchy levels from bottom to top.

assign_component(component_id: int, assignments: dict[str, str])

Assign component to hierarchy groups.

get_component_group(component_id: int, level: str) → str | None

Get the group a component belongs to at a level.

get_group_components(group_id: str, level: str) → list[int]

Get all components in a group.

get_all_groups(level: str) → list[str]

Get all group IDs at a hierarchy level.

set_group_property(group_id: str, level: str, property: str, value: Any)

Set a property for a hierarchy group.

get_group_property(group_id: str, level: str, property: str) → Any

Get a property of a hierarchy group.

compute_group_metric(group_id: str, level: str, values: ndarray, aggregation: str = 'mean') → float

Compute aggregated metric for a group.

class infralib.models.hierarchy.SimpleHierarchy

Bases: GeneralHierarchy

Simple two-level hierarchy for basic applications.

__init__()

Create a simple component-system hierarchy.

class infralib.models.hierarchy.MultiLevelHierarchy(n_levels: int = 3)

Bases: GeneralHierarchy

Multi-level hierarchy for complex systems.

__init__(n_levels: int = 3)

Create a multi-level hierarchy.

Parameters:

n_levels – Number of hierarchy levels (including component level)

Metadata Models

Metadata system with unified interface.

class infralib.models.metadata.FieldDefinition(name: str, field_type: type, required: bool = False, default_value: Any = None, description: str = '')

Bases: object

Definition of a metadata field.

name: str

field_type: type

required: bool = False

default_value: Any = None

description: str = ''

__init__(name: str, field_type: type, required: bool = False, default_value: Any = None, description: str = '') → None

class infralib.models.metadata.MetadataModel(**params)

Bases: BaseModel

Base class for metadata models with unified interface.

compute(context: ModelContext) → dict[str, Any]

Compute metadata-based metrics.

Parameters:

context – Contains component information

Returns:

Dict of metadata metrics

get_field_definitions() → list[FieldDefinition]

Return list of field definitions for this metadata.

get_component_attribute(component_id: int, attribute: str) → Any

Get a specific attribute for a component.

get_bulk_attribute(component_ids: list[int] | ndarray, attribute: str) → ndarray

Get attribute values for multiple components efficiently.

query_components(**filters) → list[int]

Query components by attributes.

reset(context: ModelContext | None = None)

Reset metadata model.

classmethod get_parameter_spec() → dict[str, tuple[type, tuple[float, float], str]]

Metadata models typically don’t have numeric parameters.

class infralib.models.metadata.GeneralMetadata(field_definitions: list[FieldDefinition] | None = None)

Bases: MetadataModel

General-purpose metadata manager for any domain.

__init__(field_definitions: list[FieldDefinition] | None = None)

Create metadata manager with field definitions.

get_field_definitions() → list[FieldDefinition]

Return list of field definitions for this metadata.

add_component(component_id: int, metadata: dict[str, Any])

Add component with metadata.

get_component_attribute(component_id: int, attribute: str) → Any

Get a specific attribute for a component.

get_bulk_attribute(component_ids: list[int] | ndarray, attribute: str) → ndarray

Get attribute values for multiple components efficiently.

query_components(**filters) → list[int]

Query components by attributes.

set_component_attribute(component_id: int, attribute: str, value: Any)

Set an attribute for a component.

compute_weighted_metric(values: ndarray, weight_attribute: str) → float

Compute weighted metric using metadata attribute as weights.

class infralib.models.metadata.SimpleMetadata

Bases: GeneralMetadata

Simple metadata with common fields.

__init__()

Create metadata with common infrastructure fields.

class infralib.models.metadata.KeyValueMetadata

Bases: GeneralMetadata

Flexible key-value metadata storage.

__init__()

Create flexible metadata that accepts any fields.

add_component(component_id: int, metadata: dict[str, Any])

Add component with any metadata fields.

get_all_fields() → list[str]

Get all fields that have been used.