补充test_mql P4邻接拓扑解析

PythonModels/systems/test_mql.py

@@ -3975,6 +3975,31 @@ class TestMqlSystem:
         spec = self._pnl0001_spec_for_node_port(node_alias, port_name)
         return self.pnl0001_assembly.lines[spec.alias]
 
+    def _pnl0002_connection_for_node_port(self, node_alias: str, port_name: str):
+        from PythonModels.systems.test_mql_nodes import TestMqlP4NodePortConnection
+
+        spec = self._pnl0002_spec_for_node_port(node_alias, port_name)
+        if spec.source_component == node_alias and spec.source_port == port_name:
+            remote_node_alias = spec.target_component
+            remote_port = spec.target_port
+        elif spec.target_component == node_alias and spec.target_port == port_name:
+            remote_node_alias = spec.source_component
+            remote_port = spec.source_port
+        else:
+            raise KeyError(f"No PNL0002 line attached to {node_alias}.{port_name}")
+        if remote_node_alias not in self.node4_assembly:
+            raise KeyError(
+                f"PNL0002 line {spec.alias} remote endpoint is not a P4 node: "
+                f"{remote_node_alias}.{remote_port}"
+            )
+        return TestMqlP4NodePortConnection(
+            line_alias=spec.alias,
+            local_node_alias=node_alias,
+            local_port=port_name,
+            remote_node_alias=remote_node_alias,
+            remote_port=remote_port,
+        )
+
     def _pnl0002_spec_for_node_port(self, node_alias: str, port_name: str):
         for spec in self.pnl0002_assembly.specs:
             if spec.source_component == node_alias and spec.source_port == port_name:
@@ -3987,6 +4012,65 @@ class TestMqlSystem:
         spec = self._pnl0002_spec_for_node_port(node_alias, port_name)
         return self.pnl0002_assembly.lines[spec.alias]
 
+    def _p4_primary_connection_for_node(self, node_alias: str):
+        from PythonModels.systems.test_mql_nodes import TestMqlP4NodePrimaryConnection
+
+        spec = self._pnl0001_spec_for_node_port(node_alias, "port_2")
+        if spec.source_component == node_alias and spec.source_port == "port_2":
+            chamber_alias = spec.target_component
+            chamber_port = spec.target_port
+        elif spec.target_component == node_alias and spec.target_port == "port_2":
+            chamber_alias = spec.source_component
+            chamber_port = spec.source_port
+        else:
+            raise KeyError(f"No PNL0001 primary line attached to {node_alias}.port_2")
+        return TestMqlP4NodePrimaryConnection(
+            line_alias=spec.alias,
+            node_alias=node_alias,
+            node_port="port_2",
+            chamber_alias=chamber_alias,
+            chamber_port=chamber_port,
+        )
+
+    def _p4_port4_orifice_connection_for_node(self, node_alias: str):
+        from PythonModels.systems.test_mql_nodes import TestMqlP4NodeOrificeConnection
+
+        for spec in CONNECTION_SPECS:
+            if spec["submodel"] != "DIRECT":
+                continue
+            if spec["target_component"] == node_alias and spec["target_port"] == "port_4":
+                orifice_alias = str(spec["source_component"])
+                if orifice_alias in self.pneumatic_assembly.variable_orifices:
+                    return TestMqlP4NodeOrificeConnection(
+                        orifice_alias=orifice_alias,
+                        node_alias=node_alias,
+                        node_port="port_4",
+                        direct_line_alias=str(spec["alias"]),
+                    )
+            if spec["source_component"] == node_alias and spec["source_port"] == "port_4":
+                orifice_alias = str(spec["target_component"])
+                if orifice_alias in self.pneumatic_assembly.variable_orifices:
+                    return TestMqlP4NodeOrificeConnection(
+                        orifice_alias=orifice_alias,
+                        node_alias=node_alias,
+                        node_port="port_4",
+                        direct_line_alias=str(spec["alias"]),
+                    )
+        raise KeyError(f"No PNVO orifice attached to {node_alias}.port_4")
+
+    def p4_node_neighborhood(self, node_alias: str):
+        from PythonModels.systems.test_mql_nodes import TestMqlP4NodeNeighborhood
+
+        if node_alias not in self.node4_assembly:
+            raise KeyError(f"Unknown P4 node: {node_alias}")
+        return TestMqlP4NodeNeighborhood(
+            node_alias=node_alias,
+            primary=self._p4_primary_connection_for_node(node_alias),
+            port_1=self._pnl0002_connection_for_node_port(node_alias, "port_1"),
+            port_3=self._pnl0002_connection_for_node_port(node_alias, "port_3"),
+            port_4=self._p4_port4_orifice_connection_for_node(node_alias),
+        )
+
     def _p4_node_alias_for_orifice_port_2(self, orifice_alias: str) -> str:
         for spec in CONNECTION_SPECS:
             if spec["submodel"] != "DIRECT":

PythonModels/systems/test_mql_nodes.py

@@ -165,6 +165,41 @@ class TestMqlPneumaticNode4:
         )
 
 
+@dataclass(frozen=True)
+class TestMqlP4NodePortConnection:
+    line_alias: str
+    local_node_alias: str
+    local_port: str
+    remote_node_alias: str
+    remote_port: str
+
+
+@dataclass(frozen=True)
+class TestMqlP4NodePrimaryConnection:
+    line_alias: str
+    node_alias: str
+    node_port: str
+    chamber_alias: str
+    chamber_port: str
+
+
+@dataclass(frozen=True)
+class TestMqlP4NodeOrificeConnection:
+    orifice_alias: str
+    node_alias: str
+    node_port: str
+    direct_line_alias: str
+
+
+@dataclass(frozen=True)
+class TestMqlP4NodeNeighborhood:
+    node_alias: str
+    primary: TestMqlP4NodePrimaryConnection
+    port_1: TestMqlP4NodePortConnection
+    port_3: TestMqlP4NodePortConnection
+    port_4: TestMqlP4NodeOrificeConnection
+
+
 def build_test_mql_node3_assembly() -> dict[str, TestMqlPneumaticNode3]:
     return {
         str(spec["alias"]): TestMqlPneumaticNode3(alias=str(spec["alias"]))

tests/test_test_mql_nodes.py

@@ -99,6 +99,78 @@ class TestMqlPneumaticNode4Tests(unittest.TestCase):
         self.assertEqual(system.typed_node4_count, 8)
         self.assertEqual(system.node4_assembly["pnnode4_17"].alias, "pnnode4_17")
 
+    def test_pnnode4_17_neighborhood_maps_adjacent_p4_nodes(self) -> None:
+        system = TestMqlSystem()
+        neighborhood = system.p4_node_neighborhood("pnnode4_17")
+
+        self.assertEqual(neighborhood.node_alias, "pnnode4_17")
+        self.assertEqual(neighborhood.primary.line_alias, "pneumatic_68")
+        self.assertEqual(neighborhood.primary.chamber_alias, "pn_c1_9")
+        self.assertEqual(neighborhood.port_1.line_alias, "pneumatic_86")
+        self.assertEqual(neighborhood.port_1.remote_node_alias, "pnnode4_18")
+        self.assertEqual(neighborhood.port_1.remote_port, "port_3")
+        self.assertEqual(neighborhood.port_3.line_alias, "pneumatic_83")
+        self.assertEqual(neighborhood.port_3.remote_node_alias, "pnnode4_16")
+        self.assertEqual(neighborhood.port_3.remote_port, "port_1")
+        self.assertEqual(neighborhood.port_4.orifice_alias, "pn_morifice_9")
+        self.assertEqual(neighborhood.port_4.direct_line_alias, "pneumatic_90")
+
+    def test_adjacent_p4_node_primary_pressures_match_their_pnl0001_states(self) -> None:
+        index = 500
+
+        self.assertAlmostEqual(
+            self.results.series("press1@pnnode4_18")[index],
+            self.results.series("p2@pneumatic_66")[index],
+        )
+        self.assertAlmostEqual(
+            self.results.series("press1@pnnode4_16")[index],
+            self.results.series("p2@pneumatic_69")[index],
+        )
+
+    def test_adjacent_p4_node_baselines_match_adjacent_flow_sums(self) -> None:
+        index = 500
+        pnnode4_18 = self.nodes["pnnode4_18"].balance(
+            port_2_temperature_k=self.results.series("temp1@pnnode4_18")[index],
+            port_2_pressure_pa=(
+                self.results.series("press1@pnnode4_18")[index] + 101_300.0
+            ),
+            port_1_enthalpy_flow_w=self.results.series("dh1@pneumatic_80")[index],
+            port_1_mass_flow_g_s=self.results.series("dm1@pneumatic_80")[index],
+            port_3_enthalpy_flow_w=self.results.series("dh2@pneumatic_86")[index],
+            port_3_mass_flow_g_s=self.results.series("dm2@pneumatic_86")[index],
+            port_4_enthalpy_flow_w=self.results.series("dh2@pn_morifice_10")[index],
+            port_4_mass_flow_g_s=self.results.series("dm2@pn_morifice_10")[index],
+        )
+        pnnode4_16 = self.nodes["pnnode4_16"].balance(
+            port_2_temperature_k=self.results.series("temp1@pnnode4_16")[index],
+            port_2_pressure_pa=(
+                self.results.series("press1@pnnode4_16")[index] + 101_300.0
+            ),
+            port_1_enthalpy_flow_w=self.results.series("dh1@pneumatic_83")[index],
+            port_1_mass_flow_g_s=self.results.series("dm1@pneumatic_83")[index],
+            port_3_enthalpy_flow_w=self.results.series("dh2@pneumatic_95")[index],
+            port_3_mass_flow_g_s=self.results.series("dm2@pneumatic_95")[index],
+            port_4_enthalpy_flow_w=self.results.series("dh2@pn_morifice_1")[index],
+            port_4_mass_flow_g_s=self.results.series("dm2@pn_morifice_1")[index],
+        )
+
+        self.assertAlmostEqual(
+            pnnode4_18.port_2_mass_flow_g_s,
+            self.results.series("dm2@pnnode4_18")[index],
+        )
+        self.assertAlmostEqual(
+            pnnode4_18.port_2_enthalpy_flow_w,
+            self.results.series("dh2@pnnode4_18")[index],
+        )
+        self.assertAlmostEqual(
+            pnnode4_16.port_2_mass_flow_g_s,
+            self.results.series("dm2@pnnode4_16")[index],
+        )
+        self.assertAlmostEqual(
+            pnnode4_16.port_2_enthalpy_flow_w,
+            self.results.series("dh2@pnnode4_16")[index],
+        )
+
     def test_port_two_flow_is_exact_sum_of_ports_one_three_and_four(self) -> None:
         balance = self.nodes["pnnode4_17"].balance(
             port_2_temperature_k=281.0,