diff --git a/PythonModels/reporting/test_mql_observations.py b/PythonModels/reporting/test_mql_observations.py new file mode 100644 index 0000000..fe48070 --- /dev/null +++ b/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))) diff --git a/tests/test_test_mql_observations.py b/tests/test_test_mql_observations.py new file mode 100644 index 0000000..47b3dd5 --- /dev/null +++ b/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()