"""FMI 3.0 co-simulation FMU wrapping the TMHP ASHPB ``step()`` kernel.
This adapter mirrors :mod:`tmhp.integrations.fmu` at the TMHP boundary but uses
``pythonfmu3`` and the FMI 3.0 ``Fmi3StepResult`` return contract. It targets
FMI 3.0 Co-Simulation only: the FMU owns the ASHPB dynamic state and advances it
once for each importer communication step.
Build::
pythonfmu3 build -f src/tmhp/integrations/fmu3.py .
The adapter intentionally does not expose FMI 3.0 clocks, Scheduled Execution,
or array variables. Those features are useful for embedded controls and
multi-rate models; TMHP's current ASHPB FMU boundary is scalar Co-Simulation.
"""
from __future__ import annotations
from typing import Any, cast
from xml.etree.ElementTree import Element
from pythonfmu3 import (
Boolean,
Float64,
Fmi3Causality,
Fmi3Slave,
Fmi3Status,
Fmi3StepResult,
Fmi3Variability,
String,
)
from tmhp import AirSourceHeatPumpBoiler
from tmhp.dynamic_context import DynamicState
from tmhp.integrations import _fmi_common
_finite = _fmi_common.finite
_is_finite = _fmi_common.is_finite
_failure_reason = _fmi_common.failure_reason
_DESCRIPTIONS = _fmi_common.VARIABLE_DESCRIPTIONS
_REAL_UNITS = {
"time": "s",
"hp_capacity": "W",
"T_tank_w_init": "degC",
"T_sur": "degC",
"T0": "degC",
"dhw_draw": "m3/s",
"T_sup_w": "degC",
"E_cmp": "W",
"E_tot": "W",
"Q_ref_tank": "W",
"cop_sys": "1",
"T_tank_w": "degC",
"V_cmp_disp_cc": "cm3",
"dV_fan_a_rated": "m3/s",
}
_UNIT_SPECS: tuple[tuple[str, dict[str, str]], ...] = (
("W", {"kg": "1", "m": "2", "s": "-3"}),
("s", {"s": "1"}),
("degC", {"K": "1", "offset": "273.15"}),
("m3/s", {"m": "3", "s": "-1"}),
("cm3", {"m": "3", "factor": "1e-6"}),
("1", {}),
)
[docs]
class TmhpAshpbFmi3Slave(Fmi3Slave):
"""ASHPB single-timestep co-simulation kernel (FMI 3.0)."""
author = "BET Lab"
description = "TMHP ASHPB one-dt co-simulation kernel (FMI 3.0)"
[docs]
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.time = 0.0
self.register_variable(
Float64(
"time",
causality=Fmi3Causality.independent,
variability=Fmi3Variability.continuous,
unit=_REAL_UNITS["time"],
)
)
self.ref = "R32"
self.hp_capacity = 15000.0
self.T_tank_w_init = 55.0
self.T_sur = 20.0
self.preset = ""
self.V_cmp_disp_cc = 0.0
self.dV_fan_a_rated = 0.0
self.register_variable(
String(
"ref",
causality=Fmi3Causality.parameter,
variability=Fmi3Variability.fixed,
description=_DESCRIPTIONS["ref"],
)
)
for name in ("hp_capacity", "T_tank_w_init", "T_sur"):
self.register_variable(
Float64(
name,
causality=Fmi3Causality.parameter,
variability=Fmi3Variability.fixed,
unit=_REAL_UNITS[name],
description=_DESCRIPTIONS[name],
)
)
self.T0 = 7.0
self.dhw_draw = 0.0
self.T_sup_w = 15.0
for name in ("T0", "dhw_draw", "T_sup_w"):
self.register_variable(
Float64(
name,
causality=Fmi3Causality.input,
variability=Fmi3Variability.continuous,
unit=_REAL_UNITS[name],
description=_DESCRIPTIONS[name],
)
)
self.E_cmp = 0.0
self.E_tot = 0.0
self.Q_ref_tank = 0.0
self.cop_sys = 0.0
self.T_tank_w = self.T_tank_w_init
self.hp_is_on = False
self.converged = True
self.failure_reason = "none"
for name in ("E_cmp", "E_tot", "Q_ref_tank", "cop_sys", "T_tank_w"):
self.register_variable(
Float64(
name,
causality=Fmi3Causality.output,
variability=Fmi3Variability.continuous,
unit=_REAL_UNITS[name],
description=_DESCRIPTIONS[name],
)
)
for name in ("hp_is_on", "converged"):
self.register_variable(
Boolean(
name,
causality=Fmi3Causality.output,
variability=Fmi3Variability.discrete,
description=_DESCRIPTIONS[name],
)
)
self.register_variable(
String(
"failure_reason",
causality=Fmi3Causality.output,
variability=Fmi3Variability.discrete,
description=_DESCRIPTIONS["failure_reason"],
)
)
# Appended parameters preserve all existing FMI 3.0 value references.
self.register_variable(
String(
"preset",
causality=Fmi3Causality.parameter,
variability=Fmi3Variability.fixed,
description=_DESCRIPTIONS["preset"],
)
)
for name in ("V_cmp_disp_cc", "dV_fan_a_rated"):
self.register_variable(
Float64(
name,
causality=Fmi3Causality.parameter,
variability=Fmi3Variability.fixed,
unit=_REAL_UNITS[name],
description=_DESCRIPTIONS[name],
)
)
self._hp: AirSourceHeatPumpBoiler | None = None
self._state: DynamicState | None = None
self._n = 0
[docs]
def to_xml(self, model_options: dict[str, str] | None = None) -> Element:
"""Build a static FMI 3.0 model description for PythonFMU3."""
root = cast(Element, super().to_xml({} if model_options is None else model_options))
_fmi_common.ensure_unit_definitions(
root,
_UNIT_SPECS,
insertion_after={"CoSimulation", "ModelExchange", "ScheduledExecution"},
)
return root
[docs]
def exit_initialization_mode(self) -> None:
"""Finalize parameters and initialize the carried ASHPB state."""
preset_kwargs = _fmi_common.preset_kwargs(
self.preset,
ref=self.ref,
hp_capacity=self.hp_capacity,
V_cmp_disp_cc=self.V_cmp_disp_cc,
dV_fan_a_rated=self.dV_fan_a_rated,
)
self._hp = AirSourceHeatPumpBoiler(ref=self.ref, hp_capacity=self.hp_capacity, **preset_kwargs)
self._state = self._hp.make_initial_state(self.T_tank_w_init)
self._n = 0
self.T_tank_w = self.T_tank_w_init
[docs]
def do_step(self, current_time: float, step_size: float) -> Fmi3StepResult:
"""Advance the FMU by one FMI 3.0 communication step."""
if self._hp is None or self._state is None:
raise RuntimeError("FMU slave used before exit_initialization_mode()")
if not (
_is_finite(current_time)
and _is_finite(step_size)
and float(step_size) > 0.0
and _is_finite(self.T0)
and _is_finite(self.dhw_draw)
and float(self.dhw_draw) >= 0.0
and _is_finite(self.T_sup_w)
and _is_finite(self.T_sur)
):
self.hp_is_on = False
self.converged = False
self.failure_reason = "invalid_input"
return Fmi3StepResult(status=Fmi3Status.discard, earlyReturn=True)
inputs = {
"n": self._n,
"current_time_s": float(current_time),
"T0": float(self.T0),
"dV_mix_w_out": float(self.dhw_draw),
"T_sup_w": float(self.T_sup_w),
"T_sur": float(self.T_sur),
"I_DN": 0.0,
"I_dH": 0.0,
}
self._state, res = self._hp.step(self._state, inputs, float(step_size))
self.time = float(current_time) + float(step_size)
self.E_cmp = _finite(res["E_cmp [W]"])
self.E_tot = _finite(res["E_tot [W]"])
self.Q_ref_tank = _finite(res["Q_ref_tank [W]"])
self.cop_sys = _finite(res.get("cop_sys [-]", float("nan")))
self.T_tank_w = _finite(res["T_tank_w [°C]"])
self.hp_is_on = bool(res.get("hp_is_on", self.E_cmp > 0.0))
self.converged = bool(res.get("converged", True))
self.failure_reason = _failure_reason(res.get("failure_reason", "none"))
self._n += 1
return Fmi3StepResult(status=Fmi3Status.ok)