diff --git a/PythonModels/systems/test_mql.py b/PythonModels/systems/test_mql.py index 5fce3ae..f9d6bdc 100644 --- a/PythonModels/systems/test_mql.py +++ b/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, diff --git a/tests/test_test_mql_branch_simulation.py b/tests/test_test_mql_branch_simulation.py new file mode 100644 index 0000000..ea88c9d --- /dev/null +++ b/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()