from __future__ import annotations from abc import ABC, abstractmethod from collections.abc import Mapping from typing import TYPE_CHECKING, Any, ClassVar from app.simulation.core.catalog import ComponentDisplaySpec from app.simulation.core.equations import EquationResidual 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 ) 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 ] @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()." ) 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 def update_flow_temperature_references( self, connected_h: Mapping[str, float], ) -> None: """Update enthalpy references used only by pressure-flow laws. Most components use the normal stream enthalpy for both energy transport and upstream-property evaluation. AMESim node submodels can expose a distinct temperature reference, so the default is a no-op. """ return None def pneumatic_volume_outputs(self) -> Mapping[str, tuple[float, float]]: """Return directed ``volume``/``volume_flow`` values by pneumatic port. Most pneumatic components contribute no external chamber volume. Moving boundaries such as PNRP17 override this hook; the network resolver then propagates the pair to the component connected at the same physical port. """ return {} 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."""