PythonModels/reporting/test_mql_observations.py

@@ -0,0 +1,210 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from PythonModels.reporting.amesim_results import AmesimResults
+from PythonModels.reporting.test_mql_chamber_observations import (
+    TestMqlChamberObservationCatalog,
+    build_test_mql_chamber_observation_catalog,
+)
+from PythonModels.reporting.test_mql_line_observations import (
+    TestMqlLineObservationCatalog,
+    build_test_mql_line_observation_catalog,
+)
+from PythonModels.reporting.test_mql_mechanical_observations import (
+    TestMqlMechanicalObservationCatalog,
+    build_test_mql_mechanical_observation_catalog,
+)
+from PythonModels.reporting.test_mql_orifice_observations import (
+    TestMqlOrificeObservationCatalog,
+    build_test_mql_orifice_observation_catalog,
+)
+from PythonModels.reporting.test_mql_variables import (
+    TestMqlVariableCatalog,
+    build_test_mql_variable_catalog,
+)
+
+
+@dataclass(frozen=True)
+class TestMqlObservationCatalog:
+    variable_catalog: TestMqlVariableCatalog
+    chambers: TestMqlChamberObservationCatalog
+    orifices: TestMqlOrificeObservationCatalog
+    lines: TestMqlLineObservationCatalog
+    mechanical: TestMqlMechanicalObservationCatalog
+
+    @property
+    def binding_count(self) -> int:
+        return (
+            len(self.chambers.bindings)
+            + len(self.orifices.bindings)
+            + self.lines.line_count
+            + self.mechanical.binding_count
+        )
+
+    def data_paths_by_domain(self) -> dict[str, tuple[str, ...]]:
+        return {
+            "chambers": _sorted_unique(_chamber_data_paths(self.chambers)),
+            "orifices": _sorted_unique(_orifice_data_paths(self.orifices)),
+            "lines": _sorted_unique(_line_data_paths(self.lines)),
+            "mechanical": _sorted_unique(_mechanical_data_paths(self.mechanical)),
+        }
+
+    def data_paths(self) -> tuple[str, ...]:
+        paths = []
+        for domain_paths in self.data_paths_by_domain().values():
+            paths.extend(domain_paths)
+        return _sorted_unique(paths)
+
+    def baseline_series_by_data_path(
+        self,
+        results: AmesimResults,
+        data_paths: tuple[str, ...] | list[str] | None = None,
+    ) -> dict[str, tuple[float, ...]]:
+        selected_paths = tuple(data_paths) if data_paths is not None else self.data_paths()
+        _validate_observed_paths(self, selected_paths)
+        return {data_path: results.series(data_path) for data_path in selected_paths}
+
+
+def build_test_mql_observation_catalog(results: AmesimResults) -> TestMqlObservationCatalog:
+    variable_catalog = build_test_mql_variable_catalog(results)
+    return TestMqlObservationCatalog(
+        variable_catalog=variable_catalog,
+        chambers=build_test_mql_chamber_observation_catalog(
+            results,
+            variable_catalog=variable_catalog,
+        ),
+        orifices=build_test_mql_orifice_observation_catalog(
+            results,
+            variable_catalog=variable_catalog,
+        ),
+        lines=build_test_mql_line_observation_catalog(
+            results,
+            variable_catalog=variable_catalog,
+        ),
+        mechanical=build_test_mql_mechanical_observation_catalog(
+            results,
+            variable_catalog=variable_catalog,
+        ),
+    )
+
+
+def _chamber_data_paths(catalog: TestMqlChamberObservationCatalog) -> tuple[str, ...]:
+    paths = []
+    for binding in catalog.bindings:
+        paths.extend(
+            [
+                binding.pressure_path,
+                binding.temperature_path,
+                binding.gas_mass_path,
+                *binding.pressure_duplicate_paths,
+                *binding.temperature_duplicate_paths,
+            ]
+        )
+        if binding.volume_path is not None:
+            paths.append(binding.volume_path)
+    return tuple(paths)
+
+
+def _orifice_data_paths(catalog: TestMqlOrificeObservationCatalog) -> tuple[str, ...]:
+    paths = []
+    for binding in catalog.bindings:
+        paths.extend(
+            [
+                binding.primary_mass_flow_path,
+                binding.primary_enthalpy_flow_path,
+                binding.reversed_mass_flow_path,
+                binding.reversed_enthalpy_flow_path,
+                binding.mass_flow_parameter_path,
+                binding.gas_velocity_path,
+            ]
+        )
+        if binding.opening_path is not None:
+            paths.append(binding.opening_path)
+    return tuple(paths)
+
+
+def _line_data_paths(catalog: TestMqlLineObservationCatalog) -> tuple[str, ...]:
+    paths = []
+    for binding in catalog.bindings:
+        paths.extend(binding.mass_flow_paths)
+        paths.extend(binding.enthalpy_flow_paths)
+        paths.extend(binding.pressure_paths)
+        paths.extend(binding.temperature_paths)
+        if binding.gas_mass_path is not None:
+            paths.append(binding.gas_mass_path)
+        paths.extend(
+            [
+                binding.reynolds_path,
+                binding.mass_flow_parameter_path,
+                binding.gas_velocity_path,
+                binding.friction_factor_path,
+            ]
+        )
+    return tuple(paths)
+
+
+def _mechanical_data_paths(catalog: TestMqlMechanicalObservationCatalog) -> tuple[str, ...]:
+    paths = []
+    for binding in catalog.pistons.values():
+        paths.extend(
+            [
+                binding.volume_path,
+                binding.volume_rate_path,
+                binding.length_path,
+                binding.force_port_2_path,
+                binding.force_port_3_path,
+                binding.displacement_port_2_path,
+                binding.velocity_port_2_path,
+                binding.displacement_port_3_path,
+                binding.velocity_port_3_path,
+            ]
+        )
+    for binding in catalog.masses.values():
+        paths.extend(
+            [
+                binding.displacement_path,
+                binding.velocity_path,
+                binding.acceleration_path,
+                binding.displacement_duplicate_path,
+                binding.velocity_duplicate_path,
+                binding.acceleration_duplicate_path,
+                binding.lower_contact_force_path,
+                binding.upper_contact_force_path,
+                binding.viscous_friction_force_path,
+                binding.dry_friction_force_path,
+                binding.stick_flag_path,
+            ]
+        )
+    for binding in catalog.elastic_endstops.values():
+        paths.extend(
+            [
+                binding.force_path,
+                binding.duplicate_force_path,
+                binding.gap_path,
+                binding.stiffness_path,
+            ]
+        )
+    for binding in catalog.zero_force_sources.values():
+        paths.append(binding.force_path)
+    for binding in catalog.force_connectors.values():
+        paths.append(binding.force_path)
+    for binding in catalog.mechanical_nodes.values():
+        paths.extend(binding.velocity_paths_by_port.values())
+        paths.extend(binding.displacement_paths_by_port.values())
+        paths.append(binding.total_force_path)
+    return tuple(paths)
+
+
+def _validate_observed_paths(
+    catalog: TestMqlObservationCatalog,
+    data_paths: tuple[str, ...],
+) -> None:
+    observed_paths = set(catalog.data_paths())
+    missing = [data_path for data_path in data_paths if data_path not in observed_paths]
+    if missing:
+        raise KeyError(f"Data_Path values are not in the test_mql observation catalog: {missing}")
+
+
+def _sorted_unique(data_paths: tuple[str, ...] | list[str]) -> tuple[str, ...]:
+    return tuple(sorted(set(data_paths)))

tests/test_test_mql_observations.py

@@ -0,0 +1,95 @@
+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_comparison import compare_test_mql_series
+from PythonModels.reporting.test_mql_observations import build_test_mql_observation_catalog
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame"
+
+
+class TestMqlObservationCatalogTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME)
+        cls.catalog = build_test_mql_observation_catalog(cls.amesim_results)
+
+    def test_aggregates_observation_catalog_counts(self) -> None:
+        self.assertEqual(len(self.catalog.chambers.bindings), 12)
+        self.assertEqual(len(self.catalog.orifices.bindings), 16)
+        self.assertEqual(self.catalog.lines.line_count, 40)
+        self.assertEqual(self.catalog.mechanical.binding_count, 46)
+        self.assertEqual(self.catalog.binding_count, 114)
+
+    def test_reports_observed_data_paths_by_domain(self) -> None:
+        paths_by_domain = self.catalog.data_paths_by_domain()
+
+        self.assertEqual(set(paths_by_domain), {"chambers", "orifices", "lines", "mechanical"})
+        self.assertEqual(len(paths_by_domain["chambers"]), 100)
+        self.assertEqual(len(paths_by_domain["orifices"]), 104)
+        self.assertEqual(len(paths_by_domain["lines"]), 388)
+        self.assertEqual(len(paths_by_domain["mechanical"]), 266)
+        self.assertEqual(len(self.catalog.data_paths()), 858)
+        self.assertIn("press@pn_general_chamber", paths_by_domain["chambers"])
+        self.assertIn("dm1@pn_orifice_18", paths_by_domain["orifices"])
+        self.assertIn("pctr@pneumatic_80", paths_by_domain["lines"])
+        self.assertIn("tforce@dynamic_mechanical_node_alternative_2", paths_by_domain["mechanical"])
+
+    def test_all_observed_data_paths_exist_in_saved_amesim_results(self) -> None:
+        for data_path in self.catalog.data_paths():
+            with self.subTest(data_path=data_path):
+                self.assertIn(data_path, self.amesim_results.series_by_data_path)
+
+    def test_baseline_series_can_be_selected_for_comparison(self) -> None:
+        selected_paths = (
+            "press@pn_general_chamber",
+            "dm1@pn_orifice_18",
+            "dm1@pneumatic_69",
+            "vol1@pn_brp2_8",
+            "tforce@dynamic_mechanical_node_alternative_2",
+        )
+        baseline = self.catalog.baseline_series_by_data_path(
+            self.amesim_results,
+            selected_paths,
+        )
+
+        self.assertEqual(tuple(baseline), selected_paths)
+        self.assertEqual(len(baseline["press@pn_general_chamber"]), len(self.amesim_results.times))
+        self.assertAlmostEqual(baseline["vol1@pn_brp2_8"][-1], 34242.54636512914)
+
+    def test_baseline_series_rejects_unobserved_data_path(self) -> None:
+        with self.assertRaises(KeyError):
+            self.catalog.baseline_series_by_data_path(
+                self.amesim_results,
+                ("not_a_real_signal@not_a_real_owner",),
+            )
+
+    def test_observation_baseline_round_trips_through_comparison_with_zero_error(self) -> None:
+        selected_paths = (
+            "press@pn_c1_8",
+            "dm2@pn_morifice_11",
+            "dmctr@pneumatic_75",
+            "x1@mass_friction_endstops_10",
+        )
+        baseline = self.catalog.baseline_series_by_data_path(
+            self.amesim_results,
+            selected_paths,
+        )
+        comparison = compare_test_mql_series(
+            python_times=self.amesim_results.times,
+            python_series_by_data_path=baseline,
+            amesim_results=self.amesim_results,
+            data_paths=selected_paths,
+        )
+
+        self.assertEqual(len(comparison.metrics), len(selected_paths))
+        self.assertEqual(comparison.max_abs_error, 0.0)
+        self.assertEqual(comparison.max_rel_error, 0.0)
+
+
+if __name__ == "__main__":
+    unittest.main()