diff --git a/PythonModels/components/amesim_pneumatic.py b/PythonModels/components/amesim_pneumatic.py
index 7e9cc99..de110bf 100644
--- a/PythonModels/components/amesim_pneumatic.py
+++ b/PythonModels/components/amesim_pneumatic.py
@@ -93,6 +93,7 @@ class AmesimPneumaticVolume(DynamicComponent):
U0 = m0 * gas.specific_internal_energy(T0)
self.state = VolumeState(m=m0, U=U0)
self.port_a = PortState()
+ self.port_b = PortState()
@classmethod
def from_liters(
@@ -130,11 +131,37 @@ class AmesimPneumaticVolume(DynamicComponent):
h = self.gas.specific_enthalpy(T)
self.port_a.p = p
self.port_a.h_outflow = h
+ self.port_b.p = p
+ self.port_b.h_outflow = h
return ThermodynamicProperties(p=p, T=T, rho=rho, u=u, h=h)
def derivatives(self, inlet_h: float, m_flow: float) -> VolumeState:
return VolumeState(m=m_flow, U=m_flow * inlet_h)
+ def derivatives_from_two_connections(
+ self,
+ *,
+ port_a_m_flow: float,
+ connected_h_a: float,
+ port_b_m_flow: float,
+ connected_h_b: float,
+ internal_h: float,
+ ) -> VolumeState:
+ inlet_h_a = self.connection_inlet_enthalpy(
+ port_m_flow=port_a_m_flow,
+ connected_h=connected_h_a,
+ internal_h=internal_h,
+ )
+ inlet_h_b = self.connection_inlet_enthalpy(
+ port_m_flow=port_b_m_flow,
+ connected_h=connected_h_b,
+ internal_h=internal_h,
+ )
+ return VolumeState(
+ m=port_a_m_flow + port_b_m_flow,
+ U=port_a_m_flow * inlet_h_a + port_b_m_flow * inlet_h_b,
+ )
+
class AmesimVariablePneumaticVolume(AmesimPneumaticVolume):
"""PNCH012-style volume with a dead volume plus an external moving volume."""
diff --git a/PythonModels/systems/test_mql.py b/PythonModels/systems/test_mql.py
index b1a1572..55dd178 100644
--- a/PythonModels/systems/test_mql.py
+++ b/PythonModels/systems/test_mql.py
@@ -3920,8 +3920,23 @@ class TestMqlSystem:
TestMqlPneumaticTopologyCandidate,
TestMqlPneumaticTopologyDiscovery,
)
+ from PythonModels.systems.test_mql_pneumatic_topology import (
+ discover_fixed_chamber_segments,
+ )
+ from PythonModels.systems.test_mql_topology import load_test_mql_cir_topology
+
+ topology = load_test_mql_cir_topology(self.archive_path)
+ chamber_segment_specs = discover_fixed_chamber_segments(
+ topology,
+ COMPONENT_SPECS,
+ CONNECTION_SPECS,
+ )
typed_aliases = set(self._pneumatic_components_by_alias())
+ component_submodels = {
+ str(component["alias"]): str(component["submodel"])
+ for component in COMPONENT_SPECS
+ }
blocked_candidates = []
for spec in CONNECTION_SPECS:
source_component = str(spec["source_component"])
@@ -3932,12 +3947,28 @@ class TestMqlSystem:
continue
line_submodel = str(spec["submodel"])
- if source_is_typed and target_is_typed:
- reason = "direct typed pneumatic connection is not a complete volume-orifice-volume branch"
- elif line_submodel == "DIRECT":
- reason = "requires AMESim boundary component interpretation before closure construction"
+ endpoint_submodels = {
+ component_submodels.get(source_component, ""),
+ component_submodels.get(target_component, ""),
+ }
+ if "PNRP17" in endpoint_submodels:
+ category = "piston-boundary"
+ reason = "requires pneumatic piston and mechanical coupling interpretation"
+ elif "STEP0" in endpoint_submodels:
+ category = "control-boundary"
+ reason = "requires modulated-orifice control signal evaluation"
+ elif endpoint_submodels & {"PN3NODE2", "P4NODE2"}:
+ category = "node-line-boundary"
+ reason = "requires AMESim node/line equations before full-network closure"
+ elif source_is_typed and target_is_typed:
+ category = "typed-direct"
+ reason = "direct typed connection is not a complete simulated subnetwork"
+ elif line_submodel != "DIRECT":
+ category = "line-boundary"
+ reason = "requires AMESim line equations before full-network closure"
else:
- reason = "requires AMESim node/line submodel interpretation before closure construction"
+ category = "component-boundary"
+ reason = "requires AMESim boundary component interpretation"
blocked_candidates.append(
TestMqlPneumaticTopologyCandidate(
@@ -3949,15 +3980,97 @@ class TestMqlSystem:
target_port=str(spec["target_port"]),
source_is_typed_pneumatic=source_is_typed,
target_is_typed_pneumatic=target_is_typed,
+ category=category,
reason=reason,
)
)
return TestMqlPneumaticTopologyDiscovery(
branch_specs=(),
+ chamber_segment_specs=chamber_segment_specs,
blocked_candidates=tuple(blocked_candidates),
)
+ def pneumatic_chamber_segment_closure_from_spec(
+ self,
+ spec,
+ *,
+ inlet_pressure_pa: float,
+ outlet_pressure_pa: float,
+ inlet_temperature_k: float = 293.15,
+ outlet_temperature_k: float = 293.15,
+ ):
+ """Build a reduced segment using absolute-pressure line boundaries.
+
+ Boundary values apply at the PNL0001/orifice interfaces, not at the nodes.
+ """
+ from PythonModels.components.amesim_pneumatic import (
+ AmesimPneumaticOrifice,
+ AmesimPneumaticVolume,
+ )
+ from PythonModels.systems.test_mql_closure import (
+ TestMqlPneumaticBoundaryCondition,
+ TestMqlPneumaticChamberSegmentClosure,
+ TestMqlPneumaticChamberSegmentComponents,
+ )
+
+ volume = self.network.components[spec.volume_alias]
+ inlet_orifice = self.network.components[spec.inlet_orifice_alias]
+ outlet_orifice = self.network.components[spec.outlet_orifice_alias]
+ if not isinstance(volume, AmesimPneumaticVolume):
+ raise TypeError(f"{spec.volume_alias} is not an AMESim pneumatic volume")
+ if not isinstance(inlet_orifice, AmesimPneumaticOrifice):
+ raise TypeError(
+ f"{spec.inlet_orifice_alias} is not an AMESim pneumatic orifice"
+ )
+ if not isinstance(outlet_orifice, AmesimPneumaticOrifice):
+ raise TypeError(
+ f"{spec.outlet_orifice_alias} is not an AMESim pneumatic orifice"
+ )
+ return TestMqlPneumaticChamberSegmentClosure(
+ components=TestMqlPneumaticChamberSegmentComponents(
+ volume=volume,
+ inlet_orifice=inlet_orifice,
+ outlet_orifice=outlet_orifice,
+ spec=spec,
+ ),
+ inlet_boundary=TestMqlPneumaticBoundaryCondition(
+ pressure_pa=inlet_pressure_pa,
+ temperature_k=inlet_temperature_k,
+ ),
+ outlet_boundary=TestMqlPneumaticBoundaryCondition(
+ pressure_pa=outlet_pressure_pa,
+ temperature_k=outlet_temperature_k,
+ ),
+ )
+
+ def simulate_pneumatic_chamber_segment_from_spec(
+ self,
+ spec,
+ *,
+ inlet_pressure_pa: float,
+ outlet_pressure_pa: float,
+ inlet_temperature_k: float = 293.15,
+ outlet_temperature_k: float = 293.15,
+ config: SolveIVPConfig | None = None,
+ t_eval: list[float] | None = None,
+ ):
+ """Integrate a reduced segment with absolute-pressure line boundaries."""
+ closure = self.pneumatic_chamber_segment_closure_from_spec(
+ spec,
+ inlet_pressure_pa=inlet_pressure_pa,
+ outlet_pressure_pa=outlet_pressure_pa,
+ inlet_temperature_k=inlet_temperature_k,
+ outlet_temperature_k=outlet_temperature_k,
+ )
+ run_config = config or SolveIVPConfig(t_stop=1.0e-4, max_step=1.0e-5)
+ return integrate_ode(
+ rhs=lambda t, state: closure.rhs(state),
+ initial_state=closure.initial_state_vector(),
+ config=run_config,
+ t_eval=t_eval,
+ )
+
def pneumatic_branch_closure_from_spec(self, spec):
return self.pneumatic_branch_closure(
name=spec.name,
diff --git a/PythonModels/systems/test_mql_closure.py b/PythonModels/systems/test_mql_closure.py
index d7c0488..9a5651b 100644
--- a/PythonModels/systems/test_mql_closure.py
+++ b/PythonModels/systems/test_mql_closure.py
@@ -4,10 +4,12 @@ from dataclasses import dataclass
from typing import Callable
from PythonModels.components.amesim_pneumatic import (
+ AmesimPneumaticGas,
AmesimPneumaticOrifice,
AmesimPneumaticVolume,
)
from PythonModels.core.medium import ThermodynamicProperties
+from PythonModels.core.ports import PortState
from PythonModels.core.state import VolumeState
@@ -30,15 +32,72 @@ class TestMqlPneumaticTopologyCandidate:
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 TestMqlPneumaticBranchComponents:
name: str
@@ -184,3 +243,139 @@ class TestMqlPneumaticClosure:
*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()
diff --git a/PythonModels/systems/test_mql_pneumatic_topology.py b/PythonModels/systems/test_mql_pneumatic_topology.py
new file mode 100644
index 0000000..70babcc
--- /dev/null
+++ b/PythonModels/systems/test_mql_pneumatic_topology.py
@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from PythonModels.systems.test_mql_closure import TestMqlPneumaticChamberSegmentSpec
+from PythonModels.systems.test_mql_topology import TestMqlCirTopology
+
+
+def discover_fixed_chamber_segments(
+ topology: TestMqlCirTopology,
+ component_specs: list[dict[str, object]],
+ connection_specs: list[dict[str, object]],
+) -> tuple[TestMqlPneumaticChamberSegmentSpec, ...]:
+ submodel_by_alias = {
+ str(component["alias"]): str(component["submodel"])
+ for component in component_specs
+ }
+ segments = []
+ for component in component_specs:
+ volume_alias = str(component["alias"])
+ if component["submodel"] != "PNCH023":
+ continue
+
+ orifice_contacts = []
+ for contact in topology.contacts_for(volume_alias):
+ other_alias, other_port = contact.other_endpoint(volume_alias)
+ if submodel_by_alias.get(other_alias) == "PNOR001":
+ orifice_contacts.append(
+ (
+ other_alias,
+ other_port,
+ contact.port_for(volume_alias),
+ )
+ )
+ if len(orifice_contacts) != 2:
+ raise ValueError(
+ f"{volume_alias} must contact exactly two PNOR001 orifices; "
+ f"found {len(orifice_contacts)}"
+ )
+
+ sides = [
+ _resolve_orifice_boundary(
+ orifice_alias=orifice_alias,
+ orifice_volume_port=orifice_volume_port,
+ volume_port=volume_port,
+ connection_specs=connection_specs,
+ submodel_by_alias=submodel_by_alias,
+ )
+ for orifice_alias, orifice_volume_port, volume_port in orifice_contacts
+ ]
+ inlet_sides = [side for side in sides if side["role"] == "inlet"]
+ outlet_sides = [side for side in sides if side["role"] == "outlet"]
+ if len(inlet_sides) != 1 or len(outlet_sides) != 1:
+ raise ValueError(
+ f"{volume_alias} requires one inlet and one outlet topology side"
+ )
+ inlet = inlet_sides[0]
+ outlet = outlet_sides[0]
+ segments.append(
+ TestMqlPneumaticChamberSegmentSpec(
+ name=f"{volume_alias}_segment",
+ inlet_node_alias=inlet["node_alias"],
+ inlet_line_alias=inlet["line_alias"],
+ inlet_orifice_alias=inlet["orifice_alias"],
+ inlet_orifice_boundary_port=inlet["orifice_boundary_port"],
+ inlet_orifice_volume_port=inlet["orifice_volume_port"],
+ volume_alias=volume_alias,
+ volume_inlet_port=inlet["volume_port"],
+ volume_outlet_port=outlet["volume_port"],
+ outlet_orifice_alias=outlet["orifice_alias"],
+ outlet_orifice_volume_port=outlet["orifice_volume_port"],
+ outlet_orifice_boundary_port=outlet["orifice_boundary_port"],
+ outlet_line_alias=outlet["line_alias"],
+ outlet_node_alias=outlet["node_alias"],
+ )
+ )
+ return tuple(segments)
+
+
+def _resolve_orifice_boundary(
+ *,
+ orifice_alias: str,
+ orifice_volume_port: str,
+ volume_port: str,
+ connection_specs: list[dict[str, object]],
+ submodel_by_alias: dict[str, str],
+) -> dict[str, str]:
+ boundary_connections = []
+ for connection in connection_specs:
+ if (
+ connection["source_component"] == orifice_alias
+ and connection["source_port"] != orifice_volume_port
+ ) or (
+ connection["target_component"] == orifice_alias
+ and connection["target_port"] != orifice_volume_port
+ ):
+ boundary_connections.append(connection)
+ if len(boundary_connections) != 1:
+ raise ValueError(
+ f"{orifice_alias} must have exactly one non-volume boundary connection; "
+ f"found {len(boundary_connections)}"
+ )
+
+ connection = boundary_connections[0]
+ if connection["submodel"] != "PNL0001":
+ raise ValueError(
+ f"{orifice_alias} boundary must use PNL0001, got {connection['submodel']}"
+ )
+ if connection["target_component"] == orifice_alias:
+ role = "inlet"
+ node_alias = str(connection["source_component"])
+ orifice_boundary_port = str(connection["target_port"])
+ else:
+ role = "outlet"
+ node_alias = str(connection["target_component"])
+ orifice_boundary_port = str(connection["source_port"])
+ if submodel_by_alias.get(node_alias) != "PN3NODE2":
+ raise ValueError(
+ f"{orifice_alias} PNL0001 boundary must terminate at PN3NODE2, "
+ f"got {node_alias}"
+ )
+ return {
+ "role": role,
+ "node_alias": node_alias,
+ "line_alias": str(connection["alias"]),
+ "orifice_alias": orifice_alias,
+ "orifice_boundary_port": orifice_boundary_port,
+ "orifice_volume_port": orifice_volume_port,
+ "volume_port": volume_port,
+ }
diff --git a/PythonModels/systems/test_mql_topology.py b/PythonModels/systems/test_mql_topology.py
new file mode 100644
index 0000000..4fac424
--- /dev/null
+++ b/PythonModels/systems/test_mql_topology.py
@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import re
+import tarfile
+import xml.etree.ElementTree as ET
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class TestMqlComponentContact:
+ component_a: str
+ port_a: str
+ component_b: str
+ port_b: str
+
+ def other_endpoint(self, component_alias: str) -> tuple[str, str]:
+ if component_alias == self.component_a:
+ return self.component_b, self.port_b
+ if component_alias == self.component_b:
+ return self.component_a, self.port_a
+ raise KeyError(component_alias)
+
+ def port_for(self, component_alias: str) -> str:
+ if component_alias == self.component_a:
+ return self.port_a
+ if component_alias == self.component_b:
+ return self.port_b
+ raise KeyError(component_alias)
+
+
+@dataclass(frozen=True)
+class TestMqlCirTopology:
+ component_contacts: tuple[TestMqlComponentContact, ...]
+
+ def contacts_for(self, component_alias: str) -> tuple[TestMqlComponentContact, ...]:
+ return tuple(
+ contact
+ for contact in self.component_contacts
+ if component_alias in (contact.component_a, contact.component_b)
+ )
+
+
+def load_test_mql_cir_topology(
+ archive_path: str | Path,
+ *,
+ cir_member: str = "test_mql_.cir",
+) -> TestMqlCirTopology:
+ with tarfile.open(archive_path) as archive:
+ cir_file = archive.extractfile(cir_member)
+ if cir_file is None:
+ raise ValueError(f"Missing AMESim circuit member: {cir_member}")
+ cir_text = cir_file.read().decode("latin1")
+
+ root = ET.fromstring(_topology_only_xml(cir_text))
+ components = root.findall(".//COMPS_LIST/COMP")
+ aliases = tuple(_required_text(component, "ALIAS") for component in components)
+ contacts: dict[
+ tuple[tuple[int, int], tuple[int, int]],
+ TestMqlComponentContact,
+ ] = {}
+ directed_contacts: set[tuple[tuple[int, int], tuple[int, int]]] = set()
+
+ for component_index, component in enumerate(components):
+ ports = component.findall("./COMP_PORTS_LIST/COMP_PORT")
+ for port_index, port in enumerate(ports):
+ if port.findtext("PORT_CONNECT") != "1":
+ continue
+ for connection in port.findall("./CONNECT_LIST/CONNECT"):
+ target_index = int(_required_text(connection, "CONNECT_ENTITY_NUM"))
+ target_port_index = int(_required_text(connection, "CONNECT_ENTITY_PORT"))
+ if target_index < 0 or target_index >= len(components):
+ raise ValueError(f"Component contact references unknown entity {target_index}")
+ target_ports = components[target_index].findall("./COMP_PORTS_LIST/COMP_PORT")
+ if target_port_index < 0 or target_port_index >= len(target_ports):
+ raise ValueError(
+ f"Component contact references unknown port {target_port_index} "
+ f"on {aliases[target_index]}"
+ )
+
+ endpoint = (component_index, port_index)
+ target_endpoint = (target_index, target_port_index)
+ directed_contacts.add((endpoint, target_endpoint))
+ key = tuple(sorted((endpoint, target_endpoint)))
+ first, second = key
+ contacts[key] = TestMqlComponentContact(
+ component_a=aliases[first[0]],
+ port_a=f"port_{first[1] + 1}",
+ component_b=aliases[second[0]],
+ port_b=f"port_{second[1] + 1}",
+ )
+
+ for endpoint, target_endpoint in directed_contacts:
+ if (target_endpoint, endpoint) not in directed_contacts:
+ raise ValueError(
+ "AMESim component contact is not reciprocal: "
+ f"{endpoint} -> {target_endpoint}"
+ )
+
+ return TestMqlCirTopology(component_contacts=tuple(contacts.values()))
+
+
+def _topology_only_xml(cir_text: str) -> str:
+ # AMESim expressions inside SUBMODEL contain unescaped && and <= operators.
+ # Topology lives outside those blocks, so omit them before XML parsing.
+ return re.sub(
+ r".*?",
+ "",
+ cir_text,
+ flags=re.DOTALL,
+ )
+
+
+def _required_text(element: ET.Element, child_name: str) -> str:
+ value = element.findtext(child_name)
+ if value is None:
+ raise ValueError(f"Missing AMESim circuit element: {child_name}")
+ return value
diff --git a/tests/test_amesim_pneumatic_components.py b/tests/test_amesim_pneumatic_components.py
index 5f6ae5d..57c4a90 100644
--- a/tests/test_amesim_pneumatic_components.py
+++ b/tests/test_amesim_pneumatic_components.py
@@ -55,6 +55,30 @@ class AmesimPneumaticComponentsTest(unittest.TestCase):
)
self.assertGreater(volume.gas_mass_g(), 1300.0)
+ def test_volume_balances_two_connections_with_stream_enthalpy(self) -> None:
+ volume = AmesimPneumaticVolume.from_liters(
+ name="pn_general_chamber",
+ volume_liters=57.0,
+ p0=15.3e6,
+ T0=293.15,
+ )
+ internal_h = volume.properties().h
+
+ derivative = volume.derivatives_from_two_connections(
+ port_a_m_flow=0.2,
+ connected_h_a=internal_h + 1000.0,
+ port_b_m_flow=-0.1,
+ connected_h_b=internal_h - 1000.0,
+ internal_h=internal_h,
+ )
+
+ self.assertAlmostEqual(derivative.m, 0.1)
+ self.assertAlmostEqual(
+ derivative.U,
+ 0.2 * (internal_h + 1000.0) - 0.1 * internal_h,
+ )
+ self.assertAlmostEqual(volume.port_a.p, volume.port_b.p)
+
def test_variable_volume_tracks_external_volume(self) -> None:
volume = AmesimVariablePneumaticVolume.from_liters(
name="pn_c1_8",
diff --git a/tests/test_test_mql_chamber_segment.py b/tests/test_test_mql_chamber_segment.py
new file mode 100644
index 0000000..e722624
--- /dev/null
+++ b/tests/test_test_mql_chamber_segment.py
@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import unittest
+
+from PythonModels.core.solver import SolveIVPConfig
+from PythonModels.systems.test_mql import TestMqlSystem
+
+
+class TestMqlPneumaticChamberSegmentTests(unittest.TestCase):
+ def setUp(self) -> None:
+ self.system = TestMqlSystem()
+ self.discovery = self.system.discover_pneumatic_branch_topology()
+ self.spec = self.discovery.chamber_segment_specs[0]
+
+ def test_discovers_four_fixed_chamber_segments_from_real_topology(self) -> None:
+ self.assertEqual(len(self.discovery.chamber_segment_specs), 4)
+ self.assertEqual(self.spec.volume_alias, "pn_general_chamber")
+ self.assertEqual(self.spec.inlet_node_alias, "pn_node3_8")
+ self.assertEqual(self.spec.inlet_line_alias, "pneumatic_96")
+ self.assertEqual(self.spec.inlet_orifice_alias, "pn_orifice_18")
+ self.assertEqual(self.spec.volume_inlet_port, "port_2")
+ self.assertEqual(self.spec.outlet_orifice_alias, "pn_orifice_19")
+ self.assertEqual(self.spec.outlet_line_alias, "pneumatic_97")
+ self.assertEqual(self.spec.outlet_node_alias, "pn_node3_9")
+ self.assertEqual(self.spec.source, "amesim-cir-topology")
+
+ def test_segment_closure_writes_real_amesim_port_directions(self) -> None:
+ closure = self.system.pneumatic_chamber_segment_closure_from_spec(
+ self.spec,
+ inlet_pressure_pa=16.0e6,
+ outlet_pressure_pa=1.0e6,
+ )
+
+ snapshot = closure.snapshot()
+ rhs = closure.rhs(closure.initial_state_vector())
+
+ self.assertGreater(snapshot.inlet_flow, 0.0)
+ self.assertGreater(snapshot.outlet_flow, 0.0)
+ self.assertAlmostEqual(
+ closure.components.volume.port_b.m_flow,
+ snapshot.inlet_flow,
+ )
+ self.assertAlmostEqual(
+ closure.components.volume.port_a.m_flow,
+ -snapshot.outlet_flow,
+ )
+ self.assertAlmostEqual(
+ closure.components.inlet_orifice.port_b.m_flow,
+ snapshot.inlet_flow,
+ )
+ self.assertAlmostEqual(
+ closure.components.outlet_orifice.port_a.m_flow,
+ -snapshot.outlet_flow,
+ )
+ self.assertEqual(len(rhs), 2)
+ self.assertAlmostEqual(rhs[0], snapshot.inlet_flow - snapshot.outlet_flow)
+
+ def test_segment_closure_handles_reverse_boundary_pressures(self) -> None:
+ closure = self.system.pneumatic_chamber_segment_closure_from_spec(
+ self.spec,
+ inlet_pressure_pa=1.0e6,
+ outlet_pressure_pa=16.0e6,
+ )
+
+ snapshot = closure.snapshot()
+ rhs = closure.rhs(closure.initial_state_vector())
+
+ self.assertLess(snapshot.inlet_flow, 0.0)
+ self.assertLess(snapshot.outlet_flow, 0.0)
+ self.assertAlmostEqual(rhs[0], snapshot.inlet_flow - snapshot.outlet_flow)
+
+ def test_simulates_topology_derived_segment_with_prescribed_boundaries(self) -> None:
+ solution = self.system.simulate_pneumatic_chamber_segment_from_spec(
+ self.spec,
+ inlet_pressure_pa=16.0e6,
+ outlet_pressure_pa=14.0e6,
+ config=SolveIVPConfig(
+ t_start=0.0,
+ t_stop=1.0e-5,
+ max_step=1.0e-6,
+ ),
+ t_eval=[0.0, 5.0e-6, 1.0e-5],
+ )
+
+ self.assertTrue(solution.success)
+ self.assertEqual(len(solution.t), 3)
+ self.assertEqual(len(solution.y), 2)
+ self.assertEqual([len(row) for row in solution.y], [3, 3])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_test_mql_topology.py b/tests/test_test_mql_topology.py
new file mode 100644
index 0000000..fe3947d
--- /dev/null
+++ b/tests/test_test_mql_topology.py
@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+
+from PythonModels.systems.test_mql_topology import load_test_mql_cir_topology
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame"
+
+
+class TestMqlCirTopologyTests(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls) -> None:
+ cls.topology = load_test_mql_cir_topology(TEST_MQL_AME)
+
+ def test_recovers_direct_component_contacts_omitted_from_line_specs(self) -> None:
+ self.assertEqual(len(self.topology.component_contacts), 54)
+ contacts = self.topology.contacts_for("pn_general_chamber")
+
+ self.assertEqual(len(contacts), 2)
+ self.assertEqual(
+ {
+ (
+ contact.port_for("pn_general_chamber"),
+ *contact.other_endpoint("pn_general_chamber"),
+ )
+ for contact in contacts
+ },
+ {
+ ("port_1", "pn_orifice_19", "port_2"),
+ ("port_2", "pn_orifice_18", "port_1"),
+ },
+ )
+
+ def test_component_contact_lookup_rejects_unrelated_alias(self) -> None:
+ contact = self.topology.contacts_for("pn_general_chamber")[0]
+
+ with self.assertRaises(KeyError):
+ contact.other_endpoint("not-an-endpoint")
+
+
+if __name__ == "__main__":
+ unittest.main()