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}."
+        )

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()