AmesimModels/test_mql/README.md

@@ -78,6 +78,14 @@
 - `tests/test_test_mql_variables.py`
   - 保护变量目录覆盖率、组件变量映射、连接变量映射、子模型计数和非法 `Data_Path` 报错。
 
+- `PythonModels/systems/test_mql_lines.py`
+  - 把 AMESim `CONNECTION_SPECS` 中的管路连接装配成 Python 管路目录。
+  - 当前覆盖 `PNL0001` 20 条、`PNL0002` 8 条、`PNL0003` 8 条、`PNL00R` 4 条，共 40 条。
+  - 保留 `pneumatic_*` alias、源/目标组件端口、`C-R`/`R-C-R`/`C-R-C`/`R` 模式和对应 `Data_Path`。
+
+- `tests/test_test_mql_lines.py`
+  - 保护管路数量、拓扑端口、管路模式、典型 `Data_Path` 和子模型分组。
+
 ## 物性约定
 
 AMESim 模型中 `test_mql` 使用氦气，Python 侧当前通过 `HELIUM_PR` 使用 Peng-Robinson 状态方程计算气体压缩因子和密度。当前物性层先覆盖状态方程相关量，完整焓/内能偏差函数后续在接气室能量方程时再补。
@@ -124,12 +132,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/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 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_lines.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/systems/test_mql_lines.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import re
+from collections import Counter
+from dataclasses import dataclass
+
+from PythonModels.reporting.amesim_results import AmesimResults
+from PythonModels.reporting.test_mql_variables import (
+    TestMqlVariableCatalog,
+    build_test_mql_variable_catalog,
+)
+from PythonModels.systems.test_mql import CONNECTION_SPECS
+
+
+TEST_MQL_PNEUMATIC_LINE_SUBMODELS = ("PNL0001", "PNL0002", "PNL0003", "PNL00R")
+_LINE_PATTERN_RE = re.compile(r"\(([^()]+)\)\s*$")
+
+
+@dataclass(frozen=True)
+class TestMqlLineConnection:
+    index: int
+    alias: str
+    submodel: str
+    pattern: str
+    source_component: str
+    source_port: str
+    target_component: str
+    target_port: str
+    label: str
+    data_paths: tuple[str, ...]
+    signal_names: tuple[str, ...]
+
+    @property
+    def has_compliance(self) -> bool:
+        return "C" in self.pattern
+
+    @property
+    def has_resistance(self) -> bool:
+        return "R" in self.pattern
+
+
+@dataclass(frozen=True)
+class TestMqlLineAssembly:
+    lines: tuple[TestMqlLineConnection, ...]
+
+    @property
+    def line_count(self) -> int:
+        return len(self.lines)
+
+    def by_alias(self, alias: str) -> TestMqlLineConnection:
+        for line in self.lines:
+            if line.alias == alias:
+                return line
+        raise KeyError(alias)
+
+    def by_submodel(self, submodel: str) -> tuple[TestMqlLineConnection, ...]:
+        return tuple(line for line in self.lines if line.submodel == submodel)
+
+    def counts_by_submodel(self) -> dict[str, int]:
+        return dict(Counter(line.submodel for line in self.lines))
+
+    def aliases(self) -> tuple[str, ...]:
+        return tuple(line.alias for line in self.lines)
+
+
+def build_test_mql_line_assembly(
+    amesim_results: AmesimResults,
+    variable_catalog: TestMqlVariableCatalog | None = None,
+) -> TestMqlLineAssembly:
+    variable_catalog = variable_catalog or build_test_mql_variable_catalog(amesim_results)
+    lines = []
+    for spec in CONNECTION_SPECS:
+        submodel = str(spec["submodel"])
+        if submodel not in TEST_MQL_PNEUMATIC_LINE_SUBMODELS:
+            continue
+        data_paths = variable_catalog.data_paths_for_owner(str(spec["alias"]))
+        signal_names = tuple(path.rsplit("@", 1)[0] for path in data_paths)
+        lines.append(
+            TestMqlLineConnection(
+                index=int(spec["index"]),
+                alias=str(spec["alias"]),
+                submodel=submodel,
+                pattern=_line_pattern(str(spec["label"]), submodel),
+                source_component=str(spec["source_component"]),
+                source_port=str(spec["source_port"]),
+                target_component=str(spec["target_component"]),
+                target_port=str(spec["target_port"]),
+                label=str(spec["label"]),
+                data_paths=data_paths,
+                signal_names=signal_names,
+            )
+        )
+    return TestMqlLineAssembly(lines=tuple(lines))
+
+
+def _line_pattern(label: str, submodel: str) -> str:
+    match = _LINE_PATTERN_RE.search(label)
+    if match is not None:
+        return match.group(1)
+    if submodel == "PNL00R":
+        return "R"
+    return submodel

tests/test_test_mql_lines.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
+from PythonModels.systems.test_mql_lines import build_test_mql_line_assembly
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame"
+
+
+class TestMqlLineAssemblyTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME)
+        cls.variable_catalog = build_test_mql_variable_catalog(cls.amesim_results)
+        cls.assembly = build_test_mql_line_assembly(
+            cls.amesim_results,
+            cls.variable_catalog,
+        )
+
+    def test_assembles_expected_line_counts(self) -> None:
+        self.assertEqual(self.assembly.line_count, 40)
+        self.assertEqual(
+            self.assembly.counts_by_submodel(),
+            {"PNL0001": 20, "PNL0003": 8, "PNL0002": 8, "PNL00R": 4},
+        )
+        self.assertIn("pneumatic_69", self.assembly.aliases())
+        self.assertIn("pneumatic_100", self.assembly.aliases())
+
+    def test_preserves_connection_topology(self) -> None:
+        line = self.assembly.by_alias("pneumatic_69")
+
+        self.assertEqual(line.index, 37)
+        self.assertEqual(line.submodel, "PNL0001")
+        self.assertEqual(line.pattern, "C-R")
+        self.assertEqual(line.source_component, "pnnode4_16")
+        self.assertEqual(line.source_port, "port_2")
+        self.assertEqual(line.target_component, "pn_c1_8")
+        self.assertEqual(line.target_port, "port_1")
+        self.assertTrue(line.has_compliance)
+        self.assertTrue(line.has_resistance)
+
+    def test_line_data_paths_are_attached_from_amesim_results(self) -> None:
+        line = self.assembly.by_alias("pneumatic_69")
+
+        self.assertIn("dh1@pneumatic_69", line.data_paths)
+        self.assertIn("dm1@pneumatic_69", line.data_paths)
+        self.assertIn("dh1", line.signal_names)
+        self.assertIn("dm1", line.signal_names)
+
+    def test_resistance_only_line_keeps_r_pattern(self) -> None:
+        line = self.assembly.by_alias("pneumatic_100")
+
+        self.assertEqual(line.submodel, "PNL00R")
+        self.assertEqual(line.pattern, "R")
+        self.assertFalse(line.has_compliance)
+        self.assertTrue(line.has_resistance)
+        self.assertIn("dm1@pneumatic_100", line.data_paths)
+
+    def test_rcr_and_crc_patterns_are_distinct(self) -> None:
+        self.assertEqual(self.assembly.by_alias("pneumatic_80").pattern, "R-C-R")
+        self.assertEqual(self.assembly.by_alias("pneumatic_75").pattern, "C-R-C")
+        self.assertEqual(len(self.assembly.by_submodel("PNL0002")), 8)
+        self.assertEqual(len(self.assembly.by_submodel("PNL0003")), 8)
+
+
+if __name__ == "__main__":
+    unittest.main()