Add fastapi fastapi_run_project_return_dict to return the text version of simulation results

epanet/__init__.py

@@ -1 +1 @@
-from .epanet import run_project, run_inp, dump_output
+from .epanet import run_project, run_project_return_dict, run_inp, dump_output

epanet/epanet.py

@@ -238,7 +238,40 @@ def dump_output_binary(path: str) -> str:
         bast64_data = base64.b64encode(data)
         return str(bast64_data, 'utf-8')
 
+#DingZQ, 2025-02-04, 返回dict[str, Any]
+def run_project_return_dict(name: str, readable_output: bool = False) -> dict[str, Any]:
+    if not project.have_project(name):
+        raise Exception(f'Not found project [{name}]')
 
+    dir = 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')
+
+    input = name + '.db'
+    exe = os.path.join(os.path.join(dir, 'epanet'), 'runepanet.exe')
+    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}'
+
+    data = {}
+
+    result = os.system(command)
+    if result != 0:
+        data['simulation_result'] = 'failed'
+    else:
+        data['simulation_result'] = 'successful'
+        if readable_output:
+            data |= _dump_output(opt)
+        else:
+            data['output'] = dump_output_binary(opt)
+
+    data['report'] = dump_report(rpt)
+
+    return data
+
+# original code
 def run_project(name: str, readable_output: bool = False) -> str:
     if not project.have_project(name):
         raise Exception(f'Not found project [{name}]')

main.py

@@ -207,12 +207,35 @@ async def fastapi_run_project(network: str) -> str:
         raise HTTPException(status_code=409, detail="is in simulation")   
     else:
         try:
-            result = run_project(network)
-            return result
+            return run_project(network)
         finally:
             # 手动释放锁（可选，依赖过期时间自动释放更安全）
             redis_client.delete(lock_key)
             
+# DingZQ, 2025-02-04, 返回dict[str, Any]
+# output 和 report 
+# output 是 json
+# report 是 text
+@app.get("/runprojectreturndict/")
+async def fastapi_run_project_return_dict(network: str) -> dict[str, Any]:
+    lock_key = "exclusive_api_lock"
+    timeout = 120  # 锁自动过期时间（秒）
+
+    # 尝试获取锁（NX=True: 不存在时设置，EX=timeout: 过期时间）
+    acquired = redis_client.set(lock_key, "locked", nx=True, ex=timeout)
+
+    logger.info(f"acquired : {acquired}")
+
+    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)            
+
+        
 # put in inp folder, name without extension
 @app.get("/runinp/")
 async def fastapi_run_inp(network: str) -> str:

tjnetwork.py

@@ -211,6 +211,11 @@ def import_inp(name: str, cs: ChangeSet, version: str = '3') -> bool:
 def export_inp(name: str, version: str = '3') -> ChangeSet:
     return api.export_inp(name, version)
 
+#DingZQ, 2025-02-04, 返回dict[str, Any]
+def run_project_return_dict(name: str) -> dict[str, Any]:
+    return epanet.run_project_return_dict(name)
+
+# original code
 def run_project(name: str) -> str:
     return epanet.run_project(name)