补充test_mql气动支路仿真入口

PythonModels/systems/test_mql.py

@@ -6,6 +6,7 @@ from types import SimpleNamespace
 
 from PythonModels.core.base import Component
 from PythonModels.core.network import SimulationNetwork
+from PythonModels.core.solver import SolveIVPConfig, integrate_ode
 
 
 MODEL_NAME = "test_mql"
@@ -3945,6 +3946,72 @@ class TestMqlSystem:
             apply_state_vector=apply_state_vector,
         )
 
+    def simulate_pneumatic_branch(
+        self,
+        *,
+        name: str,
+        upstream_volume_alias: str,
+        orifice_alias: str,
+        downstream_volume_alias: str,
+        config: SolveIVPConfig | None = None,
+        t_eval: list[float] | None = None,
+    ):
+        closure = self.pneumatic_branch_closure(
+            name=name,
+            upstream_volume_alias=upstream_volume_alias,
+            orifice_alias=orifice_alias,
+            downstream_volume_alias=downstream_volume_alias,
+        )
+        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 evaluate_pneumatic_branch_solution(
+        self,
+        solution: object,
+        *,
+        name: str,
+        upstream_volume_alias: str,
+        orifice_alias: str,
+        downstream_volume_alias: str,
+    ) -> dict[str, list[float]]:
+        closure = self.pneumatic_branch_closure(
+            name=name,
+            upstream_volume_alias=upstream_volume_alias,
+            orifice_alias=orifice_alias,
+            downstream_volume_alias=downstream_volume_alias,
+        )
+        series = {
+            "time": [],
+            f"{name}.flow": [],
+            f"{upstream_volume_alias}.p": [],
+            f"{upstream_volume_alias}.T": [],
+            f"{upstream_volume_alias}.m_flow": [],
+            f"{downstream_volume_alias}.p": [],
+            f"{downstream_volume_alias}.T": [],
+            f"{downstream_volume_alias}.m_flow": [],
+        }
+        for index, time_value in enumerate(solution.t):
+            state_vector = [row[index] for row in solution.y]
+            snapshot = closure.snapshot(state_vector)
+            series["time"].append(float(time_value))
+            series[f"{name}.flow"].append(snapshot.flow)
+            series[f"{upstream_volume_alias}.p"].append(snapshot.upstream.p)
+            series[f"{upstream_volume_alias}.T"].append(snapshot.upstream.T)
+            series[f"{upstream_volume_alias}.m_flow"].append(
+                closure.components.upstream_volume.port_a.m_flow
+            )
+            series[f"{downstream_volume_alias}.p"].append(snapshot.downstream.p)
+            series[f"{downstream_volume_alias}.T"].append(snapshot.downstream.T)
+            series[f"{downstream_volume_alias}.m_flow"].append(
+                closure.components.downstream_volume.port_a.m_flow
+            )
+        return series
+
     def snapshot(self) -> TestMqlSnapshot:
         return TestMqlSnapshot(
             model_name=MODEL_NAME,

tests/test_test_mql_branch_simulation.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import unittest
+
+from PythonModels.core.solver import SolveIVPConfig
+from PythonModels.systems.test_mql import TestMqlSystem
+
+
+class TestMqlPneumaticBranchSimulationTests(unittest.TestCase):
+    def test_simulate_pneumatic_branch_returns_ode_solution(self) -> None:
+        system = TestMqlSystem()
+
+        solution = system.simulate_pneumatic_branch(
+            name="sample_branch",
+            upstream_volume_alias="pn_general_chamber",
+            orifice_alias="pn_orifice_18",
+            downstream_volume_alias="pn_c1_8",
+            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), 4)
+        self.assertEqual([len(row) for row in solution.y], [3, 3, 3, 3])
+
+    def test_evaluate_pneumatic_branch_solution_reports_branch_series(self) -> None:
+        system = TestMqlSystem()
+        solution = system.simulate_pneumatic_branch(
+            name="sample_branch",
+            upstream_volume_alias="pn_general_chamber",
+            orifice_alias="pn_orifice_18",
+            downstream_volume_alias="pn_c1_8",
+            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],
+        )
+
+        series = system.evaluate_pneumatic_branch_solution(
+            solution,
+            name="sample_branch",
+            upstream_volume_alias="pn_general_chamber",
+            orifice_alias="pn_orifice_18",
+            downstream_volume_alias="pn_c1_8",
+        )
+
+        self.assertEqual(series["time"], [0.0, 5.0e-6, 1.0e-5])
+        self.assertIn("sample_branch.flow", series)
+        self.assertIn("pn_general_chamber.p", series)
+        self.assertIn("pn_c1_8.p", series)
+        self.assertEqual(len(series["sample_branch.flow"]), 3)
+        self.assertGreater(series["sample_branch.flow"][0], 0.0)
+        self.assertGreater(series["pn_general_chamber.p"][0], series["pn_general_chamber.p"][-1])
+        self.assertLess(series["pn_c1_8.p"][0], series["pn_c1_8.p"][-1])
+        self.assertAlmostEqual(
+            series["pn_general_chamber.m_flow"][0],
+            -series["sample_branch.flow"][0],
+        )
+        self.assertAlmostEqual(
+            series["pn_c1_8.m_flow"][0],
+            series["sample_branch.flow"][0],
+        )
+
+    def test_simulate_pneumatic_branch_rejects_invalid_component_aliases(self) -> None:
+        system = TestMqlSystem()
+
+        with self.assertRaises(TypeError):
+            system.simulate_pneumatic_branch(
+                name="invalid_branch",
+                upstream_volume_alias="pn_brp2_8",
+                orifice_alias="pn_orifice_18",
+                downstream_volume_alias="pn_c1_8",
+                config=SolveIVPConfig(t_start=0.0, t_stop=0.0),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()