From 9eb7d293f7a4439a389b0546b9781d7a90a32b63 Mon Sep 17 00:00:00 2001 From: huojiarong Date: Thu, 16 Jul 2026 02:27:42 +0000 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85test=5Fmql=E8=BE=93=E5=87=BA?= =?UTF-8?q?=E5=8F=98=E9=87=8F=E5=A5=91=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../reporting/test_mql_output_schema.py | 100 ++++++++++++++++++ tests/test_test_mql_output_schema.py | 91 ++++++++++++++++ 2 files changed, 191 insertions(+) create mode 100644 PythonModels/reporting/test_mql_output_schema.py create mode 100644 tests/test_test_mql_output_schema.py diff --git a/PythonModels/reporting/test_mql_output_schema.py b/PythonModels/reporting/test_mql_output_schema.py new file mode 100644 index 0000000..3a5661c --- /dev/null +++ b/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, + ) diff --git a/tests/test_test_mql_output_schema.py b/tests/test_test_mql_output_schema.py new file mode 100644 index 0000000..6c1b8d9 --- /dev/null +++ b/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()