from __future__ import annotations import unittest from app.simulation.components.amesim.flow.pipes import ( AmesimPnl0001, AmesimPnl00r, ) from app.simulation.components.amesim.junctions.nodes import AmesimPn3Node2 from app.simulation.components.amesim.media.mediums import AmesimHeliumPengRobinsonMedium from app.simulation.components.experimental.storage.cylinder import Cylinder from app.simulation.components.experimental.storage.tank import Tank from app.simulation.core.medium import IdealGasMedium from app.simulation.registry import COMPONENT_MODEL_REGISTRY from app.simulation.solvers.stream import StreamResolver from app.simulation.systems.network import SimulationNetwork class AmesimPnl00rComponentTests(unittest.TestCase): def setUp(self) -> None: self.medium = IdealGasMedium() def test_default_create_preserves_amesim_parameter_contract(self) -> None: pipe = COMPONENT_MODEL_REGISTRY["amesim_pnl00r"].create( "pnl_1", self.medium, {}, ) self.assertIsInstance(pipe, AmesimPnl00r) self.assertEqual(set(pipe.ports), {"port_1", "port_2"}) self.assertEqual(set(pipe.parameter_values), {"diam", "le", "rr", "gi"}) self.assertEqual(pipe.parameter_values["diam"], 0.01) self.assertEqual(pipe.gi, 0) def test_rejects_fractional_gas_index(self) -> None: with self.assertRaisesRegex(ValueError, "gi must be an integer"): COMPONENT_MODEL_REGISTRY["amesim_pnl00r"].create( "pnl_1", self.medium, {"gi": 1.5}, ) def test_initial_flow_temperature_reference_preserves_default_behavior(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium) initial_h = self.medium.specific_enthalpy(self.medium.T_ref) self.assertEqual( pipe._connected_h, {"port_1": initial_h, "port_2": initial_h}, ) self.assertAlmostEqual(pipe._port_temperature("port_1"), self.medium.T_ref) self.assertAlmostEqual(pipe._port_temperature("port_2"), self.medium.T_ref) forward = pipe.mass_flow(501000.0, 500000.0) reverse = pipe.mass_flow(500000.0, 501000.0) self.assertGreater(forward, 0.0) self.assertLess(reverse, 0.0) self.assertAlmostEqual(forward, -reverse, delta=abs(forward) * 0.02) def test_zero_pressure_drop_has_zero_flow_and_finite_results(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium) pipe.port_1.p = 100000.0 pipe.port_2.p = 100000.0 pipe.port_1.m_flow = 0.0 pipe.port_2.m_flow = 0.0 residuals = { residual.id.rsplit(":", 1)[1]: residual for residual in pipe.pressure_flow_equation_residuals() } self.assertEqual(pipe.mass_flow(100000.0, 100000.0), 0.0) self.assertEqual(residuals["mass_flow_balance"].value, 0.0) self.assertEqual(residuals["pressure_flow_relation"].value, 0.0) self.assertEqual(set(pipe.component_result_values()), {"re", "cm", "v", "ff"}) def test_mass_flow_is_bidirectional_by_pressure_order(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium, diam=0.014, le=1.0, rr=0.045 / 14.0) forward = pipe.mass_flow(501000.0, 500000.0) reverse = pipe.mass_flow(500000.0, 501000.0) self.assertGreater(forward, 0.0) self.assertLess(reverse, 0.0) self.assertAlmostEqual(forward, -reverse, delta=abs(forward) * 0.02) def test_mass_flow_is_continuous_across_former_relative_deadband(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium, diam=0.014, le=1.0, rr=0.045 / 14.0) base_pressure = 8.0e6 former_threshold = base_pressure * 1.0e-7 below = pipe.mass_flow(base_pressure + 0.99 * former_threshold, base_pressure) above = pipe.mass_flow(base_pressure + 1.01 * former_threshold, base_pressure) reverse = pipe.mass_flow(base_pressure, base_pressure + 0.99 * former_threshold) self.assertGreater(below, 0.0) self.assertGreater(above, below) self.assertLess(reverse, 0.0) self.assertAlmostEqual(below, -reverse, delta=abs(below) * 1.0e-9) self.assertLess(abs(above - below), abs(above) * 0.10) self.assertEqual(pipe.mass_flow(base_pressure, base_pressure), 0.0) self.assertGreater(pipe.mass_flow(15.3e6 + 0.1, 15.3e6), 0.0) def test_matches_amesim_pn2pipefr_laminar_baselines(self) -> None: medium = AmesimHeliumPengRobinsonMedium() pipe = AmesimPnl00r( "pnl_1", medium, diam=0.014, le=1.0, rr=0.045 / 14.0, ) samples = ( ( 13_887_929.5734, 288.959589373, 13_887_887.6633, 288.959239289, 1.79075554200e-4, 838.404977868, 3.32483767035e-6, 0.0521569680484, 0.076335424633, ), ( 13_093_276.1461, 285.548411921, 13_093_238.0341, 285.546325432, 1.40922897107e-4, 665.123640745, 3.09743309373e-6, 0.0429212663992, 0.0962227112065, ), ( 11_690_210.3512, 272.892693485, 11_690_180.6138, 272.889527291, 7.10081729578e-5, 345.615999811, 2.3706370264e-6, 0.0230685613948, 0.185176612295, ), ) for p_1, T_1, p_2, T_2, mass_flow, re, cm, velocity, ff in samples: with self.subTest(pressure=p_1): pipe.port_1.p = p_1 pipe.port_2.p = p_2 pipe.update_stream_outflows( { "port_1": medium.specific_enthalpy_at_pressure(p_1, T_1), "port_2": medium.specific_enthalpy_at_pressure(p_2, T_2), } ) actual_mass_flow = pipe.mass_flow(p_1, p_2) results = pipe.component_result_values() self.assertAlmostEqual(actual_mass_flow / mass_flow, 1.0, delta=0.005) self.assertAlmostEqual(results["re"] / re, 1.0, delta=0.01) self.assertAlmostEqual(results["cm"] / cm, 1.0, delta=1.0e-4) self.assertAlmostEqual(results["v"] / velocity, 1.0, delta=0.005) self.assertAlmostEqual(results["ff"] / ff, 1.0, delta=0.01) def test_matches_amesim_pn2pipefr_transition_baselines(self) -> None: medium = AmesimHeliumPengRobinsonMedium() pipe = AmesimPnl00r( "pnl_1", medium, diam=0.014, le=1.0, rr=0.045 / 14.0, ) # Unchanged test_mql_1.ame, pneumatic_105, ascending transition at # t=0.2862..0.3047 s. AMESim dm1 is positive into port 1 while this # component's port-1-to-port-2 coordinate is negative for these rows. samples = ( ( 10_631_986.90667049, 262.71180639527233, 10_632_026.310978588, 262.7092388107747, -2.8192909824026485e-4, 1408.0846374820405, 5.030700878807581e-6, -0.09668817162408228, 0.04545209089983021, ), ( 10_594_171.934888396, 262.32910587014925, 10_594_213.825687861, 262.3266782251236, -3.598950492898518e-4, 1799.2599253847663, 5.723572909300966e-6, -0.1236749841566422, 0.03590026408283139, ), ( 10_554_721.69137201, 261.9257072357455, 10_554_766.280823693, 261.92350790162175, -4.3973197060195834e-4, 2200.69088342098, 6.529418698258151e-6, -0.1514267390526478, 0.03111112083131272, ), ( 10_544_953.846352266, 261.82512637406904, 10_544_999.25593169, 261.82299814935993, -4.592053720488228e-4, 2298.7460389094877, 6.783006948981891e-6, -0.15821443006863944, 0.03074226487938153, ), ( 10_514_807.25092133, 261.5128728347395, 10_514_855.822238008, 261.5109955483938, -5.190048780471613e-4, 2600.199197308607, 7.798343300840328e-6, -0.17910295287093314, 0.03166638237616583, ), ( 10_432_760.829010766, 260.6482173603332, 10_432_818.817992153, 260.6472539768074, -6.759582533840764e-4, 3394.136683451109, 1.1253055264735043e-5, -0.23427502517225277, 0.03839469933871734, ), ) for p_1, T_1, p_2, T_2, mass_flow, re, cm, velocity, ff in samples: with self.subTest(reynolds=re): pipe.port_1.p = p_1 pipe.port_2.p = p_2 pipe.update_stream_outflows( { "port_1": medium.specific_enthalpy_at_pressure(p_1, T_1), "port_2": medium.specific_enthalpy_at_pressure(p_2, T_2), } ) actual_mass_flow = pipe.mass_flow(p_1, p_2) results = pipe.component_result_values() self.assertAlmostEqual(actual_mass_flow / mass_flow, 1.0, delta=0.006) self.assertAlmostEqual(results["re"] / re, 1.0, delta=0.012) self.assertAlmostEqual(results["cm"] / cm, 1.0, delta=1.0e-4) self.assertAlmostEqual(results["v"] / velocity, 1.0, delta=0.006) self.assertAlmostEqual(results["ff"] / ff, 1.0, delta=0.012) def test_upstream_temperature_uses_same_side_connected_stream(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium) h_1 = self.medium.specific_enthalpy(250.0) h_2 = self.medium.specific_enthalpy(400.0) pipe.update_stream_outflows({"port_1": h_1, "port_2": h_2}) self.assertAlmostEqual(pipe._port_temperature("port_1"), 250.0) self.assertAlmostEqual(pipe._port_temperature("port_2"), 400.0) self.assertEqual(pipe.port_1.h_outflow, h_2) self.assertEqual(pipe.port_2.h_outflow, h_1) def test_pn2pipefr_cache_keys_all_property_inputs(self) -> None: pipe = AmesimPnl00r( "pnl_1", self.medium, diam=0.014, le=1.0, rr=0.045 / 14.0, ) pipe.update_flow_temperature_references( { "port_1": self.medium.specific_enthalpy(300.0), "port_2": self.medium.specific_enthalpy(300.0), } ) AmesimPnl0001._one_way_pn2pipefr_mass_flow.cache_clear() first = pipe.mass_flow(501000.0, 500000.0) after_first = AmesimPnl0001._one_way_pn2pipefr_mass_flow.cache_info() second = pipe.mass_flow(501000.0, 500000.0) after_second = AmesimPnl0001._one_way_pn2pipefr_mass_flow.cache_info() pipe.update_flow_temperature_references( { "port_1": self.medium.specific_enthalpy(400.0), "port_2": self.medium.specific_enthalpy(300.0), } ) third = pipe.mass_flow(501000.0, 500000.0) after_temperature_change = AmesimPnl0001._one_way_pn2pipefr_mass_flow.cache_info() self.assertEqual(first, second) self.assertEqual(after_first.misses, 1) self.assertEqual(after_second.hits, after_first.hits + 1) self.assertNotEqual(third, second) self.assertEqual( after_temperature_change.misses, after_second.misses + 1, ) def test_stream_outflows_are_crossed_but_flow_temperature_is_same_side(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium) hot_h = self.medium.specific_enthalpy(420.0) cold_h = self.medium.specific_enthalpy(240.0) pipe.update_stream_outflows({"port_1": hot_h, "port_2": cold_h}) self.assertEqual(pipe.port_1.h_outflow, cold_h) self.assertEqual(pipe.port_2.h_outflow, hot_h) self.assertEqual(pipe._connected_h, {"port_1": hot_h, "port_2": cold_h}) self.assertAlmostEqual(pipe._port_temperature("port_1"), 420.0) self.assertAlmostEqual(pipe._port_temperature("port_2"), 240.0) warmer_h = self.medium.specific_enthalpy(460.0) pipe.update_flow_temperature_references( {"port_1": warmer_h, "port_2": cold_h} ) self.assertEqual( pipe._connected_h, {"port_1": warmer_h, "port_2": cold_h}, ) self.assertAlmostEqual(pipe._port_temperature("port_1"), 460.0) # Updating the pressure-flow-only reference must not change the # zero-volume component's already propagated connector outflows. self.assertEqual(pipe.port_1.h_outflow, cold_h) self.assertEqual(pipe.port_2.h_outflow, hot_h) def test_stream_closure_refreshes_same_side_cache_before_recomputed_flow(self) -> None: hot_temperature = 420.0 cold_temperature = 240.0 source = Cylinder( "source", self.medium, V=0.02, p0=501000.0, T0=hot_temperature, ) pipe = AmesimPnl00r("pnl_1", self.medium) sink = Tank( "sink", self.medium, V=0.05, p0=500000.0, T0=cold_temperature, ) network = SimulationNetwork("pnl00r-stream-refresh") for component in (source, pipe, sink): network.add_component(component) network.connect("source", "port_b", "pnl_1", "port_1") network.connect("pnl_1", "port_2", "sink", "port_a") diagnostics, connected_h = StreamResolver(network).solve() self.assertTrue(diagnostics.converged) self.assertEqual(pipe._connected_h, connected_h["pnl_1"]) self.assertAlmostEqual(pipe._port_temperature("port_1"), hot_temperature) self.assertAlmostEqual(pipe._port_temperature("port_2"), cold_temperature) # The pressure-flow layer runs again after this stream closure pass. # Its first recomputation therefore observes the freshly cached source # enthalpy, independently of the crossed connector outflow values. pipe.port_1.p = source.port_b.p pipe.port_2.p = sink.port_a.p flow = pipe.mass_flow(pipe.port_1.p, pipe.port_2.p) self.assertGreater(flow, 0.0) def test_node_port_2_reference_refresh_does_not_replace_crossed_outflows( self, ) -> None: hot = Cylinder( "hot", self.medium, V=0.1, p0=100_000.0, T0=400.0, ) cold = Cylinder( "cold", self.medium, V=0.1, p0=100_000.0, T0=200.0, ) remote = Cylinder( "remote", self.medium, V=0.1, p0=100_000.0, T0=300.0, ) node = AmesimPn3Node2("node") pipe = AmesimPnl00r("pnl_1", self.medium) node.port_1.m_flow = 1.0 node.port_2.m_flow = -1.0 node.port_3.m_flow = 0.0 network = SimulationNetwork("pnl00r-node-reference") for component in (hot, cold, remote, node, pipe): network.add_component(component) network.connect("hot", "port_b", "node", "port_1") network.connect("cold", "port_b", "node", "port_3") network.connect("node", "port_2", "pnl_1", "port_1") network.connect("pnl_1", "port_2", "remote", "port_b") resolver = StreamResolver(network) diagnostics, connected_h = resolver.solve() references = resolver.connected_temperature_reference_enthalpies() outflows_before = (pipe.port_1.h_outflow, pipe.port_2.h_outflow) self.assertTrue(diagnostics.converged) self.assertNotEqual(node.port_2.h_outflow, node.temperature_reference_h) self.assertEqual(pipe._connected_h, connected_h["pnl_1"]) self.assertNotEqual( connected_h["pnl_1"]["port_1"], references["pnl_1"]["port_1"], ) resolver.refresh_flow_temperature_references() self.assertEqual(pipe._connected_h, references["pnl_1"]) self.assertEqual( pipe._connected_h["port_1"], node.temperature_reference_h, ) self.assertEqual( (pipe.port_1.h_outflow, pipe.port_2.h_outflow), outflows_before, ) def test_hot_and_cold_upstream_flow_results_and_equations_are_consistent(self) -> None: pipe = AmesimPnl00r( "pnl_1", self.medium, diam=0.014, le=1.0, rr=0.045 / 14.0, ) hot_temperature = 420.0 cold_temperature = 240.0 pipe.update_stream_outflows( { "port_1": self.medium.specific_enthalpy(hot_temperature), "port_2": self.medium.specific_enthalpy(cold_temperature), } ) reference_hot = AmesimPnl00r( "reference_hot", self.medium, diam=pipe.diam, le=pipe.le, rr=pipe.rr, ) reference_hot.update_flow_temperature_references( { "port_1": self.medium.specific_enthalpy(hot_temperature), "port_2": self.medium.specific_enthalpy(hot_temperature), } ) reference_cold = AmesimPnl00r( "reference_cold", self.medium, diam=pipe.diam, le=pipe.le, rr=pipe.rr, ) reference_cold.update_flow_temperature_references( { "port_1": self.medium.specific_enthalpy(cold_temperature), "port_2": self.medium.specific_enthalpy(cold_temperature), } ) cases = ( ("forward_hot", 501000.0, 500000.0, hot_temperature, reference_hot), ("reverse_cold", 500000.0, 501000.0, cold_temperature, reference_cold), ) for label, p_1, p_2, upstream_temperature, reference in cases: with self.subTest(label=label): pipe.port_1.p = p_1 pipe.port_2.p = p_2 reference.port_1.p = p_1 reference.port_2.p = p_2 expected_flow = reference.mass_flow(p_1, p_2) actual_flow = pipe.mass_flow(p_1, p_2) self.assertAlmostEqual(actual_flow, expected_flow) pipe.port_1.m_flow = actual_flow pipe.port_2.m_flow = -actual_flow equation_values = pipe.pressure_flow_equation_values() residual_values = tuple( residual.value for residual in pipe.pressure_flow_equation_residuals() ) self.assertEqual(equation_values, (0.0, 0.0)) self.assertEqual(residual_values, equation_values) results = pipe.component_result_values() upstream_pressure = max(p_1, p_2) density = self.medium.density( upstream_pressure, upstream_temperature, ) expected_reynolds = pipe.reynolds_number( actual_flow, upstream_temperature, ) friction = pipe.friction_factor(expected_reynolds) flow_coefficient = ( pipe.diam / (pipe.le * friction) ) ** 0.5 self.assertAlmostEqual(results["re"], expected_reynolds) self.assertAlmostEqual( results["v"], actual_flow / (density * pipe.area), ) self.assertAlmostEqual( results["cm"], abs(actual_flow) * upstream_temperature**0.5 / (flow_coefficient * pipe.area * upstream_pressure), ) self.assertAlmostEqual( results["ff"], min(friction, 64_000_000.0), ) def test_friction_factor_transitions_from_laminar_to_turbulent(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium, rr=1e-5) self.assertGreater(pipe.friction_factor(1.0e-12), 64_000_000.0) self.assertAlmostEqual(pipe.friction_factor(100.0), 64.0 / 100.0) self.assertGreater(pipe.friction_factor(100000.0), 0.0) self.assertLess(pipe.friction_factor(100000.0), 0.1) def test_reported_friction_factor_is_bounded_at_tiny_flow(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium) pipe.mass_flow = lambda _p_1, _p_2: 1.0e-30 self.assertEqual(pipe.component_result_values()["ff"], 64_000_000.0) def test_friction_factor_matches_amesim_smooth_to_rough_transition(self) -> None: pipe_20mm = AmesimPnl00r( "pnl_20mm", self.medium, diam=0.02, rr=0.045 / 20.0, ) pipe_14mm = AmesimPnl00r( "pnl_14mm", self.medium, diam=0.014, rr=0.045 / 14.0, ) self.assertAlmostEqual( pipe_20mm.friction_factor(56_887.5547), 0.0215740061043, delta=1.0e-4, ) self.assertAlmostEqual( pipe_20mm.friction_factor(700_686.41), 0.02400535718, delta=8.0e-5, ) self.assertAlmostEqual( pipe_14mm.friction_factor(726_799.66), 0.02658268645, delta=8.0e-5, ) def test_friction_factor_matches_amesim_transition_sweep(self) -> None: pipe = AmesimPnl00r( "pnl_14mm", self.medium, diam=0.014, rr=0.045 / 14.0, ) samples = ( (1408.0846374820405, 0.04545209089983021), (1594.9147224775163, 0.04022919421391556), (1799.2599253847663, 0.03590026408283139), (2001.1290261016431, 0.03286496043823008), (2200.69088342098, 0.03111112083131272), (2298.7460389094877, 0.03074226487938153), (2396.165041443924, 0.030707477592933765), (2600.199197308607, 0.03166638237616583), (2795.6649128003255, 0.03349648426341731), (2996.892872668051, 0.035593529418766444), (3204.685791989844, 0.037354280559382634), (3394.136683451109, 0.03839469933871734), ) for reynolds, expected in samples: with self.subTest(reynolds=reynolds): self.assertAlmostEqual( pipe.friction_factor(reynolds) / expected, 1.0, delta=0.0015, ) def test_mass_flow_is_monotone_through_transition(self) -> None: medium = AmesimHeliumPengRobinsonMedium() pipe = AmesimPnl00r( "pnl_14mm", medium, diam=0.014, le=1.0, rr=0.045 / 14.0, ) mean_pressure = 10.6e6 temperature = 262.0 enthalpy = medium.specific_enthalpy_at_pressure(mean_pressure, temperature) pipe.update_flow_temperature_references( {"port_1": enthalpy, "port_2": enthalpy} ) forward: list[float] = [] reverse: list[float] = [] for index in range(221): pressure_drop = 25.0 + index * 0.25 high = mean_pressure + 0.5 * pressure_drop low = mean_pressure - 0.5 * pressure_drop forward.append(pipe.mass_flow(high, low)) reverse.append(-pipe.mass_flow(low, high)) self.assertLess(pipe.reynolds_number(forward[0], temperature), 1400.0) self.assertGreater( pipe.reynolds_number(forward[-1], temperature), 3400.0 ) self.assertTrue(all(left < right for left, right in zip(forward, forward[1:]))) for forward_flow, reverse_flow in zip(forward, reverse): self.assertAlmostEqual(forward_flow, reverse_flow, delta=1.0e-12) def test_friction_factor_matches_amesim_transition_regime(self) -> None: pipe = AmesimPnl00r( "pnl_20mm", self.medium, diam=0.02, rr=0.045 / 20.0, ) self.assertAlmostEqual( pipe.friction_factor(3_699.80236), 0.0391257647759, delta=4.0e-4, ) def test_darcy_pressure_drop_uses_flow_sign(self) -> None: pipe = AmesimPnl00r("pnl_1", self.medium) density = self.medium.density(500000.0, 300.0) forward = pipe.darcy_pressure_drop(0.01, density=density, temperature=300.0) reverse = pipe.darcy_pressure_drop(-0.01, density=density, temperature=300.0) self.assertGreater(forward, 0.0) self.assertLess(reverse, 0.0) self.assertAlmostEqual(forward, -reverse) if __name__ == "__main__": unittest.main()