Files
SystemSimulationApp/PythonModels/systems/test_mql_closure.py
T

558 lines
20 KiB
Python

from __future__ import annotations
from dataclasses import dataclass
from typing import Callable
from PythonModels.components.amesim_pneumatic import (
AmesimPneumaticGas,
AmesimPneumaticOrifice,
AmesimPneumaticVolume,
)
from PythonModels.components.amesim_pneumatic_line import AmesimPnl0001Pipe
from PythonModels.core.medium import ThermodynamicProperties
from PythonModels.core.ports import PortState
from PythonModels.core.state import VolumeState
@dataclass(frozen=True)
class TestMqlPneumaticBranchSpec:
name: str
upstream_volume_alias: str
orifice_alias: str
downstream_volume_alias: str
source: str = "manual"
@dataclass(frozen=True)
class TestMqlPneumaticTopologyCandidate:
connection_alias: str
line_submodel: str
source_component: str
source_port: str
target_component: str
target_port: str
source_is_typed_pneumatic: bool
target_is_typed_pneumatic: bool
category: str
reason: str
@dataclass(frozen=True)
class TestMqlPneumaticChamberSegmentSpec:
name: str
inlet_node_alias: str
inlet_line_alias: str
inlet_orifice_alias: str
inlet_orifice_boundary_port: str
inlet_orifice_volume_port: str
volume_alias: str
volume_inlet_port: str
volume_outlet_port: str
outlet_orifice_alias: str
outlet_orifice_volume_port: str
outlet_orifice_boundary_port: str
outlet_line_alias: str
outlet_node_alias: str
source: str = "amesim-cir-topology"
@dataclass(frozen=True)
class TestMqlPneumaticTopologyDiscovery:
branch_specs: tuple[TestMqlPneumaticBranchSpec, ...]
chamber_segment_specs: tuple[TestMqlPneumaticChamberSegmentSpec, ...]
blocked_candidates: tuple[TestMqlPneumaticTopologyCandidate, ...]
@dataclass(frozen=True)
class TestMqlPneumaticBoundaryCondition:
pressure_pa: float
temperature_k: float
def properties(self, gas: AmesimPneumaticGas) -> ThermodynamicProperties:
if self.pressure_pa <= 0.0:
raise ValueError("boundary pressure must be positive")
if self.temperature_k <= 0.0:
raise ValueError("boundary temperature must be positive")
return ThermodynamicProperties(
p=self.pressure_pa,
T=self.temperature_k,
rho=gas.density(self.pressure_pa, self.temperature_k),
u=gas.specific_internal_energy(self.temperature_k),
h=gas.specific_enthalpy(self.temperature_k),
)
@dataclass(frozen=True)
class TestMqlPneumaticChamberSegmentComponents:
volume: AmesimPneumaticVolume
inlet_orifice: AmesimPneumaticOrifice
outlet_orifice: AmesimPneumaticOrifice
spec: TestMqlPneumaticChamberSegmentSpec
@dataclass(frozen=True)
class TestMqlPneumaticChamberSegmentSnapshot:
chamber: ThermodynamicProperties
inlet_boundary: ThermodynamicProperties
outlet_boundary: ThermodynamicProperties
inlet_flow: float
outlet_flow: float
@dataclass(frozen=True)
class TestMqlPnl0001ChamberSegmentComponents:
inlet_line: AmesimPnl0001Pipe
volume: AmesimPneumaticVolume
inlet_orifice: AmesimPneumaticOrifice
outlet_orifice: AmesimPneumaticOrifice
spec: TestMqlPneumaticChamberSegmentSpec
@dataclass(frozen=True)
class TestMqlPnl0001ChamberSegmentSnapshot:
inlet_line: ThermodynamicProperties
chamber: ThermodynamicProperties
inlet_node: ThermodynamicProperties
outlet_boundary: ThermodynamicProperties
node_to_line_flow: float
line_to_chamber_flow: float
outlet_flow: float
@dataclass(frozen=True)
class TestMqlPneumaticBranchComponents:
name: str
upstream_volume: AmesimPneumaticVolume
orifice: AmesimPneumaticOrifice
downstream_volume: AmesimPneumaticVolume
spec: TestMqlPneumaticBranchSpec | None = None
@dataclass(frozen=True)
class TestMqlPneumaticBranchState:
name: str
upstream: ThermodynamicProperties
downstream: ThermodynamicProperties
flow: float
upstream_inlet_h: float
downstream_inlet_h: float
@dataclass(frozen=True)
class TestMqlPneumaticSnapshot:
branch: TestMqlPneumaticBranchState
@property
def flow(self) -> float:
return self.branch.flow
@property
def upstream(self) -> ThermodynamicProperties:
return self.branch.upstream
@property
def downstream(self) -> ThermodynamicProperties:
return self.branch.downstream
class TestMqlPneumaticClosure:
"""Minimal test_mql pneumatic closure following the existing Testmodel pattern.
The canonical branch flow is positive from ``upstream_volume`` through the
orifice port_a/port_b into ``downstream_volume``. Port ``m_flow`` values are
written with Modelica-style signs: positive means flow into that component.
"""
def __init__(
self,
*,
components: TestMqlPneumaticBranchComponents,
initial_state_vector: Callable[[], list[float]],
apply_state_vector: Callable[[list[float]], None],
) -> None:
self.components = components
self._initial_state_vector = initial_state_vector
self._apply_state_vector = apply_state_vector
def initial_state_vector(self) -> list[float]:
return self._initial_state_vector()
def apply_state_vector(self, values: list[float]) -> None:
self._apply_state_vector(values)
def snapshot(
self,
state_vector: list[float] | None = None,
) -> TestMqlPneumaticSnapshot:
if state_vector is not None:
self._apply_state_vector(state_vector)
upstream = self.components.upstream_volume.properties()
downstream = self.components.downstream_volume.properties()
flow = self._solve_branch_flow(upstream, downstream)
upstream_inlet_h = self.components.upstream_volume.connection_inlet_enthalpy(
port_m_flow=-flow,
connected_h=downstream.h,
internal_h=upstream.h,
)
downstream_inlet_h = self.components.downstream_volume.connection_inlet_enthalpy(
port_m_flow=flow,
connected_h=upstream.h,
internal_h=downstream.h,
)
branch = TestMqlPneumaticBranchState(
name=self.components.name,
upstream=upstream,
downstream=downstream,
flow=flow,
upstream_inlet_h=upstream_inlet_h,
downstream_inlet_h=downstream_inlet_h,
)
self._write_port_states(branch)
return TestMqlPneumaticSnapshot(branch=branch)
def _solve_branch_flow(
self,
upstream: ThermodynamicProperties,
downstream: ThermodynamicProperties,
) -> float:
upstream_temperature = upstream.T if upstream.p >= downstream.p else downstream.T
return self.components.orifice.mass_flow(
upstream.p,
downstream.p,
upstream_temperature,
)
def _write_port_states(self, branch: TestMqlPneumaticBranchState) -> None:
upstream_volume = self.components.upstream_volume
downstream_volume = self.components.downstream_volume
orifice = self.components.orifice
upstream_volume.port_a.p = branch.upstream.p
upstream_volume.port_a.m_flow = -branch.flow
upstream_volume.port_a.h_outflow = branch.upstream.h
orifice.port_a.p = branch.upstream.p
orifice.port_a.m_flow = branch.flow
orifice.port_a.h_outflow = branch.upstream.h
orifice.port_b.p = branch.downstream.p
orifice.port_b.m_flow = -branch.flow
orifice.port_b.h_outflow = branch.downstream.h
downstream_volume.port_a.p = branch.downstream.p
downstream_volume.port_a.m_flow = branch.flow
downstream_volume.port_a.h_outflow = branch.downstream.h
def branch_derivatives(
self,
snapshot: TestMqlPneumaticSnapshot,
) -> tuple[VolumeState, VolumeState]:
flow = snapshot.flow
upstream_derivative = self.components.upstream_volume.derivatives(
inlet_h=snapshot.branch.upstream_inlet_h,
m_flow=-flow,
)
downstream_derivative = self.components.downstream_volume.derivatives(
inlet_h=snapshot.branch.downstream_inlet_h,
m_flow=flow,
)
return upstream_derivative, downstream_derivative
def rhs(self, state_vector: list[float]) -> list[float]:
snapshot = self.snapshot(state_vector)
upstream_derivative, downstream_derivative = self.branch_derivatives(snapshot)
return [
*upstream_derivative.as_vector(),
*downstream_derivative.as_vector(),
]
class TestMqlPneumaticChamberSegmentClosure:
"""Boundary-reduced fixed chamber segment discovered from AMESim topology.
The two PNL0001 lines remain prescribed boundary interfaces in this first
closure. Their aliases are retained in the spec so line dynamics can be
inserted later without rediscovering the component path.
"""
def __init__(
self,
*,
components: TestMqlPneumaticChamberSegmentComponents,
inlet_boundary: TestMqlPneumaticBoundaryCondition,
outlet_boundary: TestMqlPneumaticBoundaryCondition,
) -> None:
self.components = components
self.inlet_boundary = inlet_boundary
self.outlet_boundary = outlet_boundary
def initial_state_vector(self) -> list[float]:
return self.components.volume.get_state_vector()
def apply_state_vector(self, values: list[float]) -> None:
if len(values) != 2:
raise ValueError("single-chamber segment state vector requires two values")
self.components.volume.set_state_vector(values)
def snapshot(
self,
state_vector: list[float] | None = None,
) -> TestMqlPneumaticChamberSegmentSnapshot:
if state_vector is not None:
self.apply_state_vector(state_vector)
chamber = self.components.volume.properties()
inlet_boundary = self.inlet_boundary.properties(self.components.volume.gas)
outlet_boundary = self.outlet_boundary.properties(self.components.volume.gas)
inlet_temperature = (
inlet_boundary.T if inlet_boundary.p >= chamber.p else chamber.T
)
outlet_temperature = (
chamber.T if chamber.p >= outlet_boundary.p else outlet_boundary.T
)
inlet_flow = self.components.inlet_orifice.mass_flow(
inlet_boundary.p,
chamber.p,
inlet_temperature,
)
outlet_flow = self.components.outlet_orifice.mass_flow(
chamber.p,
outlet_boundary.p,
outlet_temperature,
)
snapshot = TestMqlPneumaticChamberSegmentSnapshot(
chamber=chamber,
inlet_boundary=inlet_boundary,
outlet_boundary=outlet_boundary,
inlet_flow=inlet_flow,
outlet_flow=outlet_flow,
)
self._write_port_states(snapshot)
return snapshot
@staticmethod
def _port(
component: AmesimPneumaticVolume | AmesimPneumaticOrifice,
port_name: str,
) -> PortState:
if port_name == "port_1":
return component.port_a
if port_name == "port_2":
return component.port_b
raise ValueError(f"unsupported pneumatic port: {port_name}")
def _write_port_states(
self,
snapshot: TestMqlPneumaticChamberSegmentSnapshot,
) -> None:
spec = self.components.spec
chamber = self.components.volume
inlet_orifice = self.components.inlet_orifice
outlet_orifice = self.components.outlet_orifice
inlet_boundary_port = self._port(
inlet_orifice,
spec.inlet_orifice_boundary_port,
)
inlet_volume_port = self._port(inlet_orifice, spec.inlet_orifice_volume_port)
inlet_boundary_port.p = snapshot.inlet_boundary.p
inlet_boundary_port.m_flow = snapshot.inlet_flow
inlet_boundary_port.h_outflow = snapshot.inlet_boundary.h
inlet_volume_port.p = snapshot.chamber.p
inlet_volume_port.m_flow = -snapshot.inlet_flow
inlet_volume_port.h_outflow = snapshot.chamber.h
chamber_inlet_port = self._port(chamber, spec.volume_inlet_port)
chamber_outlet_port = self._port(chamber, spec.volume_outlet_port)
chamber_inlet_port.m_flow = snapshot.inlet_flow
chamber_outlet_port.m_flow = -snapshot.outlet_flow
outlet_volume_port = self._port(
outlet_orifice,
spec.outlet_orifice_volume_port,
)
outlet_boundary_port = self._port(
outlet_orifice,
spec.outlet_orifice_boundary_port,
)
outlet_volume_port.p = snapshot.chamber.p
outlet_volume_port.m_flow = snapshot.outlet_flow
outlet_volume_port.h_outflow = snapshot.chamber.h
outlet_boundary_port.p = snapshot.outlet_boundary.p
outlet_boundary_port.m_flow = -snapshot.outlet_flow
outlet_boundary_port.h_outflow = snapshot.outlet_boundary.h
def rhs(self, state_vector: list[float]) -> list[float]:
snapshot = self.snapshot(state_vector)
spec = self.components.spec
chamber = self.components.volume
derivative = chamber.derivatives_from_two_connections(
port_a_m_flow=self._port(chamber, "port_1").m_flow,
connected_h_a=(
snapshot.inlet_boundary.h
if spec.volume_inlet_port == "port_1"
else snapshot.outlet_boundary.h
),
port_b_m_flow=self._port(chamber, "port_2").m_flow,
connected_h_b=(
snapshot.inlet_boundary.h
if spec.volume_inlet_port == "port_2"
else snapshot.outlet_boundary.h
),
internal_h=snapshot.chamber.h,
)
return derivative.as_vector()
class TestMqlPnl0001ChamberSegmentClosure:
"""Fixed chamber segment with the topology-derived inlet PNL0001 state.
The inlet line has a closed causal boundary here: port-1 pressure and
temperature come from the PN3 node boundary, while port-2 flow comes from
the fixed orifice. The outlet line remains a boundary until its three-line
PN3 node balance is assembled.
"""
def __init__(
self,
*,
components: TestMqlPnl0001ChamberSegmentComponents,
inlet_node: TestMqlPneumaticBoundaryCondition,
outlet_boundary: TestMqlPneumaticBoundaryCondition,
) -> None:
self.components = components
self.inlet_node = inlet_node
self.outlet_boundary = outlet_boundary
def initial_state_vector(self) -> list[float]:
return [
*self.components.inlet_line.get_state_vector(),
*self.components.volume.get_state_vector(),
]
def apply_state_vector(self, values: list[float]) -> None:
if len(values) != 4:
raise ValueError("PNL0001/chamber segment state vector requires four values")
self.components.inlet_line.set_state_vector(values[:2])
self.components.volume.set_state_vector(values[2:])
def snapshot(
self,
state_vector: list[float] | None = None,
) -> TestMqlPnl0001ChamberSegmentSnapshot:
if state_vector is not None:
self.apply_state_vector(state_vector)
line = self.components.inlet_line.properties()
chamber = self.components.volume.properties()
inlet_node = self.inlet_node.properties(self.components.inlet_line.gas)
outlet_boundary = self.outlet_boundary.properties(self.components.volume.gas)
node_to_line_flow = self.components.inlet_line.resistance_mass_flow(
port_1_pressure_pa=inlet_node.p,
port_1_temperature_k=inlet_node.T,
)
inlet_temperature = line.T if line.p >= chamber.p else chamber.T
line_to_chamber_flow = self.components.inlet_orifice.mass_flow(
line.p,
chamber.p,
inlet_temperature,
)
outlet_temperature = (
chamber.T if chamber.p >= outlet_boundary.p else outlet_boundary.T
)
outlet_flow = self.components.outlet_orifice.mass_flow(
chamber.p,
outlet_boundary.p,
outlet_temperature,
)
snapshot = TestMqlPnl0001ChamberSegmentSnapshot(
inlet_line=line,
chamber=chamber,
inlet_node=inlet_node,
outlet_boundary=outlet_boundary,
node_to_line_flow=node_to_line_flow,
line_to_chamber_flow=line_to_chamber_flow,
outlet_flow=outlet_flow,
)
self._write_port_states(snapshot)
return snapshot
@staticmethod
def _port(
component: AmesimPneumaticVolume | AmesimPneumaticOrifice,
port_name: str,
) -> PortState:
return TestMqlPneumaticChamberSegmentClosure._port(component, port_name)
def _write_port_states(
self,
snapshot: TestMqlPnl0001ChamberSegmentSnapshot,
) -> None:
spec = self.components.spec
line = self.components.inlet_line
chamber = self.components.volume
inlet_orifice = self.components.inlet_orifice
outlet_orifice = self.components.outlet_orifice
line.port_1.p = snapshot.inlet_node.p
line.port_1.m_flow = snapshot.node_to_line_flow
line.port_1.h_outflow = snapshot.inlet_line.h
line.port_2.p = snapshot.inlet_line.p
line.port_2.m_flow = -snapshot.line_to_chamber_flow
inlet_boundary_port = self._port(
inlet_orifice,
spec.inlet_orifice_boundary_port,
)
inlet_volume_port = self._port(inlet_orifice, spec.inlet_orifice_volume_port)
inlet_boundary_port.p = snapshot.inlet_line.p
inlet_boundary_port.m_flow = snapshot.line_to_chamber_flow
inlet_boundary_port.h_outflow = snapshot.inlet_line.h
inlet_volume_port.p = snapshot.chamber.p
inlet_volume_port.m_flow = -snapshot.line_to_chamber_flow
inlet_volume_port.h_outflow = snapshot.chamber.h
chamber_inlet_port = self._port(chamber, spec.volume_inlet_port)
chamber_outlet_port = self._port(chamber, spec.volume_outlet_port)
chamber_inlet_port.m_flow = snapshot.line_to_chamber_flow
chamber_outlet_port.m_flow = -snapshot.outlet_flow
outlet_volume_port = self._port(
outlet_orifice,
spec.outlet_orifice_volume_port,
)
outlet_boundary_port = self._port(
outlet_orifice,
spec.outlet_orifice_boundary_port,
)
outlet_volume_port.p = snapshot.chamber.p
outlet_volume_port.m_flow = snapshot.outlet_flow
outlet_volume_port.h_outflow = snapshot.chamber.h
outlet_boundary_port.p = snapshot.outlet_boundary.p
outlet_boundary_port.m_flow = -snapshot.outlet_flow
outlet_boundary_port.h_outflow = snapshot.outlet_boundary.h
def rhs(self, state_vector: list[float]) -> list[float]:
snapshot = self.snapshot(state_vector)
line_derivative = self.components.inlet_line.derivatives_from_connections(
port_1_m_flow=snapshot.node_to_line_flow,
connected_h_1=snapshot.inlet_node.h,
port_2_m_flow=-snapshot.line_to_chamber_flow,
connected_h_2=snapshot.chamber.h,
)
chamber = self.components.volume
spec = self.components.spec
chamber_derivative = chamber.derivatives_from_two_connections(
port_a_m_flow=self._port(chamber, "port_1").m_flow,
connected_h_a=(
snapshot.inlet_line.h
if spec.volume_inlet_port == "port_1"
else snapshot.outlet_boundary.h
),
port_b_m_flow=self._port(chamber, "port_2").m_flow,
connected_h_b=(
snapshot.inlet_line.h
if spec.volume_inlet_port == "port_2"
else snapshot.outlet_boundary.h
),
internal_h=snapshot.chamber.h,
)
return [*line_derivative.as_vector(), *chamber_derivative.as_vector()]