127 lines
6.4 KiB
Python
127 lines
6.4 KiB
Python
from __future__ import annotations
|
|
from abc import ABC
|
|
from collections.abc import Mapping
|
|
from typing import TYPE_CHECKING, ClassVar
|
|
from app.simulation.core.catalog import ComponentDisplaySpec
|
|
from app.simulation.core.equations import EquationDefinition
|
|
from app.simulation.core.metadata import ParameterDefinition, ResultVariableDefinition, ResultVariableMetadata, THERMODYNAMIC_VOLUME_RESULT_VARIABLES
|
|
from app.simulation.core.ports import PortDefinition, PortState
|
|
if TYPE_CHECKING:
|
|
from app.simulation.core.medium import GasMedium
|
|
|
|
class Component(ABC):
|
|
MODEL_TYPE: ClassVar[str | None] = None
|
|
MODEL_VERSION: ClassVar[str | None] = None
|
|
PORTS: ClassVar[tuple[PortDefinition, ...]] = ()
|
|
PARAMETERS: ClassVar[tuple[ParameterDefinition, ...]] = ()
|
|
RESULT_VARIABLES: ClassVar[tuple[ResultVariableDefinition, ...]] = ()
|
|
DISPLAY: ClassVar[ComponentDisplaySpec | None] = None
|
|
|
|
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))
|
|
|
|
@classmethod
|
|
def active_port_definitions_for_parameters(cls, parameters: Mapping[str, float]) -> tuple[PortDefinition, ...]:
|
|
"""Declared ports enabled by one normalized parameter set."""
|
|
return cls.PORTS
|
|
|
|
@property
|
|
def active_port_definitions(self) -> tuple[PortDefinition, ...]:
|
|
"""Instance ports that participate in execution and result reporting."""
|
|
return self.port_definitions
|
|
|
|
@property
|
|
def required_connection_ports(self) -> tuple[str, ...]:
|
|
"""Physical ports that must have an external connection before simulation."""
|
|
return tuple((definition.name for definition in self.active_port_definitions if definition.kind == 'physical'))
|
|
|
|
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 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.active_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]
|
|
|
|
@classmethod
|
|
def create(cls, *, name: str, medium: GasMedium, parameters: Mapping[str, float]) -> Component:
|
|
"""Create a catalog model from normalized SI parameters."""
|
|
raise NotImplementedError(f'Component model {cls.__name__} must implement create().')
|
|
EQUATIONS = ()
|
|
|
|
def equation_definitions(self):
|
|
|
|
def bind(value):
|
|
if isinstance(value, str):
|
|
return value.replace('__MODEL__', self.name)
|
|
return tuple((bind(v) for v in value))
|
|
return tuple((EquationDefinition(id=bind(e['id']), owner=e['owner'], owner_id=self.name, relation=e['relation'], variables=bind(e['variables']), role=e['role']) for e in self.EQUATIONS))
|
|
|
|
class DynamicComponent(Component):
|
|
state_size = 2
|
|
|
|
class ThermodynamicVolumeComponent(DynamicComponent):
|
|
"""Two-state gas volume exposing the shared thermodynamic result contract."""
|
|
RESULT_VARIABLES = THERMODYNAMIC_VOLUME_RESULT_VARIABLES
|
|
|
|
class AlgebraicComponent(Component):
|
|
"""Stateless element described by algebraic constraints only."""
|