PythonModels/reporting/test_mql_line_observations.py

@@ -0,0 +1,211 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from PythonModels.reporting.amesim_results import AmesimResults
+from PythonModels.reporting.test_mql_variables import (
+    TestMqlVariableBinding,
+    TestMqlVariableCatalog,
+    build_test_mql_variable_catalog,
+)
+from PythonModels.systems.test_mql_lines import (
+    TestMqlLineAssembly,
+    build_test_mql_line_assembly,
+)
+
+
+G_PER_S_TO_KG_PER_S = 1.0e-3
+
+
+@dataclass(frozen=True)
+class TestMqlLineObservation:
+    time: float
+    mass_flows_kg_s: dict[str, float]
+    enthalpy_flows_w: dict[str, float]
+    pressures_pa: dict[str, float]
+    temperatures_k: dict[str, float]
+    gas_mass_g: float | None
+    reynolds_number: float
+    mass_flow_parameter: float
+    gas_velocity_m_s: float
+    friction_factor: float
+
+
+@dataclass(frozen=True)
+class TestMqlLineObservationBinding:
+    alias: str
+    submodel: str
+    pattern: str
+    mass_flow_paths: tuple[str, ...]
+    enthalpy_flow_paths: tuple[str, ...]
+    pressure_paths: tuple[str, ...]
+    temperature_paths: tuple[str, ...]
+    gas_mass_path: str | None
+    reynolds_path: str
+    mass_flow_parameter_path: str
+    gas_velocity_path: str
+    friction_factor_path: str
+
+    def mass_flow_kg_s_series(
+        self,
+        results: AmesimResults,
+        data_path: str | None = None,
+    ) -> tuple[float, ...]:
+        path = data_path or self.mass_flow_paths[0]
+        if path not in self.mass_flow_paths:
+            raise KeyError(path)
+        return tuple(value * G_PER_S_TO_KG_PER_S for value in results.series(path))
+
+    def observation_at(self, results: AmesimResults, index: int) -> TestMqlLineObservation:
+        return TestMqlLineObservation(
+            time=results.times[index],
+            mass_flows_kg_s={
+                path: results.series(path)[index] * G_PER_S_TO_KG_PER_S
+                for path in self.mass_flow_paths
+            },
+            enthalpy_flows_w={
+                path: results.series(path)[index]
+                for path in self.enthalpy_flow_paths
+            },
+            pressures_pa={
+                path: results.series(path)[index]
+                for path in self.pressure_paths
+            },
+            temperatures_k={
+                path: results.series(path)[index]
+                for path in self.temperature_paths
+            },
+            gas_mass_g=(
+                results.series(self.gas_mass_path)[index]
+                if self.gas_mass_path is not None
+                else None
+            ),
+            reynolds_number=results.series(self.reynolds_path)[index],
+            mass_flow_parameter=results.series(self.mass_flow_parameter_path)[index],
+            gas_velocity_m_s=results.series(self.gas_velocity_path)[index],
+            friction_factor=results.series(self.friction_factor_path)[index],
+        )
+
+
+@dataclass(frozen=True)
+class TestMqlLineObservationCatalog:
+    bindings: tuple[TestMqlLineObservationBinding, ...]
+
+    @property
+    def line_count(self) -> int:
+        return len(self.bindings)
+
+    def by_alias(self, alias: str) -> TestMqlLineObservationBinding:
+        for binding in self.bindings:
+            if binding.alias == alias:
+                return binding
+        raise KeyError(alias)
+
+    def by_submodel(self, submodel: str) -> tuple[TestMqlLineObservationBinding, ...]:
+        return tuple(binding for binding in self.bindings if binding.submodel == submodel)
+
+
+def build_test_mql_line_observation_catalog(
+    results: AmesimResults,
+    *,
+    variable_catalog: TestMqlVariableCatalog | None = None,
+    line_assembly: TestMqlLineAssembly | None = None,
+) -> TestMqlLineObservationCatalog:
+    variable_catalog = variable_catalog or build_test_mql_variable_catalog(results)
+    line_assembly = line_assembly or build_test_mql_line_assembly(results, variable_catalog)
+    bindings = []
+    for line in line_assembly.lines:
+        variables = tuple(
+            variable
+            for variable in variable_catalog.variables
+            if variable.owner_alias == line.alias
+        )
+        bindings.append(
+            TestMqlLineObservationBinding(
+                alias=line.alias,
+                submodel=line.submodel,
+                pattern=line.pattern,
+                mass_flow_paths=_paths_with_prefix(variables, "dm", expected_units="g/s"),
+                enthalpy_flow_paths=_paths_with_prefix(variables, "dh", expected_units="J/s"),
+                pressure_paths=_paths_with_prefix(variables, "p", expected_units="Pa"),
+                temperature_paths=_paths_with_prefix(variables, "t", expected_units="K"),
+                gas_mass_path=_optional_path(variables, "mgas", expected_units="g"),
+                reynolds_path=_required_path(variables, "re", expected_units=None),
+                mass_flow_parameter_path=_required_path(
+                    variables,
+                    "cm",
+                    expected_units="(kg*K/J)**(1/2)",
+                ),
+                gas_velocity_path=_required_path(variables, "v", expected_units="m/s"),
+                friction_factor_path=_required_path(variables, "ff", expected_units=None),
+            )
+        )
+    return TestMqlLineObservationCatalog(bindings=tuple(bindings))
+
+
+def _paths_with_prefix(
+    variables: tuple[TestMqlVariableBinding, ...],
+    prefix: str,
+    *,
+    expected_units: str | None,
+) -> tuple[str, ...]:
+    matches = tuple(
+        variable
+        for variable in variables
+        if variable.signal_name.startswith(prefix)
+    )
+    for variable in matches:
+        _assert_units(variable, expected_units)
+    return tuple(variable.data_path for variable in matches)
+
+
+def _required_path(
+    variables: tuple[TestMqlVariableBinding, ...],
+    signal_name: str,
+    *,
+    expected_units: str | None,
+) -> str:
+    variable = _single_signal(variables, signal_name)
+    _assert_units(variable, expected_units)
+    return variable.data_path
+
+
+def _optional_path(
+    variables: tuple[TestMqlVariableBinding, ...],
+    signal_name: str,
+    *,
+    expected_units: str | None,
+) -> str | None:
+    matches = tuple(variable for variable in variables if variable.signal_name == signal_name)
+    if not matches:
+        return None
+    variable = _single(matches, signal_name)
+    _assert_units(variable, expected_units)
+    return variable.data_path
+
+
+def _single_signal(
+    variables: tuple[TestMqlVariableBinding, ...],
+    signal_name: str,
+) -> TestMqlVariableBinding:
+    return _single(
+        tuple(variable for variable in variables if variable.signal_name == signal_name),
+        signal_name,
+    )
+
+
+def _single(
+    matches: tuple[TestMqlVariableBinding, ...],
+    description: str,
+) -> TestMqlVariableBinding:
+    if len(matches) != 1:
+        raise ValueError(f"Expected one {description} variable, found {len(matches)}.")
+    return matches[0]
+
+
+def _assert_units(variable: TestMqlVariableBinding, expected_units: str | None) -> None:
+    if variable.units != expected_units:
+        raise ValueError(
+            f"Unexpected units for {variable.data_path}: "
+            f"{variable.units!r}, expected {expected_units!r}."
+        )

tests/test_test_mql_line_observations.py

@@ -0,0 +1,111 @@
+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_line_observations import (
+    G_PER_S_TO_KG_PER_S,
+    build_test_mql_line_observation_catalog,
+)
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame"
+
+
+class TestMqlLineObservationCatalogTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME)
+        cls.catalog = build_test_mql_line_observation_catalog(cls.amesim_results)
+
+    def test_builds_expected_line_observation_counts(self) -> None:
+        self.assertEqual(self.catalog.line_count, 40)
+        self.assertEqual(len(self.catalog.by_submodel("PNL0001")), 20)
+        self.assertEqual(len(self.catalog.by_submodel("PNL0002")), 8)
+        self.assertEqual(len(self.catalog.by_submodel("PNL0003")), 8)
+        self.assertEqual(len(self.catalog.by_submodel("PNL00R")), 4)
+
+    def test_pnl0001_binding_exposes_single_flow_and_compliance_observables(self) -> None:
+        binding = self.catalog.by_alias("pneumatic_69")
+
+        self.assertEqual(binding.submodel, "PNL0001")
+        self.assertEqual(binding.pattern, "C-R")
+        self.assertEqual(binding.mass_flow_paths, ("dm1@pneumatic_69",))
+        self.assertEqual(binding.enthalpy_flow_paths, ("dh1@pneumatic_69",))
+        self.assertEqual(binding.pressure_paths, ("p2@pneumatic_69",))
+        self.assertEqual(binding.temperature_paths, ("t2@pneumatic_69",))
+        self.assertEqual(binding.gas_mass_path, "mgas@pneumatic_69")
+        self.assertEqual(binding.reynolds_path, "re@pneumatic_69")
+        self.assertEqual(binding.mass_flow_parameter_path, "cm@pneumatic_69")
+        self.assertEqual(binding.gas_velocity_path, "v@pneumatic_69")
+        self.assertEqual(binding.friction_factor_path, "ff@pneumatic_69")
+
+    def test_pnl0002_and_pnl0003_keep_distinct_flow_topologies(self) -> None:
+        rcr = self.catalog.by_alias("pneumatic_80")
+        crc = self.catalog.by_alias("pneumatic_75")
+
+        self.assertEqual(rcr.pattern, "R-C-R")
+        self.assertEqual(rcr.mass_flow_paths, ("dm1@pneumatic_80", "dm2@pneumatic_80"))
+        self.assertEqual(rcr.enthalpy_flow_paths, ("dh1@pneumatic_80", "dh2@pneumatic_80"))
+        self.assertEqual(rcr.pressure_paths, ("pctr@pneumatic_80",))
+        self.assertEqual(rcr.temperature_paths, ("tctr@pneumatic_80",))
+        self.assertEqual(crc.pattern, "C-R-C")
+        self.assertEqual(crc.mass_flow_paths, ("dmctr@pneumatic_75",))
+        self.assertEqual(crc.enthalpy_flow_paths, ("dhctr@pneumatic_75",))
+        self.assertEqual(crc.pressure_paths, ("p1@pneumatic_75", "p2@pneumatic_75"))
+        self.assertEqual(crc.temperature_paths, ("t1@pneumatic_75", "t2@pneumatic_75"))
+
+    def test_mass_flow_series_are_converted_to_si_units(self) -> None:
+        binding = self.catalog.by_alias("pneumatic_69")
+        mass_flow = binding.mass_flow_kg_s_series(self.amesim_results)
+
+        self.assertAlmostEqual(
+            mass_flow[-1],
+            self.amesim_results.series("dm1@pneumatic_69")[-1] * G_PER_S_TO_KG_PER_S,
+        )
+        self.assertAlmostEqual(mass_flow[-1], -9.527778353091475e-6)
+        self.assertAlmostEqual(max(mass_flow), 0.4490761730524304)
+
+    def test_observation_exposes_line_state_and_diagnostics(self) -> None:
+        observation = self.catalog.by_alias("pneumatic_75").observation_at(
+            self.amesim_results,
+            -1,
+        )
+
+        self.assertAlmostEqual(observation.time, self.amesim_results.times[-1])
+        self.assertAlmostEqual(
+            observation.mass_flows_kg_s["dmctr@pneumatic_75"],
+            9.486502325625855e-6,
+        )
+        self.assertAlmostEqual(
+            observation.enthalpy_flows_w["dhctr@pneumatic_75"],
+            -0.7322143070637952,
+        )
+        self.assertAlmostEqual(observation.pressures_pa["p1@pneumatic_75"], 4310289.672688478)
+        self.assertAlmostEqual(observation.pressures_pa["p2@pneumatic_75"], 4310295.447776705)
+        self.assertAlmostEqual(observation.temperatures_k["t1@pneumatic_75"], 281.37837485614546)
+        self.assertAlmostEqual(observation.temperatures_k["t2@pneumatic_75"], 282.9136549991407)
+        self.assertAlmostEqual(observation.gas_mass_g, 0.7031236652270819)
+        self.assertAlmostEqual(observation.reynolds_number, 31.539799199000434)
+        self.assertAlmostEqual(observation.mass_flow_parameter, 6.351754064211156e-07)
+        self.assertAlmostEqual(observation.gas_velocity_m_s, 0.0040585650473562935)
+        self.assertAlmostEqual(observation.friction_factor, 2.0291822277051246)
+
+    def test_resistance_only_lines_preserve_reversed_duplicate_signs(self) -> None:
+        for binding in self.catalog.by_submodel("PNL00R"):
+            dm1 = self.amesim_results.series(f"dm1@{binding.alias}")
+            dm2 = self.amesim_results.series(f"dm2@{binding.alias}")
+            dh1 = self.amesim_results.series(f"dh1@{binding.alias}")
+            dh2 = self.amesim_results.series(f"dh2@{binding.alias}")
+            with self.subTest(alias=binding.alias):
+                self.assertIsNone(binding.gas_mass_path)
+                self.assertEqual(binding.pressure_paths, ())
+                self.assertEqual(binding.temperature_paths, ())
+                self.assertLess(max(abs(a + b) for a, b in zip(dm1, dm2)), 1.0e-36)
+                self.assertLess(max(abs(a + b) for a, b in zip(dh1, dh2)), 1.0e-34)
+
+
+if __name__ == "__main__":
+    unittest.main()