diff --git a/app/native/wndb/inp/exporter.py b/app/native/wndb/inp/exporter.py index 34b708b..fe5937b 100644 --- a/app/native/wndb/inp/exporter.py +++ b/app/native/wndb/inp/exporter.py @@ -120,7 +120,7 @@ def dump_inp(project: str, inp: str, version: str = '3'): file.write('\n'.join(inp_out_demand(project))) elif name == STATUS: - file.write('\n'.join(inp_out_status(project))) + file.write('\n'.join(inp_out_status(project, version=version))) elif name == PATTERNS: if version == '3': @@ -232,7 +232,7 @@ def export_inp(project: str, version: str = '3') -> ChangeSet: inp += '\n'.join(inp_out_demand(project)) elif name == STATUS: - inp += '\n'.join(inp_out_status(project)) + inp += '\n'.join(inp_out_status(project, version=version)) elif name == PATTERNS: if version == '3': diff --git a/app/native/wndb/model/status.py b/app/native/wndb/model/status.py index b056c11..7327a86 100644 --- a/app/native/wndb/model/status.py +++ b/app/native/wndb/model/status.py @@ -93,14 +93,16 @@ def inp_in_status(line: str) -> str: return str(f"insert into network.link_initial_settings (link_id, status, setting) values ({sql_literal(link)}, null, {sql_literal(float(value))});") -def inp_out_status(name: str) -> list[str]: +def inp_out_status(name: str, version: str = '2') -> list[str]: lines = [] objs = read_all(name, 'select link_id as link, status, setting from network.link_initial_settings order by link_id') for obj in objs: link = obj['link'] status = obj['status'] if obj['status'] != None else '' setting = obj['setting'] if obj['setting'] != None else '' - if status != '': + # EPANET 2 valves use their configured setting by default; ACTIVE is + # an internal state, not a valid keyword in its [STATUS] section. + if status != '' and not (version == '2' and status.upper() == LINK_STATUS_ACTIVE): lines.append(f'{link} {status}') if setting != '': lines.append(f'{link} {setting}') diff --git a/app/services/simulation.py b/app/services/simulation.py index a8bb1ff..6ae122b 100644 --- a/app/services/simulation.py +++ b/app/services/simulation.py @@ -490,6 +490,23 @@ def run_simulation( set_status(name_c, cs) # 运行并返回结果 result_data = json.loads(run_project(name_c)) + if result_data.get("simulation_result") == "failed": + report = result_data.get("report") or "" + logging.error("EPANET failed for project %s:\n%s", name_c, report) + lines = report.splitlines() + error_index = next( + (i for i, line in enumerate(lines) if line.strip().startswith("Error ")), + None, + ) + detail = ( + " ".join(line.strip() for line in lines[error_index:error_index + 2])[:600] + if error_index is not None + else "引擎未返回错误详情，请查看服务日志" + ) + raise RuntimeError(f"EPANET 计算失败：{detail}") + output_data = result_data.get("output") + if not isinstance(output_data, dict): + raise RuntimeError("run_project did not return JSON output content") refresh_materialized_views_after_commit(name_c) time_cost_end = time.perf_counter() print( @@ -498,9 +515,6 @@ def run_simulation( time_cost_end - time_cost_start, ) ) - output_data = result_data.get("output") - if not isinstance(output_data, dict): - raise RuntimeError("run_project did not return JSON output content") node_result = output_data.get("node_results") link_result = output_data.get("link_results") if node_result is None or link_result is None: diff --git a/tests/unit/test_analysis_simulation.py b/tests/unit/test_analysis_simulation.py index c15cfaa..c387e30 100644 --- a/tests/unit/test_analysis_simulation.py +++ b/tests/unit/test_analysis_simulation.py @@ -277,3 +277,51 @@ def test_extended_simulation_marks_run_failed_when_result_storage_fails(monkeypa ) assert [call[1]["status"] for call in lifecycle_calls] == ["failed"] + + +@pytest.mark.parametrize("has_output", [False, True]) +def test_engine_failure_preserves_report_and_rolls_back_model_changes(monkeypatch, has_output): + from app.services import simulation + + transaction_outcome = [] + + @contextmanager + def project_transaction(_name): + try: + yield + except Exception: + transaction_outcome.append("rollback") + raise + else: + transaction_outcome.append("commit") + + monkeypatch.setattr(simulation, "project_transaction", project_transaction) + monkeypatch.setattr(simulation, "get_time", lambda _: { + "HYDRAULIC TIMESTEP": "0:15:00", + "DURATION": "0:00", + "PATTERN START": "0:00", + }) + monkeypatch.setattr(simulation, "set_time", lambda *_: None) + result = { + "simulation_result": "failed", + "report": "EPANET report\n Error 202: illegal numeric value ACTIVE in [STATUS] section:\n V1 ACTIVE\n", + } + if has_output: + result["output"] = {"node_results": [], "link_results": [], "times": {"num_periods": 1}} + monkeypatch.setattr(simulation, "run_project", lambda _: json.dumps(result)) + refresh = Mock() + store = Mock() + monkeypatch.setattr(simulation, "refresh_materialized_views_after_commit", refresh) + monkeypatch.setattr(simulation.TimescaleInternalStorage, "store_realtime_simulation", store) + + with pytest.raises(RuntimeError, match="EPANET.*Error 202"): + simulation.run_simulation( + name="demo", + simulation_type="realtime", + modify_pattern_start_time="2026-07-30T04:30:00Z", + scada_mappings=_empty_scada_mappings(simulation), + ) + + assert transaction_outcome == ["rollback"] + refresh.assert_not_called() + store.assert_not_called() diff --git a/tests/unit/test_epanet_status_export.py b/tests/unit/test_epanet_status_export.py new file mode 100644 index 0000000..4f25dd3 --- /dev/null +++ b/tests/unit/test_epanet_status_export.py @@ -0,0 +1,70 @@ +import ctypes +import platform +from pathlib import Path + +import pytest + +from app.native.wndb.model import status + + +def test_epanet2_preserves_valve_setting_without_emitting_active(monkeypatch): + monkeypatch.setattr(status, "read_all", lambda *_: [ + {"link": "V1", "status": "ACTIVE", "setting": None}, + {"link": "V2", "status": "ACTIVE", "setting": 2.5}, + {"link": "V3", "status": "CLOSED", "setting": None}, + {"link": "P1", "status": "OPEN", "setting": 0}, + ]) + assert status.inp_out_status("model") == ["V2 2.5", "V3 CLOSED", "P1 OPEN", "P1 0"] + + +def test_epanet3_retains_explicit_active_status(monkeypatch): + monkeypatch.setattr(status, "read_all", lambda *_: [ + {"link": "V1", "status": "ACTIVE", "setting": 100}, + ]) + assert status.inp_out_status("model", version="3") == ["V1 ACTIVE", "V1 100"] + + +@pytest.mark.parametrize("version", ["2", "3"]) +def test_file_and_text_exporters_forward_the_selected_format(monkeypatch, tmp_path, version): + from app.native.wndb.inp import exporter + + monkeypatch.setattr(exporter, "have_project", lambda _: True) + monkeypatch.setattr(exporter, "section_name", [exporter.STATUS]) + monkeypatch.setattr(exporter, "section_names_for_epanetv2", [exporter.STATUS]) + monkeypatch.setattr(status, "read_all", lambda *_: [ + {"link": "V1", "status": "ACTIVE", "setting": 100}, + ]) + path = tmp_path / "model.inp" + exporter.dump_inp("model", str(path), version) + text = exporter.export_inp("model", version).operations[0]["inp"] + assert "V1 100" in path.read_text() and "V1 100" in text + assert ("V1 ACTIVE" in path.read_text()) == (version == "3") + assert ("V1 ACTIVE" in text) == (version == "3") + + +def test_exported_active_valve_passes_bundled_epanet2_input_validation(monkeypatch, tmp_path): + library = Path(__file__).resolve().parents[2] / "app/infra/epanet/linux/libepanet2.so" + if platform.system() != "Linux" or not library.exists(): + pytest.skip("Bundled Linux EPANET library is unavailable") + monkeypatch.setattr(status, "read_all", lambda *_: [ + {"link": "V1", "status": "ACTIVE", "setting": None}, + ]) + inp = tmp_path / "valve.inp" + inp.write_text( + "[JUNCTIONS]\nJ1 0 1\nJ2 0 1\n[RESERVOIRS]\nR1 100\n" + "[PIPES]\nP1 R1 J1 100 300 100 0 OPEN\n" + "[VALVES]\nV1 J1 J2 600 TCV 100 0\n[STATUS]\n" + + "\n".join(status.inp_out_status("model")) + + "\n[OPTIONS]\nUNITS LPS\n[END]\n" + ) + lib = ctypes.CDLL(str(library)) + lib.EN_createproject.argtypes = [ctypes.POINTER(ctypes.c_void_p)] + lib.EN_open.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p] + lib.EN_deleteproject.argtypes = [ctypes.c_void_p] + handle = ctypes.c_void_p() + assert lib.EN_createproject(ctypes.byref(handle)) == 0 + try: + code = lib.EN_open(handle, str(inp).encode(), str(tmp_path / "valve.rpt").encode(), b"") + assert code == 0 + finally: + lib.EN_deleteproject(handle)