From 6f9e7084ddfdd8e1cdd0451c5086740dc48da922 Mon Sep 17 00:00:00 2001 From: huojiarong Date: Thu, 16 Jul 2026 01:59:15 +0000 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85test=5Fmql=E6=B0=94=E5=AE=A4?= =?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 --- .../test_mql_chamber_observations.py | 195 ++++++++++++++++++ tests/test_test_mql_chamber_observations.py | 112 ++++++++++ 2 files changed, 307 insertions(+) create mode 100644 PythonModels/reporting/test_mql_chamber_observations.py create mode 100644 tests/test_test_mql_chamber_observations.py diff --git a/PythonModels/reporting/test_mql_chamber_observations.py b/PythonModels/reporting/test_mql_chamber_observations.py new file mode 100644 index 0000000..fd60209 --- /dev/null +++ b/PythonModels/reporting/test_mql_chamber_observations.py @@ -0,0 +1,195 @@ +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, +) + + +@dataclass(frozen=True) +class TestMqlChamberObservation: + time: float + pressure_pa: float + temperature_k: float + gas_mass_g: float + volume_cm3: float | None + + +@dataclass(frozen=True) +class TestMqlChamberBinding: + alias: str + submodel: str + pressure_path: str + temperature_path: str + gas_mass_path: str + pressure_duplicate_paths: tuple[str, ...] + temperature_duplicate_paths: tuple[str, ...] + volume_path: str | None + + @property + def is_variable(self) -> bool: + return self.volume_path is not None + + def observation_at(self, results: AmesimResults, index: int) -> TestMqlChamberObservation: + return TestMqlChamberObservation( + time=results.times[index], + pressure_pa=results.series(self.pressure_path)[index], + temperature_k=results.series(self.temperature_path)[index], + gas_mass_g=results.series(self.gas_mass_path)[index], + volume_cm3=( + results.series(self.volume_path)[index] + if self.volume_path is not None + else None + ), + ) + + +@dataclass(frozen=True) +class TestMqlChamberObservationCatalog: + bindings: tuple[TestMqlChamberBinding, ...] + + @property + def fixed_count(self) -> int: + return sum(1 for binding in self.bindings if binding.submodel == "PNCH023") + + @property + def variable_count(self) -> int: + return sum(1 for binding in self.bindings if binding.submodel == "PNCH012") + + def by_alias(self, alias: str) -> TestMqlChamberBinding: + for binding in self.bindings: + if binding.alias == alias: + return binding + raise KeyError(alias) + + +def build_test_mql_chamber_observation_catalog( + results: AmesimResults, + *, + variable_catalog: TestMqlVariableCatalog | None = None, + assembly: TestMqlPneumaticAssembly | None = None, +) -> TestMqlChamberObservationCatalog: + variable_catalog = variable_catalog or build_test_mql_variable_catalog(results) + assembly = assembly or build_test_mql_pneumatic_assembly() + chamber_aliases = { + **{alias: "PNCH023" for alias in assembly.fixed_chambers}, + **{alias: "PNCH012" for alias in assembly.variable_chambers}, + } + bindings = [] + for alias, submodel in chamber_aliases.items(): + variables = tuple( + variable + for variable in variable_catalog.variables + if variable.owner_alias == alias + ) + pressure = _primary_observable(variables, "press", expected_units="Pa") + temperature = _primary_observable(variables, "temp", expected_units="K") + gas_mass = _required_path( + variables, + "mgas1" if submodel == "PNCH012" else "mgas", + expected_units="g", + ) + volume = _optional_path(variables, "vol", expected_units="cm**3") + bindings.append( + TestMqlChamberBinding( + alias=alias, + submodel=submodel, + pressure_path=pressure.data_path, + temperature_path=temperature.data_path, + gas_mass_path=gas_mass, + pressure_duplicate_paths=_duplicate_paths(variables, "press", expected_units="Pa"), + temperature_duplicate_paths=_duplicate_paths(variables, "temp", expected_units="K"), + volume_path=volume, + ) + ) + return TestMqlChamberObservationCatalog( + bindings=tuple(sorted(bindings, key=lambda binding: binding.alias)) + ) + + +def _primary_observable( + variables: tuple[TestMqlVariableBinding, ...], + signal_prefix: str, + *, + expected_units: str, +) -> TestMqlVariableBinding: + matches = tuple( + variable + for variable in variables + if variable.signal_name == signal_prefix + and "duplicate" not in variable.label + ) + variable = _single(matches, f"primary {signal_prefix}") + _assert_units(variable, expected_units) + return variable + + +def _duplicate_paths( + variables: tuple[TestMqlVariableBinding, ...], + signal_prefix: str, + *, + expected_units: str, +) -> tuple[str, ...]: + matches = tuple( + variable + for variable in variables + if variable.signal_name.startswith(signal_prefix) + and variable.signal_name != signal_prefix + and "duplicate" in variable.label + ) + 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, +) -> str: + variable = _single( + tuple(variable for variable in variables if variable.signal_name == signal_name), + signal_name, + ) + _assert_units(variable, expected_units) + return variable.data_path + + +def _optional_path( + variables: tuple[TestMqlVariableBinding, ...], + signal_name: str, + *, + expected_units: str, +) -> 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( + 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: + 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_chamber_observations.py b/tests/test_test_mql_chamber_observations.py new file mode 100644 index 0000000..7a2cc92 --- /dev/null +++ b/tests/test_test_mql_chamber_observations.py @@ -0,0 +1,112 @@ +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_chamber_observations import ( + build_test_mql_chamber_observation_catalog, +) + + +REPO_ROOT = Path(__file__).resolve().parents[1] +TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame" + + +class TestMqlChamberObservationCatalogTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME) + cls.catalog = build_test_mql_chamber_observation_catalog(cls.amesim_results) + + def test_builds_expected_chamber_observation_bindings(self) -> None: + self.assertEqual(len(self.catalog.bindings), 12) + self.assertEqual(self.catalog.fixed_count, 4) + self.assertEqual(self.catalog.variable_count, 8) + + fixed = self.catalog.by_alias("pn_general_chamber") + variable = self.catalog.by_alias("pn_c1_8") + + self.assertEqual(fixed.submodel, "PNCH023") + self.assertFalse(fixed.is_variable) + self.assertEqual(fixed.pressure_path, "press@pn_general_chamber") + self.assertEqual(fixed.temperature_path, "temp@pn_general_chamber") + self.assertEqual(fixed.gas_mass_path, "mgas@pn_general_chamber") + self.assertEqual(fixed.pressure_duplicate_paths, ("press2@pn_general_chamber",)) + self.assertEqual(fixed.temperature_duplicate_paths, ("temp2@pn_general_chamber",)) + self.assertIsNone(fixed.volume_path) + self.assertEqual(variable.submodel, "PNCH012") + self.assertTrue(variable.is_variable) + self.assertEqual(variable.pressure_path, "press@pn_c1_8") + self.assertEqual(variable.temperature_path, "temp@pn_c1_8") + self.assertEqual(variable.gas_mass_path, "mgas1@pn_c1_8") + self.assertEqual( + variable.pressure_duplicate_paths, + ("press3@pn_c1_8", "press2@pn_c1_8", "press4@pn_c1_8"), + ) + self.assertEqual( + variable.temperature_duplicate_paths, + ("temp3@pn_c1_8", "temp2@pn_c1_8", "temp4@pn_c1_8"), + ) + self.assertEqual(variable.volume_path, "vol@pn_c1_8") + + def test_fixed_chamber_observation_matches_amesim_samples(self) -> None: + observation = self.catalog.by_alias("pn_general_chamber").observation_at( + self.amesim_results, + -1, + ) + + self.assertAlmostEqual(observation.time, self.amesim_results.times[-1]) + self.assertAlmostEqual(observation.pressure_pa, 4310273.332655074) + self.assertAlmostEqual(observation.temperature_k, 179.646447269399) + self.assertAlmostEqual(observation.gas_mass_g, 666.3851161956942) + self.assertIsNone(observation.volume_cm3) + + def test_variable_chamber_observation_matches_amesim_samples(self) -> None: + observation = self.catalog.by_alias("pn_c1_8").observation_at( + self.amesim_results, + -1, + ) + + self.assertAlmostEqual(observation.pressure_pa, 4310310.457796034) + self.assertAlmostEqual(observation.temperature_k, 293.1143259012449) + self.assertAlmostEqual(observation.gas_mass_g, 353.6887889918552) + self.assertAlmostEqual(observation.volume_cm3, 49242.54636512914) + + def test_initial_pressure_conventions_are_preserved(self) -> None: + fixed = self.catalog.by_alias("pn_general_chamber").observation_at( + self.amesim_results, + 0, + ) + variable = self.catalog.by_alias("pn_c1_8").observation_at( + self.amesim_results, + 0, + ) + + self.assertAlmostEqual(fixed.pressure_pa, 15198699.999999998) + self.assertAlmostEqual(variable.pressure_pa, -1299.99999999999) + self.assertAlmostEqual(fixed.temperature_k, 293.15) + self.assertAlmostEqual(variable.temperature_k, 293.15) + self.assertAlmostEqual(variable.volume_cm3, 15000.0) + + def test_duplicate_pressure_and_temperature_paths_match_primary_series(self) -> None: + for binding in self.catalog.bindings: + pressure = self.amesim_results.series(binding.pressure_path) + temperature = self.amesim_results.series(binding.temperature_path) + with self.subTest(alias=binding.alias): + for duplicate_path in binding.pressure_duplicate_paths: + duplicate = self.amesim_results.series(duplicate_path) + self.assertLess( + max(abs(a - b) for a, b in zip(pressure, duplicate)), + 1.0e-12, + ) + for duplicate_path in binding.temperature_duplicate_paths: + duplicate = self.amesim_results.series(duplicate_path) + self.assertLess( + max(abs(a - b) for a, b in zip(temperature, duplicate)), + 1.0e-12, + ) + + +if __name__ == "__main__": + unittest.main()