PythonModels/reporting/test_mql_output_schema.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from collections import Counter
+from dataclasses import dataclass
+
+from PythonModels.reporting.amesim_results import AmesimResults
+from PythonModels.reporting.test_mql_observations import (
+    TestMqlObservationCatalog,
+    build_test_mql_observation_catalog,
+)
+from PythonModels.reporting.test_mql_variables import TestMqlVariableBinding
+
+
+@dataclass(frozen=True)
+class TestMqlOutputSignal:
+    data_path: str
+    domain: str
+    owner_alias: str
+    owner_kind: str
+    submodel: str
+    signal_name: str
+    units: str | None
+    amesim_index: int
+    saved: bool
+
+
+@dataclass(frozen=True)
+class TestMqlOutputSchema:
+    signals: tuple[TestMqlOutputSignal, ...]
+
+    @property
+    def signal_count(self) -> int:
+        return len(self.signals)
+
+    def by_data_path(self, data_path: str) -> TestMqlOutputSignal:
+        for signal in self.signals:
+            if signal.data_path == data_path:
+                return signal
+        raise KeyError(data_path)
+
+    def data_paths(self) -> tuple[str, ...]:
+        return tuple(signal.data_path for signal in self.signals)
+
+    def data_paths_by_domain(self, domain: str) -> tuple[str, ...]:
+        return tuple(signal.data_path for signal in self.signals if signal.domain == domain)
+
+    def counts_by_domain(self) -> dict[str, int]:
+        return dict(Counter(signal.domain for signal in self.signals))
+
+    def counts_by_submodel(self) -> dict[str, int]:
+        return dict(Counter(signal.submodel for signal in self.signals))
+
+    def counts_by_owner_kind(self) -> dict[str, int]:
+        return dict(Counter(signal.owner_kind for signal in self.signals))
+
+    def counts_by_units(self) -> dict[str | None, int]:
+        return dict(Counter(signal.units for signal in self.signals))
+
+
+def build_test_mql_output_schema(
+    results: AmesimResults,
+    *,
+    observation_catalog: TestMqlObservationCatalog | None = None,
+) -> TestMqlOutputSchema:
+    observation_catalog = observation_catalog or build_test_mql_observation_catalog(results)
+    domain_by_data_path = _domain_by_data_path(observation_catalog)
+    signals = []
+    for data_path in sorted(domain_by_data_path):
+        variable = observation_catalog.variable_catalog.by_data_path(data_path)
+        signals.append(_signal_from_variable(variable, domain_by_data_path[data_path]))
+    return TestMqlOutputSchema(signals=tuple(signals))
+
+
+def _domain_by_data_path(
+    observation_catalog: TestMqlObservationCatalog,
+) -> dict[str, str]:
+    domain_by_data_path = {}
+    for domain, data_paths in observation_catalog.data_paths_by_domain().items():
+        for data_path in data_paths:
+            if data_path in domain_by_data_path:
+                raise ValueError(f"Data_Path {data_path!r} is assigned to multiple domains.")
+            domain_by_data_path[data_path] = domain
+    return domain_by_data_path
+
+
+def _signal_from_variable(
+    variable: TestMqlVariableBinding,
+    domain: str,
+) -> TestMqlOutputSignal:
+    return TestMqlOutputSignal(
+        data_path=variable.data_path,
+        domain=domain,
+        owner_alias=variable.owner_alias,
+        owner_kind=variable.owner_kind,
+        submodel=variable.submodel,
+        signal_name=variable.signal_name,
+        units=variable.units,
+        amesim_index=variable.index,
+        saved=variable.saved,
+    )

tests/test_test_mql_output_schema.py

@@ -0,0 +1,91 @@
+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_observations import build_test_mql_observation_catalog
+from PythonModels.reporting.test_mql_output_schema import build_test_mql_output_schema
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame"
+
+
+class TestMqlOutputSchemaTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME)
+        cls.observation_catalog = build_test_mql_observation_catalog(cls.amesim_results)
+        cls.schema = build_test_mql_output_schema(
+            cls.amesim_results,
+            observation_catalog=cls.observation_catalog,
+        )
+
+    def test_schema_contains_observed_output_data_paths(self) -> None:
+        self.assertEqual(self.schema.signal_count, 858)
+        self.assertEqual(self.schema.data_paths(), self.observation_catalog.data_paths())
+        self.assertEqual(len(set(self.schema.data_paths())), self.schema.signal_count)
+
+    def test_counts_by_domain_owner_kind_and_submodel(self) -> None:
+        self.assertEqual(
+            self.schema.counts_by_domain(),
+            {"chambers": 100, "lines": 388, "mechanical": 266, "orifices": 104},
+        )
+        self.assertEqual(
+            self.schema.counts_by_owner_kind(),
+            {"component": 470, "connection": 388},
+        )
+        self.assertEqual(self.schema.counts_by_submodel()["PNCH012"], 80)
+        self.assertEqual(self.schema.counts_by_submodel()["PNCH023"], 20)
+        self.assertEqual(self.schema.counts_by_submodel()["PNL0001"], 180)
+        self.assertEqual(self.schema.counts_by_submodel()["MECMAS21"], 110)
+        self.assertEqual(self.schema.counts_by_submodel()["LMECHN1"], 34)
+
+    def test_counts_by_units_preserve_amesim_units(self) -> None:
+        counts = self.schema.counts_by_units()
+
+        self.assertEqual(counts["m/s"], 108)
+        self.assertEqual(counts[None], 98)
+        self.assertEqual(counts["N"], 92)
+        self.assertEqual(counts["J/s"], 84)
+        self.assertEqual(counts["g/s"], 84)
+        self.assertEqual(counts["Pa"], 84)
+        self.assertEqual(counts["K"], 84)
+        self.assertEqual(counts["cm**3"], 16)
+        self.assertEqual(counts["L/min"], 8)
+
+    def test_typical_signal_metadata_is_available_by_data_path(self) -> None:
+        chamber = self.schema.by_data_path("press@pn_general_chamber")
+        line = self.schema.by_data_path("dm1@pneumatic_69")
+        mechanical = self.schema.by_data_path("tforce@dynamic_mechanical_node_alternative_2")
+
+        self.assertEqual(chamber.domain, "chambers")
+        self.assertEqual(chamber.owner_alias, "pn_general_chamber")
+        self.assertEqual(chamber.owner_kind, "component")
+        self.assertEqual(chamber.submodel, "PNCH023")
+        self.assertEqual(chamber.signal_name, "press")
+        self.assertEqual(chamber.units, "Pa")
+        self.assertTrue(chamber.saved)
+        self.assertEqual(line.domain, "lines")
+        self.assertEqual(line.owner_kind, "connection")
+        self.assertEqual(line.submodel, "PNL0001")
+        self.assertEqual(line.units, "g/s")
+        self.assertEqual(mechanical.domain, "mechanical")
+        self.assertEqual(mechanical.submodel, "LMECHN1")
+        self.assertEqual(mechanical.units, "N")
+
+    def test_data_paths_can_be_selected_by_domain(self) -> None:
+        self.assertEqual(len(self.schema.data_paths_by_domain("chambers")), 100)
+        self.assertEqual(len(self.schema.data_paths_by_domain("orifices")), 104)
+        self.assertEqual(len(self.schema.data_paths_by_domain("lines")), 388)
+        self.assertEqual(len(self.schema.data_paths_by_domain("mechanical")), 266)
+        self.assertEqual(self.schema.data_paths_by_domain("missing"), ())
+
+    def test_unknown_data_path_raises_key_error(self) -> None:
+        with self.assertRaises(KeyError):
+            self.schema.by_data_path("not_a_signal@not_an_owner")
+
+
+if __name__ == "__main__":
+    unittest.main()