diff --git a/PythonModels/systems/test_mql.py b/PythonModels/systems/test_mql.py index 62f5af0..5dd6944 100644 --- a/PythonModels/systems/test_mql.py +++ b/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": diff --git a/PythonModels/systems/test_mql_nodes.py b/PythonModels/systems/test_mql_nodes.py index 596e8d0..1227abc 100644 --- a/PythonModels/systems/test_mql_nodes.py +++ b/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"])) diff --git a/tests/test_test_mql_nodes.py b/tests/test_test_mql_nodes.py index e1df079..f1e20b3 100644 --- a/tests/test_test_mql_nodes.py +++ b/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,