schemas

ScenarioModelAir

Bases: AbstractScenarioModel

Air scenario schema extending the base scenario schema.

ScenarioModelSoil

Bases: AbstractScenarioModel

Soil scenario schema extending the base scenario schema.

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 level-0 names are valid circuit names or the measurement-point prefix.

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 level-0 names are valid circuit names or the measurement-point prefix."""
    circuit_names = df.columns.get_level_values(0).unique()
    for name in circuit_names:
        if name == MEASUREMENT_POINT_KEY_PREFIX:
            continue  # Skip validation for measurement-point columns
        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

Validate level-1 and level-2 values by column type.

• For cable result columns: level 1 must be a valid CablePosition. Level 2 must be a valid CableLayer.
• For measurement-point columns: level 1 must be a string starting with 'x='. Level 2 must be a string starting with 'y='.

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:
    """Validate level-1 and level-2 values by column type.

    - For cable result columns: level 1 must be a valid CablePosition. Level 2 must be a valid CableLayer.
    - For measurement-point columns: level 1 must be a string starting with 'x='. Level 2 must be a string
        starting with 'y='.
    """
    level0_values = df.columns.get_level_values(0)
    level1_values = df.columns.get_level_values(1)
    level2_values = df.columns.get_level_values(2)

    for level0, level1, level2 in zip(level0_values, level1_values, level2_values, strict=True):
        if level0 == MEASUREMENT_POINT_KEY_PREFIX:
            if not isinstance(level1, str) or not level1.startswith("x="):
                raise ValueError("Measurement-point columns must have level 1 as a string starting with 'x='.")
            if not isinstance(level2, str) or not level2.startswith("y="):
                raise ValueError("Measurement-point columns must have level 2 as a string starting with 'y='.")
        else:
            CablePosition(level1)
            CableLayer(level2)

    return True

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
static_env_hash	int	str: Deterministic hash of the static environment, used for validation and consistency checks.
temperature	dict[CableKey, ndarray]	dict[CableKey, np.ndarray]: Combines the self-heating contribution with the ambient temperature profile and, for a StateSoil object, the mutual-heating contribution.
self_heating_contribution	dict[CableKey, ndarray]	dict[CableKey, np.ndarray]: The temperature delta profile as a result of self-heating due to the load.

check_solution_consistency

check_solution_consistency()

Validate that temperature and self_heating_contribution 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 temperature and self_heating_contribution share the same cable keys."""
    keys_temperature = set(self.temperature.keys())
    keys_solution = set(self.self_heating_contribution.keys())
    if keys_temperature != keys_solution:
        raise ValueError(
            f"Inconsistent keys between temperature and self_heating_contribution. "
            f"Keys in temperature: {keys_temperature}, "
            f"keys in self_heating_contribution: {keys_solution}"
        )
    return self

serialize_temperature

serialize_temperature(arrays: dict[CableKey, ndarray]) -> dict[str, list[float]]

Serialize numpy arrays to lists for JSON compatibility.

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

@field_serializer("temperature", "self_heating_contribution")
def serialize_temperature(self, arrays: dict[CableKey, np.ndarray]) -> dict[str, list[float]]:
    """Serialize numpy arrays to lists for JSON compatibility."""
    return _serialize_temperature_dict(arrays)

deserialize_temperature

deserialize_temperature(value: dict[str | CableKey, list[float] | ndarray]) -> dict[CableKey, ndarray]

Deserialize lists back to numpy arrays.

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

@field_validator("temperature", "self_heating_contribution", mode="before")
def deserialize_temperature(
    cls, value: dict[str | CableKey, list[float] | np.ndarray]
) -> dict[CableKey, np.ndarray]:
    """Deserialize lists back to numpy arrays."""
    return _deserialize_temperature_dict(value)

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 keys 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 keys in StateAir belong to the same circuit."""
    cable_keys = self.temperature.keys()
    circuit_names = {cable_key.circuit_name for cable_key in cable_keys}
    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_contribution and its validation.

Attributes:

Name	Type	Description
mutual_heating_contribution	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.

serialize_mutual_heating_contribution

serialize_mutual_heating_contribution(arrays: dict[CableKey, ndarray]) -> dict[str, list[float]]

Serialize mutual-heating arrays to lists for JSON compatibility.

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

@field_serializer("mutual_heating_contribution")
def serialize_mutual_heating_contribution(self, arrays: dict[CableKey, np.ndarray]) -> dict[str, list[float]]:
    """Serialize mutual-heating arrays to lists for JSON compatibility."""
    return _serialize_temperature_dict(arrays)

deserialize_mutual_heating_contribution

deserialize_mutual_heating_contribution(value: dict[str | CableKey, list[float] | ndarray]) -> dict[CableKey, ndarray]

Deserialize lists back to numpy arrays for mutual-heating.

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

@field_validator("mutual_heating_contribution", mode="before")
def deserialize_mutual_heating_contribution(
    cls, value: dict[str | CableKey, list[float] | np.ndarray]
) -> dict[CableKey, np.ndarray]:
    """Deserialize lists back to numpy arrays for mutual-heating."""
    return _deserialize_temperature_dict(value)

validate_mutual_heating_contribution

validate_mutual_heating_contribution()

Validate that mutual_heating_contribution keys match cable keys.

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

@model_validator(mode="after")
def validate_mutual_heating_contribution(self):
    """Validate that mutual_heating_contribution keys match cable keys."""
    found_keys = set(self.mutual_heating_contribution.keys())
    expected_keys = set(self.temperature.keys())
    if found_keys != expected_keys:
        raise ValueError(
            "CableKeys of mutual_heating_contribution should match with cable_keys of temperature."
            f"Found keys: {found_keys}, expected keys: {expected_keys}"
        )
    return self