PythonModels/systems/test_mql_baseline.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from PythonModels.reporting.amesim_results import AmesimResults, load_test_mql_amesim_results
+from PythonModels.reporting.test_mql_comparison import TestMqlComparisonResult
+from PythonModels.reporting.test_mql_observations import (
+    TestMqlObservationCatalog,
+    build_test_mql_observation_catalog,
+)
+from PythonModels.reporting.test_mql_output_schema import (
+    TestMqlOutputSchema,
+    build_test_mql_output_schema,
+)
+from PythonModels.reporting.test_mql_output_validation import (
+    TestMqlValidatedOutput,
+    compare_validated_test_mql_output,
+    validate_test_mql_output,
+)
+
+
+@dataclass(frozen=True)
+class TestMqlBaselineRun:
+    amesim_results: AmesimResults
+    observation_catalog: TestMqlObservationCatalog
+    output_schema: TestMqlOutputSchema
+    output: TestMqlValidatedOutput
+    comparison: TestMqlComparisonResult
+
+    @property
+    def sample_count(self) -> int:
+        return len(self.output.times)
+
+    @property
+    def signal_count(self) -> int:
+        return len(self.output.data_paths)
+
+
+def run_test_mql_baseline_passthrough(
+    archive_path: Path,
+    *,
+    data_paths: tuple[str, ...] | list[str] | None = None,
+) -> TestMqlBaselineRun:
+    amesim_results = load_test_mql_amesim_results(archive_path)
+    observation_catalog = build_test_mql_observation_catalog(amesim_results)
+    output_schema = build_test_mql_output_schema(
+        amesim_results,
+        observation_catalog=observation_catalog,
+    )
+    selected_paths = tuple(data_paths) if data_paths is not None else output_schema.data_paths()
+    baseline_series = observation_catalog.baseline_series_by_data_path(
+        amesim_results,
+        selected_paths,
+    )
+    output = validate_test_mql_output(
+        times=amesim_results.times,
+        series_by_data_path=baseline_series,
+        schema=output_schema,
+        data_paths=selected_paths,
+        require_all_schema_paths=data_paths is None,
+    )
+    comparison = compare_validated_test_mql_output(
+        times=output.times,
+        series_by_data_path=output.series_by_data_path,
+        schema=output_schema,
+        amesim_results=amesim_results,
+        data_paths=output.data_paths,
+        require_all_schema_paths=data_paths is None,
+    )
+    return TestMqlBaselineRun(
+        amesim_results=amesim_results,
+        observation_catalog=observation_catalog,
+        output_schema=output_schema,
+        output=output,
+        comparison=comparison,
+    )

tests/test_test_mql_baseline.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+
+from PythonModels.systems.test_mql_baseline import run_test_mql_baseline_passthrough
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame"
+
+
+class TestMqlBaselinePassthroughTests(unittest.TestCase):
+    def test_full_baseline_passthrough_satisfies_output_contract_with_zero_error(self) -> None:
+        run = run_test_mql_baseline_passthrough(TEST_MQL_AME)
+
+        self.assertEqual(run.sample_count, 1002)
+        self.assertEqual(run.signal_count, 858)
+        self.assertEqual(run.output_schema.signal_count, 858)
+        self.assertEqual(run.observation_catalog.binding_count, 114)
+        self.assertEqual(len(run.comparison.metrics), 858)
+        self.assertEqual(run.comparison.max_abs_error, 0.0)
+        self.assertEqual(run.comparison.max_rel_error, 0.0)
+        self.assertEqual(run.output.data_paths, run.output_schema.data_paths())
+        self.assertAlmostEqual(run.output.series("press@pn_c1_8")[-1], 4310310.457796034)
+
+    def test_selected_baseline_passthrough_preserves_requested_order(self) -> None:
+        data_paths = (
+            "vol1@pn_brp2_8",
+            "press@pn_c1_8",
+            "dm1@pneumatic_69",
+        )
+
+        run = run_test_mql_baseline_passthrough(TEST_MQL_AME, data_paths=data_paths)
+
+        self.assertEqual(run.signal_count, 3)
+        self.assertEqual(run.output.data_paths, data_paths)
+        self.assertEqual(tuple(metric.data_path for metric in run.comparison.metrics), data_paths)
+        self.assertEqual(run.comparison.max_abs_error, 0.0)
+
+    def test_baseline_passthrough_rejects_unobserved_data_path(self) -> None:
+        with self.assertRaises(KeyError):
+            run_test_mql_baseline_passthrough(
+                TEST_MQL_AME,
+                data_paths=("not_a_signal@not_an_owner",),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()