接入test_mql真实固定气室拓扑

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,

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()

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,
+    }

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"<SUBMODEL>.*?</SUBMODEL>",
+        "<SUBMODEL />",
+        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