diff --git a/AmesimModels/test_mql/README.md b/AmesimModels/test_mql/README.md index 81723c8..54c7d8c 100644 --- a/AmesimModels/test_mql/README.md +++ b/AmesimModels/test_mql/README.md @@ -12,7 +12,8 @@ ## 源文件 - AMESim 源模型：`AmesimModels/test_mql.ame` -- 该 `.ame` 文件本质是一个 tar 包，内部包含 `test_mql_.cir`、`test_mql_.param`、`test_mql_.modelinfo`、`test_mql_.sim` 等文件。 +- 该 `.ame` 文件本质是一个 tar 包，内部包含 `test_mql_.cir`、`test_mql_.param`、`test_mql_.modelinfo`、`test_mql_.sim`、`test_mql_.data`、`test_mql_.results`、`test_mql_.var` 等文件。 +- `test_mql_.results` 是 AMESim 主时序结果文件；`test_mql_.data` 是配套结果/数据文件。 ## 当前新增文件 @@ -45,7 +46,14 @@ - `PythonModels/components/amesim_pneumatic.py` - 提供 `test_mql` 后续会用到的 AMESim 气动组件原语。 - 当前包含氦气 PR 压力闭合的气动容腔、标准可压缩孔口流量、单位换算 helper。 - - 这层仍是首版近似原语，后续必须通过 AMESim CSV 对齐修正。 + - 这层仍是首版近似原语，后续必须通过 AMESim 结果对齐修正。 + +- `PythonModels/reporting/amesim_results.py` + - 直接解析 `.ame` 包内的 AMESim 二进制结果。 + - 当前已确认 `test_mql_.results` 结构为：2 个 int32 头字段、1116 个保存变量映射、1002 行 `time + 1116变量` 的 double 时序矩阵，以及尾部 `time + 1278变量终值`。 + +- `tests/test_amesim_results.py` + - 保护 `test_mql_.results` 的时间轴、变量数量、Data_Path 映射和典型样本值。 ## 物性约定 @@ -86,19 +94,19 @@ AMESim 模型中 `test_mql` 使用氦气，Python 侧当前通过 `HELIUM_PR` 1. 组件别名优先沿用 AMESim 原名，例如 `pn_brp2_8`、`pn_c1_8`、`pn_orifice_18`。 2. 子模型族优先沿用 AMESim 子模型名，例如 `PNRP17`、`PNCH012`、`PNOR001`、`PNVO001`。 3. 输出变量命名要能追溯到 AMESim 的 `Data_Path`，避免 Python 侧改名后无法对比。 -4. Python 每完成一批物理方程，都应拿 AMESim 导出的 CSV 结果做误差对比。 -5. 当前 `.ame` 包内的 `test_mql_.results` 是 AMESim 二进制结果文件，暂不作为直接解析基准；建议后续从 AMESim 导出 CSV 结果放到本目录下。 -6. 在没有 AMESim CSV 基准前，不能把 Python 输出描述为“已经和 AMESim 几乎一致”，只能说明完成了结构、参数或某类组件方程。 +4. Python 每完成一批物理方程，都应优先使用 `test_mql_.results` 解析出的 AMESim 原始时序做误差对比。 +5. CSV 仍可作为人工检查或外部交换格式，但不是自动对齐的前置条件。 +6. 在没有完成逐变量误差对比前，不能把 Python 输出描述为“已经和 AMESim 几乎一致”，只能说明完成了结构、参数、结果读取或某类组件方程。 ## 验证方式 ```bash -python3 -m py_compile PythonModels/components/amesim_pneumatic.py PythonModels/core/peng_robinson.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/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 物理结果复刻。 ## 后续方向 diff --git a/PythonModels/reporting/amesim_results.py b/PythonModels/reporting/amesim_results.py new file mode 100644 index 0000000..9e26249 --- /dev/null +++ b/PythonModels/reporting/amesim_results.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import re +import struct +import tarfile +from dataclasses import dataclass +from pathlib import Path + + +class AmesimResultsError(ValueError): + """Raised when AMESim result files cannot be parsed consistently.""" + + +@dataclass(frozen=True) +class AmesimVariable: + index: int + label: str + data_path: str | None + param_id: int | None + hidden: bool + + +@dataclass(frozen=True) +class AmesimResults: + times: tuple[float, ...] + variables: tuple[AmesimVariable, ...] + saved_variable_indices: tuple[int, ...] + series_by_data_path: dict[str, tuple[float, ...]] + final_values_by_data_path: dict[str, float] + + @property + def point_count(self) -> int: + return len(self.times) + + @property + def saved_variable_count(self) -> int: + return len(self.saved_variable_indices) + + def series(self, data_path: str) -> tuple[float, ...]: + return self.series_by_data_path[data_path] + + def final_value(self, data_path: str) -> float: + return self.final_values_by_data_path[data_path] + + +_DATA_PATH_RE = re.compile(r"Data_Path=(\S+)") +_PARAM_ID_RE = re.compile(r"Param_Id=(\d+)") + + +def load_test_mql_amesim_results(archive_path: str | Path) -> AmesimResults: + return load_amesim_results_from_archive( + archive_path=archive_path, + var_member="test_mql_.var", + results_member="test_mql_.results", + ) + + +def load_amesim_results_from_archive( + *, + archive_path: str | Path, + var_member: str, + results_member: str, +) -> AmesimResults: + with tarfile.open(archive_path) as archive: + var_file = archive.extractfile(var_member) + results_file = archive.extractfile(results_member) + if var_file is None: + raise AmesimResultsError(f"Missing AMESim variable member: {var_member}") + if results_file is None: + raise AmesimResultsError(f"Missing AMESim results member: {results_member}") + var_lines = var_file.read().decode("latin1").splitlines() + results_data = results_file.read() + variables = tuple(_parse_variable_line(index, line) for index, line in enumerate(var_lines)) + return parse_amesim_results_bytes(results_data, variables) + + +def parse_amesim_results_bytes( + results_data: bytes, + variables: tuple[AmesimVariable, ...], +) -> AmesimResults: + if len(results_data) < 8: + raise AmesimResultsError("AMESim results data is too small.") + point_count, encoded_saved_variable_count = struct.unpack_from("<2i", results_data, 0) + saved_variable_count = abs(encoded_saved_variable_count) + if point_count <= 0 or saved_variable_count <= 0: + raise AmesimResultsError("Invalid AMESim results header.") + + mapping_offset = 8 + mapping_size = saved_variable_count * 4 + data_offset = mapping_offset + mapping_size + saved_variable_indices = struct.unpack_from( + f"<{saved_variable_count}i", + results_data, + mapping_offset, + ) + if any(index < 0 or index >= len(variables) for index in saved_variable_indices): + raise AmesimResultsError( + "AMESim results variable mapping references unknown .var rows." + ) + + row_length = 1 + saved_variable_count + main_value_count = point_count * row_length + main_byte_count = main_value_count * 8 + main_end = data_offset + main_byte_count + if main_end > len(results_data): + raise AmesimResultsError("AMESim results matrix is truncated.") + + main_values = struct.unpack_from(f"<{main_value_count}d", results_data, data_offset) + times = tuple(main_values[row * row_length] for row in range(point_count)) + series_by_data_path: dict[str, tuple[float, ...]] = {} + for column, variable_index in enumerate(saved_variable_indices, start=1): + variable = variables[variable_index] + if variable.data_path is None: + continue + series_by_data_path[variable.data_path] = tuple( + main_values[row * row_length + column] + for row in range(point_count) + ) + + final_values_by_data_path: dict[str, float] = {} + trailing_bytes = len(results_data) - main_end + expected_final_bytes = (1 + len(variables)) * 8 + if trailing_bytes >= expected_final_bytes: + final_values = struct.unpack_from(f"<{1 + len(variables)}d", results_data, main_end) + for variable, value in zip(variables, final_values[1:]): + if variable.data_path is not None: + final_values_by_data_path[variable.data_path] = value + + return AmesimResults( + times=times, + variables=variables, + saved_variable_indices=tuple(saved_variable_indices), + series_by_data_path=series_by_data_path, + final_values_by_data_path=final_values_by_data_path, + ) + + +def _parse_variable_line(index: int, line: str) -> AmesimVariable: + data_path_match = _DATA_PATH_RE.search(line) + param_id_match = _PARAM_ID_RE.search(line) + label = line + if data_path_match is not None: + label = line[: data_path_match.start()].strip() + return AmesimVariable( + index=index, + label=label, + data_path=data_path_match.group(1) if data_path_match else None, + param_id=int(param_id_match.group(1)) if param_id_match else None, + hidden="HIDDEN" in line, + ) diff --git a/tests/test_amesim_results.py b/tests/test_amesim_results.py new file mode 100644 index 0000000..d3a746f --- /dev/null +++ b/tests/test_amesim_results.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import unittest +from pathlib import Path + +from PythonModels.reporting.amesim_results import load_test_mql_amesim_results + + +REPO_ROOT = Path(__file__).resolve().parents[1] +TEST_MQL_AME = REPO_ROOT / "AmesimModels" / "test_mql.ame" + + +class AmesimResultsTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.results = load_test_mql_amesim_results(TEST_MQL_AME) + + def test_reads_binary_results_shape(self) -> None: + self.assertEqual(self.results.point_count, 1002) + self.assertEqual(self.results.saved_variable_count, 1116) + self.assertEqual(len(self.results.variables), 1278) + self.assertEqual(len(self.results.final_values_by_data_path), 1100) + + def test_time_axis_matches_amesim_simulation(self) -> None: + self.assertAlmostEqual(self.results.times[0], 0.0) + self.assertAlmostEqual(self.results.times[1], 0.01) + self.assertAlmostEqual(self.results.times[-1], 10.0) + + def test_variable_mapping_preserves_data_paths(self) -> None: + self.assertEqual(self.results.saved_variable_indices[:2], (0, 1)) + self.assertEqual(self.results.variables[0].data_path, "temp3@pn_c1_8") + self.assertEqual(self.results.variables[1].data_path, "press3@pn_c1_8") + self.assertIn("vvol1@pn_brp2_8", self.results.series_by_data_path) + self.assertIn("vol1@pn_brp2_8", self.results.series_by_data_path) + + def test_series_values_match_known_amesim_samples(self) -> None: + temp3 = self.results.series("temp3@pn_c1_8") + press3 = self.results.series("press3@pn_c1_8") + self.assertAlmostEqual(temp3[0], 293.15) + self.assertAlmostEqual(press3[0], -1300.0) + self.assertAlmostEqual(temp3[-1], self.results.final_value("temp3@pn_c1_8")) + self.assertAlmostEqual(press3[-1], self.results.final_value("press3@pn_c1_8")) + + +if __name__ == "__main__": + unittest.main()