AmesimModels/test_mql/README.md

@@ -71,6 +71,13 @@
 - `tests/test_test_mql_pneumatic.py`
   - 保护气动装配数量、alias、氦气工质、气室体积、孔口面积/流量系数和基础质量流量计算。
 
+- `PythonModels/reporting/test_mql_variables.py`
+  - 建立 AMESim `Data_Path` 到信号名、组件/连接 alias、owner 类型和子模型的变量目录。
+  - 当前 1100 个有 `Data_Path` 的变量全部可归属，其中组件变量 712 个、连接/管路变量 388 个。
+
+- `tests/test_test_mql_variables.py`
+  - 保护变量目录覆盖率、组件变量映射、连接变量映射、子模型计数和非法 `Data_Path` 报错。
+
 ## 物性约定
 
 AMESim 模型中 `test_mql` 使用氦气，Python 侧当前通过 `HELIUM_PR` 使用 Peng-Robinson 状态方程计算气体压缩因子和密度。当前物性层先覆盖状态方程相关量，完整焓/内能偏差函数后续在接气室能量方程时再补。
@@ -117,12 +124,12 @@ AMESim 模型中 `test_mql` 使用氦气，Python 侧当前通过 `HELIUM_PR`
 ## 验证方式
 
 ```bash
-python3 -m py_compile PythonModels/components/amesim_pneumatic.py PythonModels/core/peng_robinson.py PythonModels/reporting/amesim_results.py PythonModels/reporting/test_mql_comparison.py PythonModels/systems/test_mql.py PythonModels/systems/test_mql_config.py PythonModels/systems/test_mql_pneumatic.py PythonModels/scripts/run_test_mql.py
+python3 -m py_compile PythonModels/components/amesim_pneumatic.py PythonModels/core/peng_robinson.py PythonModels/reporting/amesim_results.py PythonModels/reporting/test_mql_comparison.py PythonModels/reporting/test_mql_variables.py PythonModels/systems/test_mql.py PythonModels/systems/test_mql_config.py PythonModels/systems/test_mql_pneumatic.py PythonModels/scripts/run_test_mql.py
 python3 -m PythonModels.scripts.run_test_mql
 python3 -m unittest discover -s tests -t .
 ```
 
-当前测试覆盖的是结构、配置解析、气动原语、气动件装配、AMESim 结果读取和 AMESim/Python 时序对齐工具，不代表已经完成 AMESim 物理结果复刻。
+当前测试覆盖的是结构、配置解析、气动原语、气动件装配、AMESim 结果读取、变量目录和 AMESim/Python 时序对齐工具，不代表已经完成 AMESim 物理结果复刻。
 
 ## 后续方向
 

PythonModels/reporting/test_mql_variables.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import re
+from collections import Counter
+from dataclasses import dataclass
+
+from PythonModels.reporting.amesim_results import AmesimResults, AmesimVariable
+from PythonModels.systems.test_mql import COMPONENT_SPECS, CONNECTION_SPECS
+
+
+_UNIT_RE = re.compile(r"\[([^\]]+)\]\s*$")
+
+
+@dataclass(frozen=True)
+class TestMqlVariableBinding:
+    index: int
+    data_path: str
+    signal_name: str
+    owner_alias: str
+    owner_kind: str
+    submodel: str
+    label: str
+    units: str | None
+    saved: bool
+
+
+@dataclass(frozen=True)
+class TestMqlVariableCatalog:
+    variables: tuple[TestMqlVariableBinding, ...]
+
+    @property
+    def data_path_count(self) -> int:
+        return len(self.variables)
+
+    @property
+    def saved_data_path_count(self) -> int:
+        return sum(1 for variable in self.variables if variable.saved)
+
+    def by_data_path(self, data_path: str) -> TestMqlVariableBinding:
+        for variable in self.variables:
+            if variable.data_path == data_path:
+                return variable
+        raise KeyError(data_path)
+
+    def counts_by_submodel(self) -> dict[str, int]:
+        return dict(Counter(variable.submodel for variable in self.variables))
+
+    def counts_by_owner_kind(self) -> dict[str, int]:
+        return dict(Counter(variable.owner_kind for variable in self.variables))
+
+    def data_paths_for_owner(self, owner_alias: str) -> tuple[str, ...]:
+        return tuple(
+            variable.data_path
+            for variable in self.variables
+            if variable.owner_alias == owner_alias
+        )
+
+
+def build_test_mql_variable_catalog(amesim_results: AmesimResults) -> TestMqlVariableCatalog:
+    owner_map = _build_owner_map()
+    saved_indices = set(amesim_results.saved_variable_indices)
+    bindings = []
+    for variable in amesim_results.variables:
+        if variable.data_path is None:
+            continue
+        signal_name, owner_alias = split_data_path(variable.data_path)
+        owner_kind, submodel = owner_map[owner_alias]
+        bindings.append(
+            TestMqlVariableBinding(
+                index=variable.index,
+                data_path=variable.data_path,
+                signal_name=signal_name,
+                owner_alias=owner_alias,
+                owner_kind=owner_kind,
+                submodel=submodel,
+                label=variable.label,
+                units=_extract_units(variable),
+                saved=variable.index in saved_indices,
+            )
+        )
+    return TestMqlVariableCatalog(variables=tuple(bindings))
+
+
+def split_data_path(data_path: str) -> tuple[str, str]:
+    if "@" not in data_path:
+        raise ValueError(f"AMESim Data_Path does not contain an owner alias: {data_path!r}")
+    signal_name, owner_alias = data_path.rsplit("@", 1)
+    if not signal_name or not owner_alias:
+        raise ValueError(f"Invalid AMESim Data_Path: {data_path!r}")
+    return signal_name, owner_alias
+
+
+def _build_owner_map() -> dict[str, tuple[str, str]]:
+    owner_map = {
+        str(spec["alias"]): ("component", str(spec["submodel"]))
+        for spec in COMPONENT_SPECS
+    }
+    owner_map.update(
+        {
+            str(spec["alias"]): ("connection", str(spec["submodel"]))
+            for spec in CONNECTION_SPECS
+        }
+    )
+    return owner_map
+
+
+def _extract_units(variable: AmesimVariable) -> str | None:
+    match = _UNIT_RE.search(variable.label)
+    if match is None:
+        return None
+    return match.group(1)

tests/test_test_mql_variables.py

@@ -0,0 +1,72 @@
+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_variables import (
+    build_test_mql_variable_catalog,
+    split_data_path,
+)
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame"
+
+
+class TestMqlVariableCatalogTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME)
+        cls.catalog = build_test_mql_variable_catalog(cls.amesim_results)
+
+    def test_catalog_covers_all_amesim_data_paths(self) -> None:
+        self.assertEqual(self.catalog.data_path_count, 1100)
+        self.assertEqual(self.catalog.saved_data_path_count, 1100)
+        self.assertEqual(self.catalog.counts_by_owner_kind(), {"component": 712, "connection": 388})
+
+    def test_component_variable_mapping_preserves_alias_and_submodel(self) -> None:
+        variable = self.catalog.by_data_path("temp3@pn_c1_8")
+
+        self.assertEqual(variable.index, 0)
+        self.assertEqual(variable.signal_name, "temp3")
+        self.assertEqual(variable.owner_alias, "pn_c1_8")
+        self.assertEqual(variable.owner_kind, "component")
+        self.assertEqual(variable.submodel, "PNCH012")
+        self.assertEqual(variable.units, "K")
+        self.assertTrue(variable.saved)
+
+    def test_connection_variable_mapping_preserves_line_submodel(self) -> None:
+        variable = self.catalog.by_data_path("dm1@pneumatic_69")
+
+        self.assertEqual(variable.signal_name, "dm1")
+        self.assertEqual(variable.owner_alias, "pneumatic_69")
+        self.assertEqual(variable.owner_kind, "connection")
+        self.assertEqual(variable.submodel, "PNL0001")
+        self.assertEqual(variable.units, "g/s")
+
+    def test_counts_by_submodel_include_components_and_connections(self) -> None:
+        counts = self.catalog.counts_by_submodel()
+
+        self.assertEqual(counts["PNL0001"], 180)
+        self.assertEqual(counts["PNCH012"], 80)
+        self.assertEqual(counts["PNRP17"], 72)
+        self.assertEqual(counts["PNVO001"], 56)
+        self.assertEqual(counts["PNOR001"], 48)
+        self.assertEqual(counts["PNCH023"], 20)
+
+    def test_owner_lookup_returns_data_paths_for_component(self) -> None:
+        paths = self.catalog.data_paths_for_owner("pn_c1_8")
+
+        self.assertIn("temp3@pn_c1_8", paths)
+        self.assertIn("press3@pn_c1_8", paths)
+        self.assertEqual(len(paths), 10)
+
+    def test_split_data_path_rejects_invalid_values(self) -> None:
+        self.assertEqual(split_data_path("press3@pn_c1_8"), ("press3", "pn_c1_8"))
+        with self.assertRaises(ValueError):
+            split_data_path("press3")
+
+
+if __name__ == "__main__":
+    unittest.main()