schemas

ScenarioSchemaAir

Bases: AbstractScenarioSchema

Air scenario schema extending the base scenario schema.

This schema currently does not add extra checks beyond AbstractScenarioSchema, but exists to allow air-specific evolution in the future.

ScenarioSchemaSoil

Bases: AbstractScenarioSchema

Soil scenario schema extending the base scenario with required soil properties.

check_required_soil_columns classmethod

check_required_soil_columns(df: DataFrame)

Ensure soil-specific required columns are present.

Source code in cable_thermal_model/model/schemas/model_input_schemas.py

@pa.dataframe_check(error="Scenario columns must include soil_thermal_resistivity and soil_thermal_capacity.")
@classmethod
def check_required_soil_columns(cls, df: pd.DataFrame):
    """Ensure soil-specific required columns are present."""
    req_columns = {"soil_thermal_resistivity", "soil_thermal_capacity"}
    if not all(col in df.columns for col in req_columns):
        raise ValueError(f"Scenario dataframe must include columns: {req_columns}.")
    return True

check_numeric_soil_columns classmethod

check_numeric_soil_columns(df: DataFrame)

Ensure soil thermal columns contain numeric values.

Source code in cable_thermal_model/model/schemas/model_input_schemas.py

@pa.dataframe_check(error="Soil thermal columns must contain numeric values.")
@classmethod
def check_numeric_soil_columns(cls, df: pd.DataFrame):
    """Ensure soil thermal columns contain numeric values."""
    numeric_columns = ["soil_thermal_resistivity", "soil_thermal_capacity"]
    for col in numeric_columns:
        if not pd.api.types.is_numeric_dtype(df[col]):
            raise ValueError(f"Column {col} must contain numeric values.")
    return True

ModelOutputSchema

Bases: BaseModel, Generic[StateT]

Schema for the output of the thermal cable model, containing the temperature results and the final state.

TemperatureResultSchema

Bases: DataFrameModel

Schema for temperature result DataFrame with MultiIndex columns.

Structure: - Index: datetime (time series) - Columns: MultiIndex with 3 levels: - Level 0: circuit_name (str) - Level 1: cable_position (CablePosition enum values) - Level 2: cable_layer (CableLayer enum values) - Values: temperature in degrees Celsius (float)

check_datetime_index classmethod

check_datetime_index(df: DataFrame)

Ensure index is datetime-like.

Source code in cable_thermal_model/model/schemas/model_output_schemas.py

@pa.dataframe_check(error="Temperature result index must be datetime-like.")
@classmethod
def check_datetime_index(cls, df: pd.DataFrame):
    """Ensure index is datetime-like."""
    return pd.api.types.is_datetime64_any_dtype(df.index) or pd.api.types.is_timedelta64_dtype(df.index)

check_multiindex_columns classmethod

check_multiindex_columns(df: DataFrame)

Ensure columns are a MultiIndex with 3 levels.

Source code in cable_thermal_model/model/schemas/model_output_schemas.py

@pa.dataframe_check(
    error="Temperature result columns must be a 3-level MultiIndex: (circuit_name, cable_position, cable_layer)."
)
@classmethod
def check_multiindex_columns(cls, df: pd.DataFrame):
    """Ensure columns are a MultiIndex with 3 levels."""
    expected_nlevels = 3
    return isinstance(df.columns, pd.MultiIndex) and df.columns.nlevels == expected_nlevels

check_circuit_names classmethod

check_circuit_names(df: DataFrame) -> bool

Ensure circuit names are non-empty strings.

Source code in cable_thermal_model/model/schemas/model_output_schemas.py

@pa.dataframe_check
@classmethod
def check_circuit_names(cls, df: pd.DataFrame) -> bool:
    """Ensure circuit names are non-empty strings."""
    circuit_names = df.columns.get_level_values(0).unique()
    for name in circuit_names:
        if not isinstance(name, str) or len(name) == 0:
            raise ValueError(f"Circuit name '{name}' is not a valid non-empty string.")
    return True

check_cable_positions classmethod

check_cable_positions(df: DataFrame) -> bool

Ensure cable positions are valid CablePosition enum values.

Source code in cable_thermal_model/model/schemas/model_output_schemas.py

@pa.dataframe_check
@classmethod
def check_cable_positions(cls, df: pd.DataFrame) -> bool:
    """Ensure cable positions are valid CablePosition enum values."""
    positions = df.columns.get_level_values(1).unique()
    return bool([CablePosition(pos) for pos in positions])

check_cable_layers classmethod

check_cable_layers(df: DataFrame) -> bool

Ensure cable layers are valid CableLayer enum values.

Source code in cable_thermal_model/model/schemas/model_output_schemas.py

@pa.dataframe_check
@classmethod
def check_cable_layers(cls, df: pd.DataFrame) -> bool:
    """Ensure cable layers are valid CableLayer enum values."""
    layers = df.columns.get_level_values(2).unique()
    return bool([CableLayer(layer) for layer in layers])

check_temperature_values classmethod

check_temperature_values(df: DataFrame) -> bool

Ensure temperature values are floats.

Source code in cable_thermal_model/model/schemas/model_output_schemas.py

@pa.dataframe_check
@classmethod
def check_temperature_values(cls, df: pd.DataFrame) -> bool:
    """Ensure temperature values are floats."""
    for dtype in df.dtypes:
        if not pd.api.types.is_float_dtype(dtype):
            raise ValueError("All temperature values must be of float type.")
    return True

State

Bases: BaseModel

Stores information about temperatures within cables at the final state.

The final state is reached at the end of the simulation. In addition, the relevant cable representations and their properties are stored.

Attributes:

Name	Type	Description
cable_representations	list[PosCable]	list[PosCable]: List of cable representations with their properties and positions in the environment.
full_solution	dict[CableKey, ndarray]	dict[CableKey, np.ndarray]: Combines the internal heating solution with the ambient temperature profile and, for a StateSoil object, the mutual heating solution.
internal_heating_solution	dict[CableKey, ndarray]	dict[CableKey, np.ndarray]: The temperature delta profile as a result of internal heating due to the load.

check_solution_consistency

check_solution_consistency()

Validate that full_solution and internal_heating_solution share the same cable keys.

Source code in cable_thermal_model/model/schemas/state_schemas.py

@model_validator(mode="after")
def check_solution_consistency(self):
    """Validate that full_solution and internal_heating_solution share the same cable keys."""
    keys_full_solution = set(self.full_solution.keys())
    keys_solution = set(self.internal_heating_solution.keys())
    if keys_full_solution != keys_solution:
        raise ValueError(
            f"Inconsistent keys between full_solution and solution. Keys in full_solution: {keys_full_solution}, "
            f"keys in solution: {keys_solution}"
        )
    return self

check_cable_representations_consistency

check_cable_representations_consistency()

Validate that cable_representations and internal_heating_solution share the same cable keys.

Source code in cable_thermal_model/model/schemas/state_schemas.py

@model_validator(mode="after")
def check_cable_representations_consistency(self):
    """Validate that cable_representations and internal_heating_solution share the same cable keys."""
    keys_cables = {cable.name for cable in self.cable_representations}
    keys_solution = set(self.internal_heating_solution.keys())
    if keys_solution != keys_cables:
        raise ValueError(
            f"Keys in solution must match cable representations. "
            f"Keys in solution: {keys_solution}, "
            f"Cable keys from representations: {keys_cables}"
        )
    return self

StateAir

Bases: State

StateAir has no added attributes on top of State.

However, we want to make sure there is only one circuit (check for a unique circuit_name).

validate_single_circuit

validate_single_circuit()

Ensure that all cable representations in StateAir belong to the same circuit.

Source code in cable_thermal_model/model/schemas/state_schemas.py

@model_validator(mode="after")
def validate_single_circuit(self):
    """Ensure that all cable representations in StateAir belong to the same circuit."""
    circuit_names = {cable_key.circuit_name for cable_key in self.cable_representations}
    if len(circuit_names) > 1:
        raise ValueError(f"StateAir should only contain one circuit, but found multiple: {circuit_names}")
    return self

StateSoil

Bases: State

Extends upon the base State class. Includes additional attribute mutual_heating_solutions and validation thereof.

Attributes:

Name	Type	Description
mutual_heating_solutions	dict[CableKey, ndarray]	dict[CableKey, np.ndarray] A dictionary containing the temperature increase inside a cable due to mutual heating from other cables in the environment. This is stored as a dict with CableKey as key and an array of temperature increases per grid point as value.

validate_mutual_heating_solutions

validate_mutual_heating_solutions()

Validate that mutual_heating_solutions keys match the cable representation keys.

Source code in cable_thermal_model/model/schemas/state_schemas.py

@model_validator(mode="after")
def validate_mutual_heating_solutions(self):
    """Validate that mutual_heating_solutions keys match the cable representation keys."""
    found_keys = set(self.mutual_heating_solutions.keys())
    expected_keys = {cable.name for cable in self.cable_representations}
    if found_keys != expected_keys:
        raise ValueError(
            "CableKeys of mutual_heating_solutions should match with "
            "the CableKeys in the cable representations. "
            f"Found keys: {found_keys}, expected keys: {expected_keys}"
        )
    return self