258 lines
9.0 KiB
Python
258 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from collections.abc import Mapping
|
|
from typing import Any, ClassVar
|
|
|
|
from PythonModels.core.equations import EquationResidual
|
|
from PythonModels.core.metadata import (
|
|
ParameterDefinition,
|
|
ResultVariableDefinition,
|
|
ResultVariableMetadata,
|
|
THERMODYNAMIC_VOLUME_RESULT_VARIABLES,
|
|
)
|
|
from PythonModels.core.ports import PortDefinition, PortState
|
|
|
|
|
|
class Component(ABC):
|
|
MODEL_TYPE: ClassVar[str | None] = None
|
|
PORTS: ClassVar[tuple[PortDefinition, ...]] = ()
|
|
PARAMETERS: ClassVar[tuple[ParameterDefinition, ...]] = ()
|
|
RESULT_VARIABLES: ClassVar[tuple[ResultVariableDefinition, ...]] = ()
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
self.model_type = self.MODEL_TYPE or self.__class__.__name__.lower()
|
|
self._ports: dict[str, PortState] = {}
|
|
self._parameter_values: dict[str, float] = {}
|
|
|
|
@property
|
|
def ports(self) -> dict[str, PortState]:
|
|
return dict(self._ports)
|
|
|
|
@property
|
|
def port_definitions(self) -> tuple[PortDefinition, ...]:
|
|
return tuple(
|
|
port.definition
|
|
for port in self._ports.values()
|
|
if port.definition is not None
|
|
)
|
|
|
|
def register_port(self, port: PortState) -> PortState:
|
|
definition = port.definition
|
|
if definition is None:
|
|
raise ValueError(f"Component {self.name} cannot register an undefined port.")
|
|
if definition.name in self._ports:
|
|
raise ValueError(f"Duplicate port {self.name}.{definition.name}.")
|
|
self._ports[definition.name] = port
|
|
return port
|
|
|
|
def register_declared_port(self, name: str) -> PortState:
|
|
try:
|
|
definition = next(item for item in self.PORTS if item.name == name)
|
|
except StopIteration as exc:
|
|
raise ValueError(
|
|
f"Component model {self.model_type} does not declare port {name}."
|
|
) from exc
|
|
return self.register_port(PortState(definition=definition))
|
|
|
|
def set_parameter_values(self, values: Mapping[str, float]) -> None:
|
|
definitions = {definition.name: definition for definition in self.PARAMETERS}
|
|
unknown = sorted(set(values) - set(definitions))
|
|
if unknown:
|
|
raise ValueError(
|
|
f"Component {self.name} contains unsupported parameters: "
|
|
+ ", ".join(unknown)
|
|
+ "."
|
|
)
|
|
missing = sorted(set(definitions) - set(values))
|
|
if missing:
|
|
raise ValueError(
|
|
f"Component {self.name} is missing parameters: "
|
|
+ ", ".join(missing)
|
|
+ "."
|
|
)
|
|
|
|
resolved: dict[str, float] = {}
|
|
for name, definition in definitions.items():
|
|
value = float(values[name])
|
|
message = definition.validation_message(value)
|
|
if message is not None:
|
|
raise ValueError(
|
|
f"Parameter '{name}' on component '{self.name}' {message}."
|
|
)
|
|
resolved[name] = value
|
|
self._parameter_values = resolved
|
|
|
|
@property
|
|
def parameter_values(self) -> dict[str, float]:
|
|
return dict(self._parameter_values)
|
|
|
|
def get_port(self, name: str) -> PortState:
|
|
try:
|
|
return self._ports[name]
|
|
except KeyError as exc:
|
|
raise ValueError(f"Component {self.name} has no port named {name}.") from exc
|
|
|
|
def component_result_values(self) -> Mapping[str, float]:
|
|
return {}
|
|
|
|
def result_values(self) -> dict[str, float]:
|
|
component_values = dict(self.component_result_values())
|
|
declared = {definition.name: definition for definition in self.RESULT_VARIABLES}
|
|
unknown = sorted(set(component_values) - set(declared))
|
|
if unknown:
|
|
raise ValueError(
|
|
f"Component {self.name} returned undeclared result variables: "
|
|
+ ", ".join(unknown)
|
|
+ "."
|
|
)
|
|
|
|
values: dict[str, float] = {}
|
|
for name, definition in declared.items():
|
|
if not definition.visible:
|
|
continue
|
|
if name not in component_values:
|
|
raise ValueError(
|
|
f"Component {self.name} did not provide declared result variable {name}."
|
|
)
|
|
values[name] = float(component_values[name])
|
|
|
|
for port_definition in self.port_definitions:
|
|
port = self.get_port(port_definition.name)
|
|
for variable in port_definition.variables:
|
|
if not variable.result_visible:
|
|
continue
|
|
values[f"{port_definition.name}.{variable.name}"] = float(
|
|
getattr(port, variable.name)
|
|
)
|
|
return values
|
|
|
|
def result_variable_metadata(self) -> tuple[ResultVariableMetadata, ...]:
|
|
metadata = [
|
|
ResultVariableMetadata(
|
|
key=f"{self.name}.{definition.name}",
|
|
component_id=self.name,
|
|
component_type=self.model_type,
|
|
scope="component",
|
|
name=definition.name,
|
|
label=definition.label,
|
|
quantity=definition.quantity,
|
|
unit=definition.unit,
|
|
category=definition.category,
|
|
order=definition.order,
|
|
)
|
|
for definition in self.RESULT_VARIABLES
|
|
if definition.visible
|
|
]
|
|
for port_definition in self.port_definitions:
|
|
for variable in port_definition.variables:
|
|
if not variable.result_visible:
|
|
continue
|
|
metadata.append(
|
|
ResultVariableMetadata(
|
|
key=f"{self.name}.{port_definition.name}.{variable.name}",
|
|
component_id=self.name,
|
|
component_type=self.model_type,
|
|
scope="port",
|
|
port_name=port_definition.name,
|
|
name=variable.name,
|
|
label=variable.label or variable.name,
|
|
quantity=variable.quantity or variable.name,
|
|
unit=variable.unit,
|
|
category=variable.role,
|
|
order=variable.order,
|
|
)
|
|
)
|
|
return tuple(metadata)
|
|
|
|
def parameter_interface_dicts(self) -> list[dict[str, object]]:
|
|
return [
|
|
definition.as_interface_dict(
|
|
value=self._parameter_values.get(definition.name)
|
|
)
|
|
for definition in self.PARAMETERS
|
|
]
|
|
|
|
def pressure_flow_equation_residuals(self) -> tuple[EquationResidual, ...]:
|
|
"""Return algebraic residuals after the network assigns port states."""
|
|
|
|
return ()
|
|
|
|
def update_stream_outflows(self, connected_h: Mapping[str, float]) -> None:
|
|
"""Update connector outflow properties from current flow directions."""
|
|
|
|
return None
|
|
|
|
|
|
class DynamicComponent(Component):
|
|
state_size = 2
|
|
|
|
@staticmethod
|
|
def actual_stream_enthalpy(
|
|
port_m_flow: float,
|
|
connected_h: float,
|
|
internal_h: float,
|
|
) -> float:
|
|
"""Approximate `actualStream(port.h_outflow)` for a mixed control volume port."""
|
|
|
|
return connected_h if port_m_flow > 0.0 else internal_h
|
|
|
|
def connection_inlet_enthalpy(
|
|
self,
|
|
port_m_flow: float,
|
|
connected_h: float,
|
|
internal_h: float,
|
|
) -> float:
|
|
"""Resolve the enthalpy convected into this control volume through one port."""
|
|
|
|
return self.actual_stream_enthalpy(
|
|
port_m_flow=port_m_flow,
|
|
connected_h=connected_h,
|
|
internal_h=internal_h,
|
|
)
|
|
|
|
@abstractmethod
|
|
def get_state_vector(self) -> list[float]:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
def set_state_vector(self, values: list[float]) -> None:
|
|
raise NotImplementedError
|
|
|
|
def refresh_thermodynamic_ports(self) -> Any:
|
|
raise NotImplementedError
|
|
|
|
def state_derivative_from_ports(
|
|
self,
|
|
connected_h: Mapping[str, float],
|
|
) -> list[float]:
|
|
raise NotImplementedError
|
|
|
|
|
|
class ThermodynamicVolumeComponent(DynamicComponent):
|
|
"""Two-state gas volume exposing the shared thermodynamic result contract."""
|
|
|
|
RESULT_VARIABLES = THERMODYNAMIC_VOLUME_RESULT_VARIABLES
|
|
|
|
def component_result_values(self) -> Mapping[str, float]:
|
|
state = self.get_state_vector()
|
|
if len(state) < 2:
|
|
raise ValueError(
|
|
f"Thermodynamic component {self.name} must expose mass and energy states."
|
|
)
|
|
properties = self.refresh_thermodynamic_ports()
|
|
return {
|
|
"m": float(state[0]),
|
|
"U": float(state[1]),
|
|
"p": float(properties.p),
|
|
"T": float(properties.T),
|
|
"rho": float(properties.rho),
|
|
"u": float(properties.u),
|
|
"h": float(properties.h),
|
|
}
|
|
|
|
|
|
class AlgebraicComponent(Component):
|
|
"""Stateless element described by algebraic constraints only."""
|