From 67ba9c2ac6e0eb0d5f01da5518101957ce9ef178 Mon Sep 17 00:00:00 2001 From: Huarch Date: Thu, 11 Jun 2026 10:29:04 +0800 Subject: [PATCH] fix(simulation): align epanet output handling --- app/algorithms/simulation/scenarios.py | 15 +- app/api/v1/endpoints/simulation.py | 231 ++++++------------------- app/infra/epanet/epanet.py | 118 +++++++++---- app/services/simulation.py | 45 +++-- app/services/simulation_ops.py | 39 +++-- 5 files changed, 185 insertions(+), 263 deletions(-) diff --git a/app/algorithms/simulation/scenarios.py b/app/algorithms/simulation/scenarios.py index 9db4ea7..9890fc8 100644 --- a/app/algorithms/simulation/scenarios.py +++ b/app/algorithms/simulation/scenarios.py @@ -9,7 +9,6 @@ from app.algorithms.simulation.runner import ( run_simulation_ex, from_clock_to_seconds_2, ) -from app.infra.epanet.epanet import Output from app.services.scheme_management import store_scheme_info from app.services.tjnetwork import ( ChangeSet, @@ -663,24 +662,24 @@ def age_analysis( new_name, "realtime", modify_pattern_start_time, - modify_total_duration, + duration=modify_total_duration, downloading_prohibition=True, ) + simulation_result = json.loads(result) + output_data = simulation_result.get("output") + if not isinstance(output_data, dict): + raise RuntimeError("run_simulation_ex did not return JSON output content") # step 2. restore the base model status # execute_undo(name) #有疑惑 if is_project_open(new_name): close_project(new_name) delete_project(new_name) - output = Output("./temp/{}.db.out".format(new_name)) - # element_name = output.element_name() - # node_name = element_name['nodes'] - # link_name = element_name['links'] nodes_age = [] - node_result = output.node_results() + node_result = output_data.get("node_results") or [] for node in node_result: nodes_age.append(node["result"][-1]["quality"]) links_age = [] - link_result = output.link_results() + link_result = output_data.get("link_results") or [] for link in link_result: links_age.append(link["result"][-1]["quality"]) age_result = {"nodes": nodes_age, "links": links_age} diff --git a/app/api/v1/endpoints/simulation.py b/app/api/v1/endpoints/simulation.py index 20e6d26..27be343 100644 --- a/app/api/v1/endpoints/simulation.py +++ b/app/api/v1/endpoints/simulation.py @@ -8,7 +8,6 @@ from fastapi import APIRouter, HTTPException, File, UploadFile, Query, Path, Bod from fastapi.responses import PlainTextResponse import app.services.simulation as simulation import app.services.globals as globals -from app.infra.cache.redis_client import redis_client from app.services.tjnetwork import ( run_project, run_project_return_dict, @@ -28,6 +27,7 @@ from app.algorithms.sensor import ( pressure_sensor_placement_sensitivity, pressure_sensor_placement_kmeans, ) + from app.services.network_import import network_update from app.services.simulation_ops import ( project_management, @@ -35,16 +35,22 @@ from app.services.simulation_ops import ( daily_scheduling_simulation, ) from app.services.valve_isolation import analyze_valve_isolation -from pydantic import BaseModel, Field +from app.services.time_api import parse_aware_time, parse_utc_time +from pydantic import BaseModel, Field, field_validator router = APIRouter() class RunSimulationManuallyByDate(BaseModel): name: str = Field(..., description="管网名称（或数据库名称）") - simulation_date: str = Field(..., description="模拟基准日期 (YYYY-MM-DD)") - start_time: str = Field(..., description="开始时间 (HH:MM 或 HH:MM:SS)") - duration: int = Field(..., description="持续时间 (分钟)") + start_time: str = Field(..., description="开始时间 (ISO 8601 / RFC3339，必须显式带时区)") + duration: int = Field(..., gt=0, description="持续时间 (分钟)") + + @field_validator("start_time") + @classmethod + def validate_start_time_timezone(cls, value: str) -> str: + parse_aware_time(value, field_name="start_time") + return value class BurstAnalysis(BaseModel): @@ -109,63 +115,37 @@ class PressureSensorPlacement(BaseModel): def run_simulation_manually_by_date( - network_name: str, base_date: datetime, start_time: str, duration: int + network_name: str, start_time: datetime, duration: int ) -> None: - time_parts = list(map(int, start_time.split(":"))) - if len(time_parts) == 2: - start_hour, start_minute = time_parts - start_second = 0 - elif len(time_parts) == 3: - start_hour, start_minute, start_second = time_parts - else: - raise ValueError("Invalid start_time format. Use HH:MM or HH:MM:SS") - - start_datetime = base_date.replace( - hour=start_hour, minute=start_minute, second=start_second - ) - end_datetime = start_datetime + timedelta(minutes=duration) - current_time = start_datetime + end_datetime = start_time + timedelta(minutes=duration) + current_time = start_time while current_time < end_datetime: - iso_time = current_time.strftime("%Y-%m-%dT%H:%M:%S") + "+08:00" simulation.run_simulation( name=network_name, simulation_type="realtime", - modify_pattern_start_time=iso_time, + modify_pattern_start_time=current_time.isoformat(timespec="seconds"), ) current_time += timedelta(minutes=15) # 必须用这个PlainTextResponse，不然每个key都有引号 -@router.get("/runproject/", response_class=PlainTextResponse, summary="运行项目模拟", description="基于指定的管网项目运行标准水力模拟，返回纯文本格式的模拟报告。使用分布式锁机制确保同时只有一个模拟任务运行。") +@router.get("/runproject/", response_class=PlainTextResponse, summary="运行项目模拟", description="基于指定的管网项目运行标准水力模拟，返回纯文本格式的模拟报告。") async def run_project_endpoint(network: str = Query(..., description="管网名称（或数据库名称）")) -> str: """ 运行项目模拟 - **network**: 管网名称（或数据库名称） - 使用分布式锁机制确保同一时间只有一个模拟任务在运行。如果已有任务在运行，返回409状态码。 + 运行指定管网项目的标准水力模拟并返回文本报告。 """ - lock_key = "exclusive_api_lock" - timeout = 120 # 锁自动过期时间（秒） - - # 尝试获取锁（NX=True: 不存在时设置，EX=timeout: 过期时间） - acquired = redis_client.set(lock_key, "locked", nx=True, ex=timeout) - - if not acquired: - raise HTTPException(status_code=409, detail="is in simulation") - else: - try: - return run_project(network) - finally: - # 手动释放锁（可选，依赖过期时间自动释放更安全） - redis_client.delete(lock_key) + return run_project(network) # DingZQ, 2025-02-04, 返回dict[str, Any] # output 和 report # output 是 json # report 是 text -@router.get("/runprojectreturndict/", summary="运行项目模拟（返回字典）", description="基于指定的管网项目运行标准水力模拟，返回JSON格式的字典，包含输出数据和报告文本。使用分布式锁机制确保同时只有一个模拟任务运行。") +@router.get("/runprojectreturndict/", summary="运行项目模拟（返回字典）", description="基于指定的管网项目运行标准水力模拟，返回JSON格式的字典，包含输出数据和报告文本。") async def run_project_return_dict_endpoint(network: str = Query(..., description="管网名称（或数据库名称）")) -> dict[str, Any]: """ 运行项目模拟（返回字典） @@ -176,22 +156,9 @@ async def run_project_return_dict_endpoint(network: str = Query(..., description - output: JSON格式的模拟输出数据 - report: 文本格式的模拟报告 - 使用分布式锁机制确保同一时间只有一个模拟任务在运行。 + 运行指定管网项目的标准水力模拟并返回字典结果。 """ - lock_key = "exclusive_api_lock" - timeout = 120 # 锁自动过期时间（秒） - - # 尝试获取锁（NX=True: 不存在时设置，EX=timeout: 过期时间） - acquired = redis_client.set(lock_key, "locked", nx=True, ex=timeout) - - if not acquired: - raise HTTPException(status_code=409, detail="is in simulation") - else: - try: - return run_project_return_dict(network) - finally: - # 手动释放锁（可选，依赖过期时间自动释放更安全） - redis_client.delete(lock_key) + return run_project_return_dict(network) # put in inp folder, name without extension @@ -221,26 +188,6 @@ async def dump_output_endpoint(output: str = Query(..., description="模拟输 # Analysis Endpoints -@router.get("/burstanalysis/", summary="爆管分析（基础）", description="对管网中的爆管事件进行分析，包括爆管对管网压力和流量的影响。此为基础版本，接收简化的查询参数。") -async def burst_analysis_endpoint( - network: str = Query(..., description="管网名称（或数据库名称）"), - pipe_id: str = Query(..., description="管段ID"), - start_time: str = Query(..., description="分析开始时间（ISO 8601格式）"), - end_time: str = Query(..., description="分析结束时间（ISO 8601格式）"), - burst_flow: float = Query(..., description="爆管流量大小（L/s）"), -): - """ - 爆管分析（基础版本） - - - **network**: 管网名称（或数据库名称） - - **pipe_id**: 管段ID - - **start_time**: 分析开始时间 - - **end_time**: 分析结束时间 - - **burst_flow**: 爆管流量大小 - """ - return burst_analysis(network, pipe_id, start_time, end_time, burst_flow) - - @router.get("/burst_analysis/", summary="爆管分析（高级）", description="高级版本的爆管分析，支持在指定时间点修改泵控制模式和阀门开度，以分析这些改变对爆管影响的作用。支持固定泵和变速泵的独立控制。") async def fastapi_burst_analysis( network: str = Query(..., description="管网名称（或数据库名称）"), @@ -273,30 +220,13 @@ async def fastapi_burst_analysis( return "success" -@router.get("/valvecloseanalysis/", summary="阀门关闭分析（基础）", description="对管网中的阀门关闭事件进行分析，评估关闭阀门对管网的影响。此为基础版本。") -async def valve_close_analysis_endpoint( - network: str = Query(..., description="管网名称（或数据库名称）"), - valve_id: str = Query(..., description="阀门ID"), - start_time: str = Query(..., description="分析开始时间（ISO 8601格式）"), - end_time: str = Query(..., description="分析结束时间（ISO 8601格式）"), -): - """ - 阀门关闭分析（基础版本） - - - **network**: 管网名称（或数据库名称） - - **valve_id**: 阀门ID - - **start_time**: 分析开始时间 - - **end_time**: 分析结束时间 - """ - return valve_close_analysis(network, valve_id, start_time, end_time) - - @router.get("/valve_close_analysis/", response_class=PlainTextResponse, summary="阀门关闭分析（高级）", description="高级版本的阀门关闭分析，支持同时关闭多个阀门，并在指定持续时间内进行模拟。返回纯文本格式的分析结果。") async def fastapi_valve_close_analysis( network: str = Query(..., description="管网名称（或数据库名称）"), start_time: str = Query(..., description="阀门关闭开始时间（ISO 8601格式）"), valves: List[str] = Query(..., description="要关闭的阀门ID列表"), duration: int | None = Query(None, description="模拟持续时间（秒），默认900秒"), + scheme_name: str = Query(..., description="阀门关闭方案名称"), ) -> str: """ 阀门关闭分析（高级版本） @@ -305,6 +235,7 @@ async def fastapi_valve_close_analysis( - **start_time**: 阀门关闭开始时间 - **valves**: 要关闭的阀门ID列表 - **duration**: 模拟持续时间（秒，可选，默认900） + - **scheme_name**: 阀门关闭方案名称 支持同时关闭多个阀门进行分析。 """ @@ -313,6 +244,7 @@ async def fastapi_valve_close_analysis( modify_pattern_start_time=start_time, modify_total_duration=duration or 900, modify_valve_opening={valve_id: 0.0 for valve_id in valves}, + scheme_name=scheme_name, ) return result or "success" @@ -336,47 +268,27 @@ async def valve_isolation_endpoint( - affected_nodes: 受影响的节点列表 - isolatable: 是否可以有效隔离 """ - result = { - "accident_element": "P461309", - "accident_elements": ["P461309"], - "affected_nodes": [ - "J316629_A", - "J317037_B", - "J317060_B", - "J408189_B", - "J499996", - "J524940", - "J535933", - "J58841", - ], - "isolatable": True, - "must_close_valves": ["210521658", "V12974", "V12986", "V12993"], - "optional_valves": [], - } + # result = { + # "accident_element": "P461309", + # "accident_elements": ["P461309"], + # "affected_nodes": [ + # "J316629_A", + # "J317037_B", + # "J317060_B", + # "J408189_B", + # "J499996", + # "J524940", + # "J535933", + # "J58841", + # ], + # "isolatable": True, + # "must_close_valves": ["210521658", "V12974", "V12986", "V12993"], + # "optional_valves": [], + # } result = analyze_valve_isolation(network, accident_element, disabled_valves) return result -@router.get("/flushinganalysis/", summary="冲洗分析（基础）", description="对管网的冲洗操作进行分析，评估冲洗流量和持续时间对管网的影响。此为基础版本。") -async def flushing_analysis_endpoint( - network: str = Query(..., description="管网名称（或数据库名称）"), - pipe_id: str = Query(..., description="要冲洗的管段ID"), - start_time: str = Query(..., description="冲洗开始时间（ISO 8601格式）"), - duration: float = Query(..., description="冲洗持续时间（分钟）"), - flow: float = Query(..., description="冲洗流量（L/s）"), -): - """ - 冲洗分析（基础版本） - - - **network**: 管网名称（或数据库名称） - - **pipe_id**: 要冲洗的管段ID - - **start_time**: 冲洗开始时间 - - **duration**: 冲洗持续时间（分钟） - - **flow**: 冲洗流量（L/s） - """ - return flushing_analysis(network, pipe_id, start_time, duration, flow) - - @router.get("/flushing_analysis/", response_class=PlainTextResponse, summary="冲洗分析（高级）", description="高级版本的冲洗分析，支持同时开启多个阀门进行冲洗，指定排污节点，并设置固定的冲洗流量。返回纯文本格式的分析结果。") async def fastapi_flushing_analysis( network: str = Query(..., description="管网名称（或数据库名称）"), @@ -386,7 +298,7 @@ async def fastapi_flushing_analysis( drainage_node_ID: str = Query(..., description="排污节点ID"), flush_flow: float = Query(0, description="冲洗流量（L/s），0表示自动计算"), duration: int | None = Query(None, description="模拟持续时间（秒），默认900秒"), - scheme_name: str | None = Query(None, description="冲洗方案名称（可选）"), + scheme_name: str = Query(..., description="冲洗方案名称"), ) -> str: """ 冲洗分析（高级版本） @@ -398,7 +310,7 @@ async def fastapi_flushing_analysis( - **drainage_node_ID**: 排污节点ID - **flush_flow**: 冲洗流量（L/s） - **duration**: 模拟持续时间（秒，可选，默认900） - - **scheme_name**: 冲洗方案名称（可选） + - **scheme_name**: 冲洗方案名称 支持多阀联合冲洗操作。 """ @@ -424,7 +336,7 @@ async def fastapi_contaminant_simulation( source: str = Query(..., description="污染源节点ID"), concentration: float = Query(..., description="污染浓度（mg/L）"), duration: int = Query(..., description="模拟持续时间（秒）"), - scheme_name: str | None = Query(None, description="模拟方案名称（可选）"), + scheme_name: str = Query(..., description="模拟方案名称"), pattern: str | None = Query(None, description="污染源模式ID（可选）"), ) -> str: """ @@ -435,7 +347,7 @@ async def fastapi_contaminant_simulation( - **source**: 污染源节点ID - **concentration**: 污染浓度（mg/L） - **duration**: 模拟持续时间（秒） - - **scheme_name**: 模拟方案名称（可选） + - **scheme_name**: 模拟方案名称 - **pattern**: 污染源模式ID（可选） 用于评估管网中污染物的传播和影响范围。 @@ -452,23 +364,10 @@ async def fastapi_contaminant_simulation( return result or "success" -@router.get("/ageanalysis/", summary="水龄分析（基础）", description="对管网中的水体停留时间（水龄）进行分析。此为基础版本。") -async def age_analysis_endpoint(network: str = Query(..., description="管网名称（或数据库名称）")): - """ - 水龄分析（基础版本） - - - **network**: 管网名称（或数据库名称） - - 分析管网中各节点的水体停留时间。 - """ - return age_analysis(network) - - @router.get("/age_analysis/", response_class=PlainTextResponse, summary="水龄分析（高级）", description="高级版本的水龄分析，在指定时间点进行分析，支持自定义模拟持续时间。返回纯文本格式的分析结果。") async def fastapi_age_analysis( network: str = Query(..., description="管网名称（或数据库名称）"), start_time: str = Query(..., description="分析开始时间（ISO 8601格式）"), - end_time: str = Query(..., description="分析结束时间（ISO 8601格式）"), duration: int = Query(..., description="模拟持续时间（秒）"), ) -> str: """ @@ -476,7 +375,6 @@ async def fastapi_age_analysis( - **network**: 管网名称（或数据库名称） - **start_time**: 分析开始时间 - - **end_time**: 分析结束时间（可选） - **duration**: 模拟持续时间（秒） 分析指定时间段内管网中各节点的水体停留时间。 @@ -546,18 +444,6 @@ async def fastapi_pressure_regulation(data: PressureRegulation = Body(..., descr return "success" -@router.get("/projectmanagement/", summary="项目管理（基础）", description="对管网项目进行基础的管理操作。此为基础版本。") -async def project_management_endpoint(network: str = Query(..., description="管网名称（或数据库名称）")): - """ - 项目管理（基础版本） - - - **network**: 管网名称（或数据库名称） - - 进行基础的项目管理操作。 - """ - return project_management(network) - - @router.post("/project_management/", summary="项目管理（高级）", description="高级版本的项目管理，通过JSON请求体提供详细的控制参数，包括泵控制策略、水箱初始水位和区域需水量控制。") async def fastapi_project_management(data: ProjectManagement = Body(..., description="项目管理控制参数")) -> str: """ @@ -659,18 +545,6 @@ async def fastapi_network_project(file: UploadFile = File(..., description="INP return run_inp(temp_file_name) -@router.get("/networkupdate/", summary="管网更新（基础）", description="对指定管网项目进行基础的更新操作。此为基础版本。") -async def network_update_endpoint(network: str = Query(..., description="管网名称（或数据库名称）")): - """ - 管网更新（基础版本） - - - **network**: 管网名称（或数据库名称） - - 进行管网的基础更新操作。 - """ - return network_update(network) - - @router.post("/network_update/", summary="管网更新（高级）", description="通过上传更新文件对管网进行高级的更新操作。系统将处理更新文件并应用到数据库。") async def fastapi_network_update(file: UploadFile = File(..., description="包含管网更新信息的文件")) -> str: """ @@ -889,7 +763,7 @@ async def fastapi_pressure_sensor_placement( return "success" -@router.post("/runsimulationmanuallybydate/", summary="手动运行日期指定模拟", description="根据指定的日期、开始时间和持续时间，手动运行水力模拟。系统将自动查询管网参数并执行模拟。") +@router.post("/runsimulationmanuallybydate/", summary="手动运行日期指定模拟", description="根据指定的开始时间和持续时间，手动运行水力模拟。开始时间必须是显式带时区的 ISO 8601 / RFC3339 时间。") async def fastapi_run_simulation_manually_by_date( data: RunSimulationManuallyByDate = Body(..., description="模拟运行参数"), ) -> dict[str, str]: @@ -898,14 +772,13 @@ async def fastapi_run_simulation_manually_by_date( 请求体参数： - **name**: 管网名称（或数据库名称） - - **simulation_date**: 模拟基准日期（YYYY-MM-DD格式） - - **start_time**: 开始时间（HH:MM或HH:MM:SS格式） + - **start_time**: 开始时间（ISO 8601 / RFC3339，必须显式带时区） - **duration**: 模拟持续时间（分钟） - 系统将从指定日期和时间开始，按15分钟间隔多次运行模拟。 + 系统将从指定时间开始，按15分钟间隔多次运行模拟。 每次模拟间隔15分钟，直至达到指定的总持续时间。 """ - item = data.dict() + item = data.model_dump() try: simulation.query_corresponding_element_id_and_query_id(item["name"]) simulation.query_corresponding_pattern_id_and_query_id(item["name"]) @@ -932,10 +805,10 @@ async def fastapi_run_simulation_manually_by_date( globals.source_outflow_region_id, globals.realtime_region_pipe_flow_and_demand_id, ) - base_date = datetime.strptime(item["simulation_date"], "%Y-%m-%d") + start_time = parse_utc_time(item["start_time"], field_name="start_time") run_simulation_manually_by_date( - item["name"], base_date, item["start_time"], item["duration"] + item["name"], start_time, item["duration"] ) return {"status": "success"} except Exception as exc: - return {"status": "error", "message": str(exc)} + raise HTTPException(status_code=500, detail=str(exc)) from exc diff --git a/app/infra/epanet/epanet.py b/app/infra/epanet/epanet.py index 8283190..a2e490d 100644 --- a/app/infra/epanet/epanet.py +++ b/app/infra/epanet/epanet.py @@ -8,6 +8,7 @@ from datetime import datetime import subprocess import logging from typing import Any +import uuid sys.path.append("..") from app.native.wndb import project @@ -281,8 +282,6 @@ class Output: def _dump_output(path: str) -> dict[str, Any]: opt = Output(path) data = opt.dump() - with open(path + ".json", "w") as f: - json.dump(data, f) return data @@ -302,25 +301,43 @@ def dump_output_binary(path: str) -> str: return str(bast64_data, "utf-8") +def _safe_remove(path: str) -> None: + try: + if os.path.exists(path): + os.remove(path) + except Exception: + logging.warning("failed to remove temp file: %s", path) + + +def _make_isolated_run_paths(base_name: str, cwd: str) -> tuple[str, str, str]: + # 确保保存临时文件的目录存在 + db_inp_dir = os.path.join(cwd, "db_inp") + temp_dir = os.path.join(cwd, "temp") + os.makedirs(db_inp_dir, exist_ok=True) + os.makedirs(temp_dir, exist_ok=True) + + # 进程号 + UUID 生成唯一后缀，避免并发进程互相覆盖临时文件。 + token = f"{os.getpid()}_{uuid.uuid4().hex}" + inp = os.path.join(db_inp_dir, f"{base_name}.db.{token}.inp") + rpt = os.path.join(temp_dir, f"{base_name}.db.{token}.rpt") + opt = os.path.join(temp_dir, f"{base_name}.db.{token}.opt") + return inp, rpt, opt + + # DingZQ, 2025-02-04, 返回dict[str, Any] -def run_project_return_dict(name: str, readable_output: bool = False) -> dict[str, Any]: +def run_project_return_dict(name: str, readable_output: bool = True) -> dict[str, Any]: if not project.have_project(name): raise Exception(f"Not found project [{name}]") - dir = os.path.abspath(os.getcwd()) + cwd = os.path.abspath(os.getcwd()) - db_inp = os.path.join(os.path.join(dir, "db_inp"), name + ".db.inp") - inp_out.dump_inp(name, db_inp, "2") + inp, rpt, opt = _make_isolated_run_paths(name, cwd) + inp_out.dump_inp(name, inp, "2") - input = name + ".db" if platform.system() == "Windows": exe = os.path.join(os.path.dirname(__file__), "windows", "runepanet.exe") else: exe = os.path.join(os.path.dirname(__file__), "linux", "runepanet") - inp = os.path.join(os.path.join(dir, "db_inp"), input + ".inp") - rpt = os.path.join(os.path.join(dir, "temp"), input + ".rpt") - opt = os.path.join(os.path.join(dir, "temp"), input + ".opt") - command = f"{exe} {inp} {rpt} {opt}" if platform.system() != "Windows": if not os.access(exe, os.X_OK): @@ -334,12 +351,17 @@ def run_project_return_dict(name: str, readable_output: bool = False) -> dict[st lib_dir = os.path.dirname(exe) env["LD_LIBRARY_PATH"] = f"{lib_dir}:{env.get('LD_LIBRARY_PATH', '')}" - # 使用 subprocess 替代 os.system 以传递 env - process = subprocess.run(command, shell=True, env=env) + process = subprocess.run([exe, inp, rpt, opt], env=env, capture_output=True, text=True) result = process.returncode if result != 0: + logging.error(f"EPANET failed with return code {result}") + logging.error(f"EPANET stdout: {process.stdout}") + logging.error(f"EPANET stderr: {process.stderr}") data["simulation_result"] = "failed" + data["error_code"] = result + data["stdout"] = process.stdout + data["stderr"] = process.stderr else: data["simulation_result"] = "successful" if readable_output: @@ -347,33 +369,41 @@ def run_project_return_dict(name: str, readable_output: bool = False) -> dict[st else: data["output"] = dump_output_binary(opt) - data["report"] = dump_report(rpt) + data["input_file"] = inp + data["report_file"] = rpt + data["output_file"] = opt + + if os.path.exists(rpt): + data["report"] = dump_report(rpt) + else: + logging.error(f"EPANET report file not found: {rpt}") + data["report"] = f"Error: EPANET report file not found. Simulation return code: {result}. Check server logs for stdout/stderr." + + # 返回内容后删除仿真临时文件，避免临时文件堆积。 + _safe_remove(inp) + _safe_remove(rpt) + _safe_remove(opt) return data # original code -def run_project(name: str, readable_output: bool = False) -> str: +def run_project(name: str, readable_output: bool = True) -> str: if not project.have_project(name): raise Exception(f"Not found project [{name}]") - dir = os.path.abspath(os.getcwd()) + cwd = os.path.abspath(os.getcwd()) - db_inp = os.path.join(os.path.join(dir, "db_inp"), name + ".db.inp") - inp_out.dump_inp(name, db_inp, "2") + inp, rpt, opt = _make_isolated_run_paths(name, cwd) + inp_out.dump_inp(name, inp, "2") - input = name + ".db" if platform.system() == "Windows": exe = os.path.join(os.path.dirname(__file__), "windows", "runepanet.exe") else: exe = os.path.join(os.path.dirname(__file__), "linux", "runepanet") - inp = os.path.join(os.path.join(dir, "db_inp"), input + ".inp") - rpt = os.path.join(os.path.join(dir, "temp"), input + ".rpt") - opt = os.path.join(os.path.join(dir, "temp"), input + ".opt") - command = f"{exe} {inp} {rpt} {opt}" logging.info(f"Run simulation at {datetime.now()}") - logging.info(command) + logging.info("%s %s %s %s", exe, inp, rpt, opt) if platform.system() != "Windows": if not os.access(exe, os.X_OK): @@ -388,7 +418,7 @@ def run_project(name: str, readable_output: bool = False) -> str: env["LD_LIBRARY_PATH"] = f"{lib_dir}:{env.get('LD_LIBRARY_PATH', '')}" # DingZQ, 2025-06-02, 使用subprocess替代os.system - process = subprocess.run(command, shell=True, env=env) + process = subprocess.run([exe, inp, rpt, opt], env=env) result = process.returncode # logging.info(f"Simulation result: {result}") @@ -402,27 +432,35 @@ def run_project(name: str, readable_output: bool = False) -> str: logging.info("simulation successful") if readable_output: - data |= _dump_output(opt) + data["output"] = _dump_output(opt) else: data["output"] = dump_output_binary(opt) + data["input_file"] = inp + data["report_file"] = rpt + data["output_file"] = opt data["report"] = dump_report(rpt) + + # 返回内容后删除仿真临时文件，避免临时文件堆积。 + _safe_remove(inp) + _safe_remove(rpt) + _safe_remove(opt) # logging.info(f"Report: {data['report']}") return json.dumps(data) -def run_inp(name: str) -> str: - dir = os.path.abspath(os.getcwd()) +def run_inp(name: str, readable_output: bool = True) -> str: + cwd = os.path.abspath(os.getcwd()) if platform.system() == "Windows": exe = os.path.join(os.path.dirname(__file__), "windows", "runepanet.exe") else: exe = os.path.join(os.path.dirname(__file__), "linux", "runepanet") - inp = os.path.join(os.path.join(dir, "inp"), name + ".inp") - rpt = os.path.join(os.path.join(dir, "temp"), name + ".rpt") - opt = os.path.join(os.path.join(dir, "temp"), name + ".opt") - command = f"{exe} {inp} {rpt} {opt}" + source_inp = os.path.join(cwd, "inp", name + ".inp") + token = f"{os.getpid()}_{uuid.uuid4().hex}" + rpt = os.path.join(cwd, "temp", f"{name}.{token}.rpt") + opt = os.path.join(cwd, "temp", f"{name}.{token}.opt") if platform.system() != "Windows": if not os.access(exe, os.X_OK): @@ -436,16 +474,26 @@ def run_inp(name: str) -> str: lib_dir = os.path.dirname(exe) env["LD_LIBRARY_PATH"] = f"{lib_dir}:{env.get('LD_LIBRARY_PATH', '')}" - process = subprocess.run(command, shell=True, env=env) + process = subprocess.run([exe, source_inp, rpt, opt], env=env) result = process.returncode if result != 0: data["simulation_result"] = "failed" else: data["simulation_result"] = "successful" - # data |= _dump_output(opt) - data["output"] = dump_output_binary(opt) + if readable_output: + data["output"] = _dump_output(opt) + else: + data["output"] = dump_output_binary(opt) + data["input_file"] = source_inp + data["report_file"] = rpt + data["output_file"] = opt data["report"] = dump_report(rpt) + # 返回内容后删除仿真临时文件，避免临时文件堆积。 + _safe_remove(source_inp) + _safe_remove(rpt) + _safe_remove(opt) + return json.dumps(data) diff --git a/app/services/simulation.py b/app/services/simulation.py index b2c684f..825976d 100644 --- a/app/services/simulation.py +++ b/app/services/simulation.py @@ -27,15 +27,13 @@ import json import pytz import requests import time -import shutil -from app.infra.epanet.epanet import Output from typing import Optional, Tuple import typing import psycopg import logging import app.services.globals as globals -import uuid import app.services.project_info as project_info +from app.services.time_api import parse_beijing_time from app.core.config import get_pgconn_string from app.infra.db.timescaledb.internal_queries import ( InternalQueries as TimescaleInternalQueries, @@ -663,13 +661,14 @@ def from_seconds_to_clock(secs: int) -> str: def convert_time_format(original_time: str) -> str: """ - 格式转换，将“2024-04-13T08:00:00+08:00"转为“2024-04-13 08:00:00” - :param original_time: str， “2024-04-13T08:00:00+08:00"格式的时间 + 格式转换，将带时区的 ISO 8601 / RFC3339 时间转为北京时间的“YYYY-MM-DD HH:MM:SS” + :param original_time: str，带显式时区的时间 :return: str，“2024-04-13 08:00:00”格式的时间 """ - new_time = original_time.replace("T", " ") - new_time = new_time.replace("+08:00", "") - return new_time + normalized_time = parse_beijing_time( + original_time, field_name="modify_pattern_start_time" + ) + return normalized_time.replace(microsecond=0).strftime("%Y-%m-%d %H:%M:%S") def get_history_pattern_info(project_name, pattern_name): @@ -1218,7 +1217,7 @@ def run_simulation( cs.append(valve_status) set_status(name_c, cs) # 运行并返回结果 - run_project(name_c) + result_data = json.loads(run_project(name_c)) time_cost_end = time.perf_counter() print( "{} -- Hydraulic simulation finished, cost time: {:.2f} s.".format( @@ -1226,23 +1225,22 @@ def run_simulation( time_cost_end - time_cost_start, ) ) - # DingZQ 下面这几句一定要这样，不然读取不了 - # time.sleep(5) # wait 5 seconds - - # TODO: 2025/03/24 - # DingZQ 这个名字要用随机数来处理 - tmp_file = f"./temp/simulation_{uuid.uuid4()}.result.out" - shutil.copy(f"./temp/{name_c}.db.opt", tmp_file) - - output = Output(tmp_file) - node_result = output.node_results() - link_result = output.link_results() - + 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: + raise RuntimeError("run_project output missing node_results or link_results") + # link_flow = [] # for link in link_result: # link_flow.append(link['result'][-1]['flow']) # print(link_flow) - num_periods_result = output.times()["num_periods"] + times_info = output_data.get("times") or {} + num_periods_result = times_info.get("num_periods") + if num_periods_result is None: + raise RuntimeError("run_project output missing times.num_periods") print("simulation_type", simulation_type) print("before store result") # print(num_periods_result) @@ -1275,9 +1273,6 @@ def run_simulation( print("after store result") - del output - os.remove(tmp_file) - if __name__ == "__main__": # 计算前，获取scada_info中的信息，按照设定的方法修改pg数据库 diff --git a/app/services/simulation_ops.py b/app/services/simulation_ops.py index 2110deb..08b8020 100644 --- a/app/services/simulation_ops.py +++ b/app/services/simulation_ops.py @@ -5,7 +5,6 @@ from math import pi import pytz from app.algorithms.simulation.runner import run_simulation_ex -from app.infra.epanet.epanet import Output from app.services.tjnetwork import ( close_project, copy_project, @@ -110,10 +109,16 @@ def scheduling_simulation( + " -- Database Loading OK." ) - run_simulation_ex( - new_name, "realtime", start_time, duration=0, pump_control=pump_control + simulation_result = json.loads( + run_simulation_ex( + new_name, "realtime", start_time, duration=0, pump_control=pump_control + ) ) + output_data = simulation_result.get("output") + if not isinstance(output_data, dict): + raise RuntimeError("run_simulation_ex did not return JSON output content") + if not is_project_open(new_name): open_project(new_name) @@ -131,18 +136,14 @@ def scheduling_simulation( else: tank_pipe_flow_direction[pipe_id] = -1 - output = Output("./temp/{}.db.out".format(new_name)) - - node_results = ( - output.node_results() - ) # [{'node': str, 'result': [{'pressure': float}]}] + node_results = output_data.get("node_results") or [] # [{'node': str, 'result': [{'pressure': float}]}] water_plant_output_pressure = 0 for node_result in node_results: if node_result["node"] == water_plant_output_id: # 水厂出水压力(m) water_plant_output_pressure = node_result["result"][-1]["pressure"] water_plant_output_pressure /= 100 # 预计水厂出水压力(Mpa) - pipe_results = output.link_results() # [{'link': str, 'result': [{'flow': float}]}] + pipe_results = output_data.get("link_results") or [] # [{'link': str, 'result': [{'flow': float}]}] tank_inflow = 0 for pipe_result in pipe_results: for pipe_id in tank_pipes_id: # 遍历与水塔相连的管道 @@ -200,18 +201,24 @@ def daily_scheduling_simulation( + " -- Database Loading OK." ) - run_simulation_ex( - new_name, "realtime", start_time, duration=86400, pump_control=pump_control + simulation_result = json.loads( + run_simulation_ex( + new_name, + "realtime", + start_time, + duration=86400, + pump_control=pump_control, + ) ) + output_data = simulation_result.get("output") + if not isinstance(output_data, dict): + raise RuntimeError("run_simulation_ex did not return JSON output content") + if not is_project_open(new_name): open_project(new_name) - output = Output("./temp/{}.db.out".format(new_name)) - - node_results = ( - output.node_results() - ) # [{'node': str, 'result': [{'pressure': float, 'head': float}]}] + node_results = output_data.get("node_results") or [] # [{'node': str, 'result': [{'pressure': float, 'head': float}]}] water_plant_output_pressure = [] reservoir_level = [] tank_level = []