接入test_mql PN3节点管路闭合

PythonModels/systems/test_mql.py

@@ -3940,6 +3940,22 @@ class TestMqlSystem:
     def typed_node3_count(self) -> int:
         return len(self.node3_assembly)
 
+    def _pnl0003_line_for_node_port(self, node_alias: str, port_name: str):
+        for spec in self.pnl0003_assembly.specs:
+            if spec.source_component == node_alias and spec.source_port == port_name:
+                return self.pnl0003_assembly.lines[spec.alias]
+            if spec.target_component == node_alias and spec.target_port == port_name:
+                return self.pnl0003_assembly.lines[spec.alias]
+        raise KeyError(f"No PNL0003 line attached to {node_alias}.{port_name}")
+
+    def _pnl00r_line_for_node_port(self, node_alias: str, port_name: str):
+        for spec in self.pnl00r_assembly.specs:
+            if spec.source_component == node_alias and spec.source_port == port_name:
+                return self.pnl00r_assembly.lines[spec.alias]
+            if spec.target_component == node_alias and spec.target_port == port_name:
+                return self.pnl00r_assembly.lines[spec.alias]
+        raise KeyError(f"No PNL00R line attached to {node_alias}.{port_name}")
+
     def pneumatic_state_vector(self) -> list[float]:
         return self.network.initial_state_vector()
 
@@ -4276,6 +4292,110 @@ class TestMqlSystem:
             t_eval=t_eval,
         )
 
+    def pn3_node_chamber_segment_closure_from_spec(
+        self,
+        spec,
+        *,
+        inlet_node_pressure_pa: float,
+        resistance_boundary_pressure_pa: float,
+        inlet_node_temperature_k: float = 293.15,
+        resistance_boundary_temperature_k: float = 293.15,
+        pnl0003_port_2_mass_flow_kg_s: float = 0.0,
+        pnl0003_port_2_temperature_k: float = 293.15,
+    ):
+        """Close a PNL0001 chamber segment with the real outlet PN3 node.
+
+        The PN3 primary pressure/temperature is taken from the attached PNL0003
+        port-1 compliance. The PNL0003 port-2 side and the far side of PNL00R
+        remain prescribed boundaries until PNVO/P4 semantics are inserted.
+        """
+        from PythonModels.components.amesim_pneumatic import (
+            AmesimPneumaticOrifice,
+            AmesimPneumaticVolume,
+        )
+        from PythonModels.systems.test_mql_closure import (
+            TestMqlPneumaticBoundaryCondition,
+            TestMqlPneumaticFlowBoundaryCondition,
+            TestMqlPn3NodeChamberSegmentClosure,
+            TestMqlPn3NodeChamberSegmentComponents,
+        )
+
+        inlet_line = self.pnl0001_assembly.lines[spec.inlet_line_alias]
+        outlet_line = self.pnl0001_assembly.lines[spec.outlet_line_alias]
+        node_line = self._pnl0003_line_for_node_port(spec.outlet_node_alias, "port_2")
+        node_resistance = self._pnl00r_line_for_node_port(
+            spec.outlet_node_alias, "port_1"
+        )
+        node = self.node3_assembly[spec.outlet_node_alias]
+        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 TestMqlPn3NodeChamberSegmentClosure(
+            components=TestMqlPn3NodeChamberSegmentComponents(
+                inlet_line=inlet_line,
+                outlet_line=outlet_line,
+                node_line=node_line,
+                node_resistance=node_resistance,
+                node=node,
+                volume=volume,
+                inlet_orifice=inlet_orifice,
+                outlet_orifice=outlet_orifice,
+                spec=spec,
+            ),
+            inlet_node=TestMqlPneumaticBoundaryCondition(
+                pressure_pa=inlet_node_pressure_pa,
+                temperature_k=inlet_node_temperature_k,
+            ),
+            resistance_boundary=TestMqlPneumaticBoundaryCondition(
+                pressure_pa=resistance_boundary_pressure_pa,
+                temperature_k=resistance_boundary_temperature_k,
+            ),
+            pnl0003_port_2_flow=TestMqlPneumaticFlowBoundaryCondition(
+                mass_flow_kg_s=pnl0003_port_2_mass_flow_kg_s,
+                temperature_k=pnl0003_port_2_temperature_k,
+            ),
+        )
+
+    def simulate_pn3_node_chamber_segment_from_spec(
+        self,
+        spec,
+        *,
+        inlet_node_pressure_pa: float,
+        resistance_boundary_pressure_pa: float,
+        inlet_node_temperature_k: float = 293.15,
+        resistance_boundary_temperature_k: float = 293.15,
+        pnl0003_port_2_mass_flow_kg_s: float = 0.0,
+        pnl0003_port_2_temperature_k: float = 293.15,
+        config: SolveIVPConfig | None = None,
+        t_eval: list[float] | None = None,
+    ):
+        closure = self.pn3_node_chamber_segment_closure_from_spec(
+            spec,
+            inlet_node_pressure_pa=inlet_node_pressure_pa,
+            resistance_boundary_pressure_pa=resistance_boundary_pressure_pa,
+            inlet_node_temperature_k=inlet_node_temperature_k,
+            resistance_boundary_temperature_k=resistance_boundary_temperature_k,
+            pnl0003_port_2_mass_flow_kg_s=pnl0003_port_2_mass_flow_kg_s,
+            pnl0003_port_2_temperature_k=pnl0003_port_2_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,