diff --git a/PythonModels/reporting/test_mql_orifice_observations.py b/PythonModels/reporting/test_mql_orifice_observations.py new file mode 100644 index 0000000..defe414 --- /dev/null +++ b/PythonModels/reporting/test_mql_orifice_observations.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from PythonModels.reporting.amesim_results import AmesimResults +from PythonModels.reporting.test_mql_variables import ( + TestMqlVariableBinding, + TestMqlVariableCatalog, + build_test_mql_variable_catalog, +) +from PythonModels.systems.test_mql_pneumatic import ( + TestMqlPneumaticAssembly, + build_test_mql_pneumatic_assembly, +) + + +G_PER_S_TO_KG_PER_S = 1.0e-3 + + +@dataclass(frozen=True) +class TestMqlOrificeObservation: + time: float + mass_flow_kg_s: float + enthalpy_flow_w: float + mass_flow_parameter: float + gas_velocity_m_s: float + opening: float + effective_area_m2: float + + +@dataclass(frozen=True) +class TestMqlOrificeBinding: + alias: str + submodel: str + nominal_area_m2: float + flow_coefficient: float + primary_mass_flow_path: str + primary_enthalpy_flow_path: str + reversed_mass_flow_path: str + reversed_enthalpy_flow_path: str + mass_flow_parameter_path: str + gas_velocity_path: str + opening_path: str | None + + @property + def is_variable(self) -> bool: + return self.opening_path is not None + + def opening_series(self, results: AmesimResults) -> tuple[float, ...]: + if self.opening_path is None: + return tuple(1.0 for _ in results.times) + return tuple(results.series(self.opening_path)) + + def mass_flow_kg_s_series(self, results: AmesimResults) -> tuple[float, ...]: + return tuple(value * G_PER_S_TO_KG_PER_S for value in results.series(self.primary_mass_flow_path)) + + def reversed_mass_flow_kg_s_series(self, results: AmesimResults) -> tuple[float, ...]: + return tuple(value * G_PER_S_TO_KG_PER_S for value in results.series(self.reversed_mass_flow_path)) + + def effective_area_series(self, results: AmesimResults) -> tuple[float, ...]: + return tuple(self.nominal_area_m2 * max(opening, 0.0) for opening in self.opening_series(results)) + + def observation_at(self, results: AmesimResults, index: int) -> TestMqlOrificeObservation: + opening = self.opening_series(results)[index] + return TestMqlOrificeObservation( + time=results.times[index], + mass_flow_kg_s=results.series(self.primary_mass_flow_path)[index] * G_PER_S_TO_KG_PER_S, + enthalpy_flow_w=results.series(self.primary_enthalpy_flow_path)[index], + mass_flow_parameter=results.series(self.mass_flow_parameter_path)[index], + gas_velocity_m_s=results.series(self.gas_velocity_path)[index], + opening=opening, + effective_area_m2=self.nominal_area_m2 * max(opening, 0.0), + ) + + +@dataclass(frozen=True) +class TestMqlOrificeObservationCatalog: + bindings: tuple[TestMqlOrificeBinding, ...] + + @property + def fixed_count(self) -> int: + return sum(1 for binding in self.bindings if binding.submodel == "PNOR001") + + @property + def variable_count(self) -> int: + return sum(1 for binding in self.bindings if binding.submodel == "PNVO001") + + def by_alias(self, alias: str) -> TestMqlOrificeBinding: + for binding in self.bindings: + if binding.alias == alias: + return binding + raise KeyError(alias) + + +def build_test_mql_orifice_observation_catalog( + results: AmesimResults, + *, + variable_catalog: TestMqlVariableCatalog | None = None, + assembly: TestMqlPneumaticAssembly | None = None, +) -> TestMqlOrificeObservationCatalog: + variable_catalog = variable_catalog or build_test_mql_variable_catalog(results) + assembly = assembly or build_test_mql_pneumatic_assembly() + bindings = [] + for alias, orifice in { + **assembly.fixed_orifices, + **assembly.variable_orifices, + }.items(): + owner_variables = tuple( + variable + for variable in variable_catalog.variables + if variable.owner_alias == alias + ) + primary_mass_flow = _find_primary(owner_variables, signal_prefix="dm") + primary_enthalpy_flow = _find_primary(owner_variables, signal_prefix="dh") + reversed_mass_flow = _find_reversed(owner_variables, signal_prefix="dm") + reversed_enthalpy_flow = _find_reversed(owner_variables, signal_prefix="dh") + mass_flow_parameter = _find_by_signal(owner_variables, "cm") + gas_velocity = _find_by_signal(owner_variables, "gasvel") + opening = _find_optional_by_signal(owner_variables, "xv") + bindings.append( + TestMqlOrificeBinding( + alias=alias, + submodel=primary_mass_flow.submodel, + nominal_area_m2=orifice.area, + flow_coefficient=orifice.flow_coefficient, + primary_mass_flow_path=primary_mass_flow.data_path, + primary_enthalpy_flow_path=primary_enthalpy_flow.data_path, + reversed_mass_flow_path=reversed_mass_flow.data_path, + reversed_enthalpy_flow_path=reversed_enthalpy_flow.data_path, + mass_flow_parameter_path=mass_flow_parameter.data_path, + gas_velocity_path=gas_velocity.data_path, + opening_path=opening.data_path if opening is not None else None, + ) + ) + return TestMqlOrificeObservationCatalog( + bindings=tuple(sorted(bindings, key=lambda binding: binding.alias)) + ) + + +def _find_primary( + variables: tuple[TestMqlVariableBinding, ...], + *, + signal_prefix: str, +) -> TestMqlVariableBinding: + matches = [ + variable + for variable in variables + if variable.signal_name.startswith(signal_prefix) + and "sign reversed duplicate" not in variable.label + ] + return _single(matches, f"primary {signal_prefix}") + + +def _find_reversed( + variables: tuple[TestMqlVariableBinding, ...], + *, + signal_prefix: str, +) -> TestMqlVariableBinding: + matches = [ + variable + for variable in variables + if variable.signal_name.startswith(signal_prefix) + and "sign reversed duplicate" in variable.label + ] + return _single(matches, f"reversed {signal_prefix}") + + +def _find_by_signal( + variables: tuple[TestMqlVariableBinding, ...], + signal_name: str, +) -> TestMqlVariableBinding: + return _single( + [variable for variable in variables if variable.signal_name == signal_name], + signal_name, + ) + + +def _find_optional_by_signal( + variables: tuple[TestMqlVariableBinding, ...], + signal_name: str, +) -> TestMqlVariableBinding | None: + matches = [variable for variable in variables if variable.signal_name == signal_name] + if not matches: + return None + return _single(matches, signal_name) + + +def _single( + matches: list[TestMqlVariableBinding], + description: str, +) -> TestMqlVariableBinding: + if len(matches) != 1: + raise ValueError(f"Expected one {description} variable, found {len(matches)}.") + return matches[0] diff --git a/tests/test_test_mql_orifice_observations.py b/tests/test_test_mql_orifice_observations.py new file mode 100644 index 0000000..2544d32 --- /dev/null +++ b/tests/test_test_mql_orifice_observations.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import unittest +from pathlib import Path + +from PythonModels.reporting.amesim_results import load_test_mql_amesim_results +from PythonModels.reporting.test_mql_orifice_observations import ( + G_PER_S_TO_KG_PER_S, + build_test_mql_orifice_observation_catalog, +) + + +REPO_ROOT = Path(__file__).resolve().parents[1] +TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame" + + +class TestMqlOrificeObservationCatalogTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME) + cls.catalog = build_test_mql_orifice_observation_catalog(cls.amesim_results) + + def test_builds_expected_orifice_observation_bindings(self) -> None: + self.assertEqual(len(self.catalog.bindings), 16) + self.assertEqual(self.catalog.fixed_count, 8) + self.assertEqual(self.catalog.variable_count, 8) + + fixed = self.catalog.by_alias("pn_orifice_18") + variable = self.catalog.by_alias("pn_morifice_11") + + self.assertEqual(fixed.submodel, "PNOR001") + self.assertEqual(fixed.primary_mass_flow_path, "dm1@pn_orifice_18") + self.assertEqual(fixed.reversed_mass_flow_path, "dm2@pn_orifice_18") + self.assertEqual(fixed.primary_enthalpy_flow_path, "dh1@pn_orifice_18") + self.assertEqual(fixed.reversed_enthalpy_flow_path, "dh2@pn_orifice_18") + self.assertIsNone(fixed.opening_path) + self.assertEqual(variable.submodel, "PNVO001") + self.assertEqual(variable.primary_mass_flow_path, "dm2@pn_morifice_11") + self.assertEqual(variable.reversed_mass_flow_path, "dm3@pn_morifice_11") + self.assertEqual(variable.opening_path, "xv@pn_morifice_11") + + def test_fixed_orifice_opening_defaults_to_full_area(self) -> None: + binding = self.catalog.by_alias("pn_orifice_18") + opening = binding.opening_series(self.amesim_results) + effective_area = binding.effective_area_series(self.amesim_results) + + self.assertEqual(opening[0], 1.0) + self.assertEqual(opening[-1], 1.0) + self.assertAlmostEqual(effective_area[0], 78.5e-6) + self.assertAlmostEqual(effective_area[-1], 78.5e-6) + + def test_variable_orifice_opening_controls_effective_area(self) -> None: + binding = self.catalog.by_alias("pn_morifice_11") + first = binding.observation_at(self.amesim_results, 0) + last = binding.observation_at(self.amesim_results, -1) + + self.assertAlmostEqual(binding.nominal_area_m2, 78.5e-6) + self.assertAlmostEqual(binding.flow_coefficient, 0.45) + self.assertEqual(first.opening, 0.0) + self.assertEqual(first.effective_area_m2, 0.0) + self.assertEqual(last.opening, 1.0) + self.assertAlmostEqual(last.effective_area_m2, 78.5e-6) + + def test_mass_flow_series_is_converted_to_si_units(self) -> None: + binding = self.catalog.by_alias("pn_orifice_18") + mass_flow = binding.mass_flow_kg_s_series(self.amesim_results) + + self.assertAlmostEqual( + mass_flow[-1], + self.amesim_results.series("dm1@pn_orifice_18")[-1] * G_PER_S_TO_KG_PER_S, + ) + self.assertAlmostEqual(mass_flow[-1], 9.097746783809915e-6) + self.assertAlmostEqual(min(mass_flow), -0.4462054699031737) + + def test_reversed_duplicate_mass_and_enthalpy_flow_signs_match_amesim(self) -> None: + for binding in self.catalog.bindings: + primary_mass = self.amesim_results.series(binding.primary_mass_flow_path) + reversed_mass = self.amesim_results.series(binding.reversed_mass_flow_path) + primary_enthalpy = self.amesim_results.series(binding.primary_enthalpy_flow_path) + reversed_enthalpy = self.amesim_results.series(binding.reversed_enthalpy_flow_path) + with self.subTest(alias=binding.alias): + self.assertEqual(len(primary_mass), len(reversed_mass)) + self.assertEqual(len(primary_enthalpy), len(reversed_enthalpy)) + self.assertLess( + max(abs(a + b) for a, b in zip(primary_mass, reversed_mass)), + 1.0e-12, + ) + self.assertLess( + max(abs(a + b) for a, b in zip(primary_enthalpy, reversed_enthalpy)), + 1.0e-9, + ) + + def test_observation_exposes_amesim_cm_and_gas_velocity(self) -> None: + fixed = self.catalog.by_alias("pn_orifice_18").observation_at(self.amesim_results, -1) + variable = self.catalog.by_alias("pn_morifice_11").observation_at(self.amesim_results, -1) + + self.assertAlmostEqual(fixed.mass_flow_parameter, 4.831077004227076e-07) + self.assertAlmostEqual(fixed.gas_velocity_m_s, 0.016760630907703977) + self.assertAlmostEqual(variable.mass_flow_parameter, 9.302013872202105e-07) + self.assertAlmostEqual(variable.gas_velocity_m_s, -0.03277314382613749) + + +if __name__ == "__main__": + unittest.main()