diff --git a/PythonModels/systems/test_mql_baseline.py b/PythonModels/systems/test_mql_baseline.py new file mode 100644 index 0000000..fff6c38 --- /dev/null +++ b/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, + ) diff --git a/tests/test_test_mql_baseline.py b/tests/test_test_mql_baseline.py new file mode 100644 index 0000000..cb82017 --- /dev/null +++ b/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()