fix(simulation): align epanet output handling

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}

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

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)

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数据库

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