AmesimModels/test_mql/README.md

@@ -55,6 +55,14 @@
 - `tests/test_amesim_results.py`
   - 保护 `test_mql_.results` 的时间轴、变量数量、Data_Path 映射和典型样本值。
 
+- `PythonModels/reporting/test_mql_comparison.py`
+  - 提供 `test_mql` 专用的 AMESim/Python 时序对齐工具。
+  - 按 AMESim `Data_Path` 做时间插值、最大绝对误差、平均绝对误差、最大相对误差和终值误差统计。
+  - 可导出 `test_mql_amesim_baseline.csv` 和 `test_mql_amesim_comparison.csv`，供后续组件方程校验使用。
+
+- `tests/test_test_mql_comparison.py`
+  - 保护 AMESim 原始序列零误差对齐、粗时间步插值、偏移误差统计、缺失变量报错和 CSV 输出。
+
 ## 物性约定
 
 AMESim 模型中 `test_mql` 使用氦气，Python 侧当前通过 `HELIUM_PR` 使用 Peng-Robinson 状态方程计算气体压缩因子和密度。当前物性层先覆盖状态方程相关量，完整焓/内能偏差函数后续在接气室能量方程时再补。
@@ -101,12 +109,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/systems/test_mql.py PythonModels/systems/test_mql_config.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/systems/test_mql.py PythonModels/systems/test_mql_config.py PythonModels/scripts/run_test_mql.py
 python3 -m PythonModels.scripts.run_test_mql
 python3 -m unittest discover -s tests -t .
 ```
 
-当前测试覆盖的是结构、配置解析、气动原语和 AMESim 结果读取，不代表已经完成 AMESim 物理结果复刻。
+当前测试覆盖的是结构、配置解析、气动原语、AMESim 结果读取和 AMESim/Python 时序对齐工具，不代表已经完成 AMESim 物理结果复刻。
 
 ## 后续方向
 
@@ -114,6 +122,6 @@ python3 -m unittest discover -s tests -t .
 
 1. `PNGD00`：气体属性。
 2. `PNCH023 / PNCH012`：固定气室和变容气室。
-3. `PNOR001 / PNVO001`：固定孔口和可变孔口。
+3. `PNOR001 / PNVO001`：固定孔口和可变孔口。每完成一项后用 `test_mql_comparison.py` 按 `Data_Path` 对齐 AMESim 结果。
 4. `PNL0001 / PNL0002 / PNL0003 / PNL00R`：管路阻容连接。
 5. `PNRP17 / MECMAS21 / LSTP00A / LMECHN1`：气动活塞、机械负载和端止动。

PythonModels/reporting/test_mql_comparison.py

@@ -0,0 +1,237 @@
+from __future__ import annotations
+
+from bisect import bisect_left
+import csv
+from dataclasses import dataclass
+from pathlib import Path
+
+from PythonModels.reporting.amesim_results import AmesimResults
+
+
+DEFAULT_TEST_MQL_ALIGNMENT_PATHS = (
+    "temp3@pn_c1_8",
+    "press3@pn_c1_8",
+    "vvol1@pn_brp2_8",
+    "vol1@pn_brp2_8",
+)
+
+
+@dataclass(frozen=True)
+class TestMqlComparisonMetric:
+    data_path: str
+    sample_count: int
+    max_abs_error: float
+    mean_abs_error: float
+    max_rel_error: float
+    final_abs_error: float
+
+
+@dataclass(frozen=True)
+class TestMqlComparisonResult:
+    metrics: tuple[TestMqlComparisonMetric, ...]
+
+    def metric(self, data_path: str) -> TestMqlComparisonMetric:
+        for metric in self.metrics:
+            if metric.data_path == data_path:
+                return metric
+        raise KeyError(data_path)
+
+    @property
+    def max_abs_error(self) -> float:
+        return max((metric.max_abs_error for metric in self.metrics), default=0.0)
+
+    @property
+    def max_rel_error(self) -> float:
+        return max((metric.max_rel_error for metric in self.metrics), default=0.0)
+
+
+class TestMqlComparisonError(ValueError):
+    """Raised when Python and AMESim series cannot be aligned."""
+
+
+def compare_test_mql_series(
+    *,
+    python_times: tuple[float, ...] | list[float],
+    python_series_by_data_path: dict[str, tuple[float, ...] | list[float]],
+    amesim_results: AmesimResults,
+    data_paths: tuple[str, ...] | list[str] | None = None,
+    relative_floor: float = 1.0e-12,
+) -> TestMqlComparisonResult:
+    _validate_time_axis(python_times)
+    selected_paths = _select_data_paths(python_series_by_data_path, amesim_results, data_paths)
+    metrics = []
+    for data_path in selected_paths:
+        python_values = tuple(float(value) for value in python_series_by_data_path[data_path])
+        if len(python_values) != len(python_times):
+            raise TestMqlComparisonError(
+                f"Python series length mismatch for {data_path!r}: "
+                f"{len(python_values)} values for {len(python_times)} time samples."
+            )
+        amesim_values = amesim_results.series(data_path)
+        abs_errors = []
+        rel_errors = []
+        for time_value, python_value in zip(python_times, python_values):
+            amesim_value = interpolate_series_value(amesim_results.times, amesim_values, time_value)
+            abs_error = abs(python_value - amesim_value)
+            abs_errors.append(abs_error)
+            rel_errors.append(abs_error / max(abs(amesim_value), relative_floor))
+        final_amesim_value = interpolate_series_value(
+            amesim_results.times,
+            amesim_values,
+            float(python_times[-1]),
+        )
+        metrics.append(
+            TestMqlComparisonMetric(
+                data_path=data_path,
+                sample_count=len(python_times),
+                max_abs_error=max(abs_errors, default=0.0),
+                mean_abs_error=sum(abs_errors) / max(len(abs_errors), 1),
+                max_rel_error=max(rel_errors, default=0.0),
+                final_abs_error=abs(python_values[-1] - final_amesim_value),
+            )
+        )
+    return TestMqlComparisonResult(metrics=tuple(metrics))
+
+
+def write_test_mql_amesim_baseline_csv(
+    output_dir: Path,
+    amesim_results: AmesimResults,
+    data_paths: tuple[str, ...] | list[str] = DEFAULT_TEST_MQL_ALIGNMENT_PATHS,
+) -> Path:
+    output_dir.mkdir(parents=True, exist_ok=True)
+    csv_path = output_dir / "test_mql_amesim_baseline.csv"
+    _validate_amesim_data_paths(amesim_results, data_paths)
+    with csv_path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.writer(handle)
+        writer.writerow(["time_s", *data_paths])
+        for index, time_value in enumerate(amesim_results.times):
+            writer.writerow(
+                [time_value, *(amesim_results.series(data_path)[index] for data_path in data_paths)]
+            )
+    return csv_path
+
+
+def write_test_mql_comparison_csv(
+    *,
+    output_dir: Path,
+    python_times: tuple[float, ...] | list[float],
+    python_series_by_data_path: dict[str, tuple[float, ...] | list[float]],
+    amesim_results: AmesimResults,
+    data_paths: tuple[str, ...] | list[str] | None = None,
+) -> tuple[Path, Path, TestMqlComparisonResult]:
+    output_dir.mkdir(parents=True, exist_ok=True)
+    selected_paths = _select_data_paths(python_series_by_data_path, amesim_results, data_paths)
+    comparison = compare_test_mql_series(
+        python_times=python_times,
+        python_series_by_data_path=python_series_by_data_path,
+        amesim_results=amesim_results,
+        data_paths=selected_paths,
+    )
+    csv_path = output_dir / "test_mql_amesim_comparison.csv"
+    summary_path = output_dir / "test_mql_amesim_comparison_summary.txt"
+
+    with csv_path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.writer(handle)
+        header = ["time_s"]
+        for data_path in selected_paths:
+            header.extend(
+                [
+                    f"python.{data_path}",
+                    f"amesim.{data_path}",
+                    f"abs_error.{data_path}",
+                    f"rel_error.{data_path}",
+                ]
+            )
+        writer.writerow(header)
+        for index, time_value in enumerate(python_times):
+            row = [time_value]
+            for data_path in selected_paths:
+                python_value = float(python_series_by_data_path[data_path][index])
+                amesim_value = interpolate_series_value(
+                    amesim_results.times,
+                    amesim_results.series(data_path),
+                    float(time_value),
+                )
+                abs_error = abs(python_value - amesim_value)
+                rel_error = abs_error / max(abs(amesim_value), 1.0e-12)
+                row.extend([python_value, amesim_value, abs_error, rel_error])
+            writer.writerow(row)
+
+    summary_lines = [
+        (
+            f"{metric.data_path}: samples={metric.sample_count}, "
+            f"max_abs_error={metric.max_abs_error:.12g}, "
+            f"mean_abs_error={metric.mean_abs_error:.12g}, "
+            f"max_rel_error={metric.max_rel_error:.12%}, "
+            f"final_abs_error={metric.final_abs_error:.12g}"
+        )
+        for metric in comparison.metrics
+    ]
+    summary_path.write_text("\n".join(summary_lines) + "\n", encoding="utf-8")
+    return csv_path, summary_path, comparison
+
+
+def interpolate_series_value(
+    time_values: tuple[float, ...] | list[float],
+    values: tuple[float, ...] | list[float],
+    target_time: float,
+) -> float:
+    if len(time_values) != len(values):
+        raise TestMqlComparisonError("time and value series lengths differ.")
+    if not time_values:
+        raise TestMqlComparisonError("cannot interpolate an empty series.")
+    if target_time <= time_values[0]:
+        return float(values[0])
+    if target_time >= time_values[-1]:
+        return float(values[-1])
+
+    right_index = bisect_left(time_values, target_time)
+    if right_index < len(time_values) and abs(time_values[right_index] - target_time) <= 1.0e-12:
+        return float(values[right_index])
+
+    left_index = right_index - 1
+    left_time = float(time_values[left_index])
+    right_time = float(time_values[right_index])
+    fraction = (target_time - left_time) / (right_time - left_time)
+    return float(values[left_index]) + fraction * (float(values[right_index]) - float(values[left_index]))
+
+
+def _select_data_paths(
+    python_series_by_data_path: dict[str, tuple[float, ...] | list[float]],
+    amesim_results: AmesimResults,
+    data_paths: tuple[str, ...] | list[str] | None,
+) -> tuple[str, ...]:
+    if data_paths is None:
+        data_paths = tuple(
+            data_path
+            for data_path in python_series_by_data_path
+            if data_path in amesim_results.series_by_data_path
+        )
+    selected_paths = tuple(data_paths)
+    if not selected_paths:
+        raise TestMqlComparisonError("no common Data_Path values are available for comparison.")
+    missing_python = [data_path for data_path in selected_paths if data_path not in python_series_by_data_path]
+    if missing_python:
+        raise TestMqlComparisonError(f"Python series missing Data_Path values: {missing_python}")
+    _validate_amesim_data_paths(amesim_results, selected_paths)
+    return selected_paths
+
+
+def _validate_amesim_data_paths(
+    amesim_results: AmesimResults,
+    data_paths: tuple[str, ...] | list[str],
+) -> None:
+    missing_amesim = [data_path for data_path in data_paths if data_path not in amesim_results.series_by_data_path]
+    if missing_amesim:
+        raise TestMqlComparisonError(f"AMESim results missing Data_Path values: {missing_amesim}")
+
+
+def _validate_time_axis(time_values: tuple[float, ...] | list[float]) -> None:
+    if not time_values:
+        raise TestMqlComparisonError("Python time axis is empty.")
+    previous = float(time_values[0])
+    for value in time_values[1:]:
+        value = float(value)
+        if value < previous:
+            raise TestMqlComparisonError("Python time axis must be monotonically increasing.")
+        previous = value

tests/test_test_mql_comparison.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from PythonModels.reporting.amesim_results import load_test_mql_amesim_results
+from PythonModels.reporting.test_mql_comparison import (
+    DEFAULT_TEST_MQL_ALIGNMENT_PATHS,
+    TestMqlComparisonError,
+    compare_test_mql_series,
+    interpolate_series_value,
+    write_test_mql_amesim_baseline_csv,
+    write_test_mql_comparison_csv,
+)
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame"
+
+
+class TestMqlComparisonTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.amesim_results = load_test_mql_amesim_results(TEST_MQL_AME)
+
+    def test_exact_amesim_series_compare_with_zero_error(self) -> None:
+        data_paths = ("temp3@pn_c1_8", "press3@pn_c1_8")
+        python_series = {
+            data_path: self.amesim_results.series(data_path)
+            for data_path in data_paths
+        }
+
+        comparison = compare_test_mql_series(
+            python_times=self.amesim_results.times,
+            python_series_by_data_path=python_series,
+            amesim_results=self.amesim_results,
+            data_paths=data_paths,
+        )
+
+        self.assertEqual(len(comparison.metrics), 2)
+        self.assertEqual(comparison.max_abs_error, 0.0)
+        self.assertEqual(comparison.metric("temp3@pn_c1_8").sample_count, 1002)
+
+    def test_coarse_python_times_are_interpolated_against_amesim(self) -> None:
+        data_path = "temp3@pn_c1_8"
+        python_times = self.amesim_results.times[::100]
+        python_series = {
+            data_path: tuple(
+                interpolate_series_value(
+                    self.amesim_results.times,
+                    self.amesim_results.series(data_path),
+                    time_value,
+                )
+                for time_value in python_times
+            )
+        }
+
+        comparison = compare_test_mql_series(
+            python_times=python_times,
+            python_series_by_data_path=python_series,
+            amesim_results=self.amesim_results,
+            data_paths=(data_path,),
+        )
+
+        self.assertEqual(comparison.metric(data_path).sample_count, len(python_times))
+        self.assertEqual(comparison.max_abs_error, 0.0)
+
+    def test_offset_series_reports_abs_and_relative_error(self) -> None:
+        data_path = "temp3@pn_c1_8"
+        python_series = {
+            data_path: tuple(value + 1.0 for value in self.amesim_results.series(data_path))
+        }
+
+        comparison = compare_test_mql_series(
+            python_times=self.amesim_results.times,
+            python_series_by_data_path=python_series,
+            amesim_results=self.amesim_results,
+            data_paths=(data_path,),
+        )
+        metric = comparison.metric(data_path)
+
+        self.assertAlmostEqual(metric.max_abs_error, 1.0)
+        self.assertAlmostEqual(metric.mean_abs_error, 1.0)
+        self.assertGreater(metric.max_rel_error, 0.0)
+
+    def test_missing_data_path_is_rejected(self) -> None:
+        with self.assertRaises(TestMqlComparisonError):
+            compare_test_mql_series(
+                python_times=self.amesim_results.times,
+                python_series_by_data_path={"missing@component": (1.0,) * 1002},
+                amesim_results=self.amesim_results,
+            )
+
+    def test_writes_baseline_and_comparison_csv_files(self) -> None:
+        data_paths = DEFAULT_TEST_MQL_ALIGNMENT_PATHS[:2]
+        python_series = {
+            data_path: self.amesim_results.series(data_path)
+            for data_path in data_paths
+        }
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_dir = Path(temp_dir)
+            baseline_path = write_test_mql_amesim_baseline_csv(
+                output_dir,
+                self.amesim_results,
+                data_paths=data_paths,
+            )
+            comparison_path, summary_path, comparison = write_test_mql_comparison_csv(
+                output_dir=output_dir,
+                python_times=self.amesim_results.times,
+                python_series_by_data_path=python_series,
+                amesim_results=self.amesim_results,
+                data_paths=data_paths,
+            )
+
+            self.assertTrue(baseline_path.exists())
+            self.assertTrue(comparison_path.exists())
+            self.assertTrue(summary_path.exists())
+            self.assertIn("time_s,temp3@pn_c1_8,press3@pn_c1_8", baseline_path.read_text(encoding="utf-8"))
+            self.assertEqual(comparison.max_abs_error, 0.0)
+
+
+if __name__ == "__main__":
+    unittest.main()