实现多进程 epanet 模拟，不保留临时文件

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,
@@ -666,21 +665,21 @@ def age_analysis(
         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}

app/api/v1/endpoints/simulation.py

@@ -9,7 +9,6 @@ from fastapi.responses import PlainTextResponse
 import app.infra.db.influxdb.api as influxdb_api
 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,
@@ -139,36 +138,23 @@ def run_simulation_manually_by_date(
 
 
 # 必须用这个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]:
     """
     运行项目模拟（返回字典）
@@ -179,22 +165,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

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,37 @@ 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]:
+    # 进程号 + UUID 生成唯一后缀，避免并发进程互相覆盖临时文件。
+    token = f"{os.getpid()}_{uuid.uuid4().hex}"
+    inp = os.path.join(cwd, "db_inp", f"{base_name}.db.{token}.inp")
+    rpt = os.path.join(cwd, "temp", f"{base_name}.db.{token}.rpt")
+    opt = os.path.join(cwd, "temp", 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,8 +345,7 @@ 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)
     result = process.returncode
 
     if result != 0:
@@ -347,33 +357,36 @@ def run_project_return_dict(name: str, readable_output: bool = False) -> dict[st
         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)
+
     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 +401,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 +415,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 +457,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)

app/services/simulation.py

@@ -27,15 +27,12 @@ import json
 import pytz
 import requests
 import time
-import shutil
-from app.infra.epanet.epanet import Output
 from typing import Optional, Tuple
 import app.infra.db.influxdb.api as influxdb_api
 import typing
 import psycopg
 import logging
 import app.services.globals as globals
-import uuid
 import app.services.project_info as project_info
 from app.core.config import get_pgconn_string
 from app.infra.db.timescaledb.internal_queries import (
@@ -1219,7 +1216,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(
@@ -1227,23 +1224,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)
@@ -1274,27 +1270,8 @@ def run_simulation(
     # 暂不需要再次存储 SCADA 模拟信息
     # TimescaleInternalQueries.fill_scheme_simulation_result_to_SCADA(scheme_type=scheme_type, scheme_name=scheme_name)
 
-    # if simulation_type.upper() == "REALTIME":
-    #     influxdb_api.store_realtime_simulation_result_to_influxdb(
-    #         node_result, link_result, modify_pattern_start_time
-    #     )
-    # elif simulation_type.upper() == "EXTENDED":
-    #     influxdb_api.store_scheme_simulation_result_to_influxdb(
-    #         node_result,
-    #         link_result,
-    #         modify_pattern_start_time,
-    #         num_periods_result,
-    #         scheme_type,
-    #         scheme_name,
-    #     )
-    # 暂不需要再次存储 SCADA 模拟信息
-    # influxdb_api.fill_scheme_simulation_result_to_SCADA(scheme_type=scheme_type, scheme_name=scheme_name)
-
     print("after store result")
 
-    del output
-    os.remove(tmp_file)
-
 
 if __name__ == "__main__":
     # 计算前，获取scada_info中的信息，按照设定的方法修改pg数据库

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 = []

scripts/run_server.py

@@ -16,6 +16,6 @@ if __name__ == "__main__":
         "app.main:app",
         host="0.0.0.0",
         port=8000,
-        # workers=4,  # 这里可以设置多进程
+        # workers=2,  # 这里可以设置多进程
         loop="asyncio",
     )