From 7c820cbadaf7467b0bc2edf98aceb0082007e358 Mon Sep 17 00:00:00 2001 From: huojiarong Date: Thu, 16 Jul 2026 01:54:44 +0000 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85test=5Fmql=E7=AE=A1=E8=B7=AF?= =?UTF-8?q?=E8=A7=82=E6=B5=8B=E7=9B=AE=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../reporting/test_mql_line_observations.py | 211 ++++++++++++++++++ tests/test_test_mql_line_observations.py | 111 +++++++++ 2 files changed, 322 insertions(+) create mode 100644 PythonModels/reporting/test_mql_line_observations.py create mode 100644 tests/test_test_mql_line_observations.py diff --git a/PythonModels/reporting/test_mql_line_observations.py b/PythonModels/reporting/test_mql_line_observations.py new file mode 100644 index 0000000..0a85c8a --- /dev/null +++ b/PythonModels/reporting/test_mql_line_observations.py @@ -0,0 +1,211 @@ +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_lines import ( + TestMqlLineAssembly, + build_test_mql_line_assembly, +) + + +G_PER_S_TO_KG_PER_S = 1.0e-3 + + +@dataclass(frozen=True) +class TestMqlLineObservation: + time: float + mass_flows_kg_s: dict[str, float] + enthalpy_flows_w: dict[str, float] + pressures_pa: dict[str, float] + temperatures_k: dict[str, float] + gas_mass_g: float | None + reynolds_number: float + mass_flow_parameter: float + gas_velocity_m_s: float + friction_factor: float + + +@dataclass(frozen=True) +class TestMqlLineObservationBinding: + alias: str + submodel: str + pattern: str + mass_flow_paths: tuple[str, ...] + enthalpy_flow_paths: tuple[str, ...] + pressure_paths: tuple[str, ...] + temperature_paths: tuple[str, ...] + gas_mass_path: str | None + reynolds_path: str + mass_flow_parameter_path: str + gas_velocity_path: str + friction_factor_path: str + + def mass_flow_kg_s_series( + self, + results: AmesimResults, + data_path: str | None = None, + ) -> tuple[float, ...]: + path = data_path or self.mass_flow_paths[0] + if path not in self.mass_flow_paths: + raise KeyError(path) + return tuple(value * G_PER_S_TO_KG_PER_S for value in results.series(path)) + + def observation_at(self, results: AmesimResults, index: int) -> TestMqlLineObservation: + return TestMqlLineObservation( + time=results.times[index], + mass_flows_kg_s={ + path: results.series(path)[index] * G_PER_S_TO_KG_PER_S + for path in self.mass_flow_paths + }, + enthalpy_flows_w={ + path: results.series(path)[index] + for path in self.enthalpy_flow_paths + }, + pressures_pa={ + path: results.series(path)[index] + for path in self.pressure_paths + }, + temperatures_k={ + path: results.series(path)[index] + for path in self.temperature_paths + }, + gas_mass_g=( + results.series(self.gas_mass_path)[index] + if self.gas_mass_path is not None + else None + ), + reynolds_number=results.series(self.reynolds_path)[index], + mass_flow_parameter=results.series(self.mass_flow_parameter_path)[index], + gas_velocity_m_s=results.series(self.gas_velocity_path)[index], + friction_factor=results.series(self.friction_factor_path)[index], + ) + + +@dataclass(frozen=True) +class TestMqlLineObservationCatalog: + bindings: tuple[TestMqlLineObservationBinding, ...] + + @property + def line_count(self) -> int: + return len(self.bindings) + + def by_alias(self, alias: str) -> TestMqlLineObservationBinding: + for binding in self.bindings: + if binding.alias == alias: + return binding + raise KeyError(alias) + + def by_submodel(self, submodel: str) -> tuple[TestMqlLineObservationBinding, ...]: + return tuple(binding for binding in self.bindings if binding.submodel == submodel) + + +def build_test_mql_line_observation_catalog( + results: AmesimResults, + *, + variable_catalog: TestMqlVariableCatalog | None = None, + line_assembly: TestMqlLineAssembly | None = None, +) -> TestMqlLineObservationCatalog: + variable_catalog = variable_catalog or build_test_mql_variable_catalog(results) + line_assembly = line_assembly or build_test_mql_line_assembly(results, variable_catalog) + bindings = [] + for line in line_assembly.lines: + variables = tuple( + variable + for variable in variable_catalog.variables + if variable.owner_alias == line.alias + ) + bindings.append( + TestMqlLineObservationBinding( + alias=line.alias, + submodel=line.submodel, + pattern=line.pattern, + mass_flow_paths=_paths_with_prefix(variables, "dm", expected_units="g/s"), + enthalpy_flow_paths=_paths_with_prefix(variables, "dh", expected_units="J/s"), + pressure_paths=_paths_with_prefix(variables, "p", expected_units="Pa"), + temperature_paths=_paths_with_prefix(variables, "t", expected_units="K"), + gas_mass_path=_optional_path(variables, "mgas", expected_units="g"), + reynolds_path=_required_path(variables, "re", expected_units=None), + mass_flow_parameter_path=_required_path( + variables, + "cm", + expected_units="(kg*K/J)**(1/2)", + ), + gas_velocity_path=_required_path(variables, "v", expected_units="m/s"), + friction_factor_path=_required_path(variables, "ff", expected_units=None), + ) + ) + return TestMqlLineObservationCatalog(bindings=tuple(bindings)) + + +def _paths_with_prefix( + variables: tuple[TestMqlVariableBinding, ...], + prefix: str, + *, + expected_units: str | None, +) -> tuple[str, ...]: + matches = tuple( + variable + for variable in variables + if variable.signal_name.startswith(prefix) + ) + for variable in matches: + _assert_units(variable, expected_units) + return tuple(variable.data_path for variable in matches) + + +def _required_path( + variables: tuple[TestMqlVariableBinding, ...], + signal_name: str, + *, + expected_units: str | None, +) -> str: + variable = _single_signal(variables, signal_name) + _assert_units(variable, expected_units) + return variable.data_path + + +def _optional_path( + variables: tuple[TestMqlVariableBinding, ...], + signal_name: str, + *, + expected_units: str | None, +) -> str | None: + matches = tuple(variable for variable in variables if variable.signal_name == signal_name) + if not matches: + return None + variable = _single(matches, signal_name) + _assert_units(variable, expected_units) + return variable.data_path + + +def _single_signal( + variables: tuple[TestMqlVariableBinding, ...], + signal_name: str, +) -> TestMqlVariableBinding: + return _single( + tuple(variable for variable in variables if variable.signal_name == signal_name), + signal_name, + ) + + +def _single( + matches: tuple[TestMqlVariableBinding, ...], + description: str, +) -> TestMqlVariableBinding: + if len(matches) != 1: + raise ValueError(f"Expected one {description} variable, found {len(matches)}.") + return matches[0] + + +def _assert_units(variable: TestMqlVariableBinding, expected_units: str | None) -> None: + if variable.units != expected_units: + raise ValueError( + f"Unexpected units for {variable.data_path}: " + f"{variable.units!r}, expected {expected_units!r}." + ) diff --git a/tests/test_test_mql_line_observations.py b/tests/test_test_mql_line_observations.py new file mode 100644 index 0000000..97504ac --- /dev/null +++ b/tests/test_test_mql_line_observations.py @@ -0,0 +1,111 @@ +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_line_observations import ( + G_PER_S_TO_KG_PER_S, + build_test_mql_line_observation_catalog, +) + + +REPO_ROOT = Path(__file__).resolve().parents[1] +TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame" + + +class TestMqlLineObservationCatalogTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME) + cls.catalog = build_test_mql_line_observation_catalog(cls.amesim_results) + + def test_builds_expected_line_observation_counts(self) -> None: + self.assertEqual(self.catalog.line_count, 40) + self.assertEqual(len(self.catalog.by_submodel("PNL0001")), 20) + self.assertEqual(len(self.catalog.by_submodel("PNL0002")), 8) + self.assertEqual(len(self.catalog.by_submodel("PNL0003")), 8) + self.assertEqual(len(self.catalog.by_submodel("PNL00R")), 4) + + def test_pnl0001_binding_exposes_single_flow_and_compliance_observables(self) -> None: + binding = self.catalog.by_alias("pneumatic_69") + + self.assertEqual(binding.submodel, "PNL0001") + self.assertEqual(binding.pattern, "C-R") + self.assertEqual(binding.mass_flow_paths, ("dm1@pneumatic_69",)) + self.assertEqual(binding.enthalpy_flow_paths, ("dh1@pneumatic_69",)) + self.assertEqual(binding.pressure_paths, ("p2@pneumatic_69",)) + self.assertEqual(binding.temperature_paths, ("t2@pneumatic_69",)) + self.assertEqual(binding.gas_mass_path, "mgas@pneumatic_69") + self.assertEqual(binding.reynolds_path, "re@pneumatic_69") + self.assertEqual(binding.mass_flow_parameter_path, "cm@pneumatic_69") + self.assertEqual(binding.gas_velocity_path, "v@pneumatic_69") + self.assertEqual(binding.friction_factor_path, "ff@pneumatic_69") + + def test_pnl0002_and_pnl0003_keep_distinct_flow_topologies(self) -> None: + rcr = self.catalog.by_alias("pneumatic_80") + crc = self.catalog.by_alias("pneumatic_75") + + self.assertEqual(rcr.pattern, "R-C-R") + self.assertEqual(rcr.mass_flow_paths, ("dm1@pneumatic_80", "dm2@pneumatic_80")) + self.assertEqual(rcr.enthalpy_flow_paths, ("dh1@pneumatic_80", "dh2@pneumatic_80")) + self.assertEqual(rcr.pressure_paths, ("pctr@pneumatic_80",)) + self.assertEqual(rcr.temperature_paths, ("tctr@pneumatic_80",)) + self.assertEqual(crc.pattern, "C-R-C") + self.assertEqual(crc.mass_flow_paths, ("dmctr@pneumatic_75",)) + self.assertEqual(crc.enthalpy_flow_paths, ("dhctr@pneumatic_75",)) + self.assertEqual(crc.pressure_paths, ("p1@pneumatic_75", "p2@pneumatic_75")) + self.assertEqual(crc.temperature_paths, ("t1@pneumatic_75", "t2@pneumatic_75")) + + def test_mass_flow_series_are_converted_to_si_units(self) -> None: + binding = self.catalog.by_alias("pneumatic_69") + mass_flow = binding.mass_flow_kg_s_series(self.amesim_results) + + self.assertAlmostEqual( + mass_flow[-1], + self.amesim_results.series("dm1@pneumatic_69")[-1] * G_PER_S_TO_KG_PER_S, + ) + self.assertAlmostEqual(mass_flow[-1], -9.527778353091475e-6) + self.assertAlmostEqual(max(mass_flow), 0.4490761730524304) + + def test_observation_exposes_line_state_and_diagnostics(self) -> None: + observation = self.catalog.by_alias("pneumatic_75").observation_at( + self.amesim_results, + -1, + ) + + self.assertAlmostEqual(observation.time, self.amesim_results.times[-1]) + self.assertAlmostEqual( + observation.mass_flows_kg_s["dmctr@pneumatic_75"], + 9.486502325625855e-6, + ) + self.assertAlmostEqual( + observation.enthalpy_flows_w["dhctr@pneumatic_75"], + -0.7322143070637952, + ) + self.assertAlmostEqual(observation.pressures_pa["p1@pneumatic_75"], 4310289.672688478) + self.assertAlmostEqual(observation.pressures_pa["p2@pneumatic_75"], 4310295.447776705) + self.assertAlmostEqual(observation.temperatures_k["t1@pneumatic_75"], 281.37837485614546) + self.assertAlmostEqual(observation.temperatures_k["t2@pneumatic_75"], 282.9136549991407) + self.assertAlmostEqual(observation.gas_mass_g, 0.7031236652270819) + self.assertAlmostEqual(observation.reynolds_number, 31.539799199000434) + self.assertAlmostEqual(observation.mass_flow_parameter, 6.351754064211156e-07) + self.assertAlmostEqual(observation.gas_velocity_m_s, 0.0040585650473562935) + self.assertAlmostEqual(observation.friction_factor, 2.0291822277051246) + + def test_resistance_only_lines_preserve_reversed_duplicate_signs(self) -> None: + for binding in self.catalog.by_submodel("PNL00R"): + dm1 = self.amesim_results.series(f"dm1@{binding.alias}") + dm2 = self.amesim_results.series(f"dm2@{binding.alias}") + dh1 = self.amesim_results.series(f"dh1@{binding.alias}") + dh2 = self.amesim_results.series(f"dh2@{binding.alias}") + with self.subTest(alias=binding.alias): + self.assertIsNone(binding.gas_mass_path) + self.assertEqual(binding.pressure_paths, ()) + self.assertEqual(binding.temperature_paths, ()) + self.assertLess(max(abs(a + b) for a, b in zip(dm1, dm2)), 1.0e-36) + self.assertLess(max(abs(a + b) for a, b in zip(dh1, dh2)), 1.0e-34) + + +if __name__ == "__main__": + unittest.main()