From 1f47171d078d0f3d7538cb4862a9ebf909f38dc3 Mon Sep 17 00:00:00 2001 From: huojiarong Date: Wed, 15 Jul 2026 09:40:07 +0000 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9Etest=5Fmql=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E5=AF=B9=E9=BD=90=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AmesimModels/test_mql/README.md | 14 +- PythonModels/reporting/test_mql_comparison.py | 237 ++++++++++++++++++ tests/test_test_mql_comparison.py | 125 +++++++++ 3 files changed, 373 insertions(+), 3 deletions(-) create mode 100644 PythonModels/reporting/test_mql_comparison.py create mode 100644 tests/test_test_mql_comparison.py diff --git a/AmesimModels/test_mql/README.md b/AmesimModels/test_mql/README.md index 54c7d8c..5e8bfc0 100644 --- a/AmesimModels/test_mql/README.md +++ b/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`：气动活塞、机械负载和端止动。 diff --git a/PythonModels/reporting/test_mql_comparison.py b/PythonModels/reporting/test_mql_comparison.py new file mode 100644 index 0000000..3d65cf2 --- /dev/null +++ b/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 diff --git a/tests/test_test_mql_comparison.py b/tests/test_test_mql_comparison.py new file mode 100644 index 0000000..65fcf54 --- /dev/null +++ b/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()