From 0755b1a61c1ece9ef1b3135bee74d214083a1240 Mon Sep 17 00:00:00 2001 From: Jiang Date: Mon, 2 Feb 2026 18:03:44 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E5=85=B3=E9=98=80=E5=88=86?= =?UTF-8?q?=E6=9E=90=E7=AE=97=E6=B3=95=EF=BC=8C=E6=94=AF=E6=8C=81=E5=A4=9A?= =?UTF-8?q?=E7=AE=A1=E6=AE=B5=E5=88=86=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/algorithms/valve_isolation.py | 58 +++++++++++++++++++++--------- app/api/v1/endpoints/simulation.py | 4 ++- app/services/valve_isolation.py | 4 ++- 3 files changed, 47 insertions(+), 19 deletions(-) diff --git a/app/algorithms/valve_isolation.py b/app/algorithms/valve_isolation.py index efef3d5..7c9f088 100644 --- a/app/algorithms/valve_isolation.py +++ b/app/algorithms/valve_isolation.py @@ -20,26 +20,45 @@ def _parse_link_entry(link_entry: str) -> tuple[str, str, str, str]: return parts[0], parts[1], parts[2], parts[3] -def valve_isolation_analysis(network: str, accident_element: str) -> dict[str, Any]: +def valve_isolation_analysis( + network: str, accident_elements: str | list[str] +) -> dict[str, Any]: """ 关阀搜索/分析：基于拓扑结构确定事故隔离所需关阀。 :param network: 模型名称 - :param accident_element: 事故点（节点或管道/泵/阀门ID） + :param accident_elements: 事故点（节点或管道/泵/阀门ID），可以是单个ID字符串或ID列表 :return: dict，包含受影响节点、必须关闭阀门、可选阀门等信息 """ - if is_node(network, accident_element): - start_nodes = {accident_element} - accident_type = "node" - elif is_link(network, accident_element): - accident_type = get_link_type(network, accident_element) - link_props = get_link_properties(network, accident_element) - node1 = link_props.get("node1") - node2 = link_props.get("node2") - if not node1 or not node2: - raise ValueError("Accident link missing node endpoints") - start_nodes = {node1, node2} + if isinstance(accident_elements, str): + target_elements = [accident_elements] else: - raise ValueError("Accident element not found") + target_elements = accident_elements + + start_nodes = set() + accident_types = set() + + for element in target_elements: + if is_node(network, element): + start_nodes.add(element) + accident_types.add("node") + elif is_link(network, element): + l_type = get_link_type(network, element) + accident_types.add(l_type) + link_props = get_link_properties(network, element) + node1 = link_props.get("node1") + node2 = link_props.get("node2") + if not node1 or not node2: + # 如果是批量处理，可以选择跳过错误或记录错误，这里暂时保持严谨抛出异常 + raise ValueError(f"Accident link {element} missing node endpoints") + start_nodes.add(node1) + start_nodes.add(node2) + else: + raise ValueError(f"Accident element {element} not found") + + # 如果有多种类型混合，简化返回类型描述 + accident_type_str = ( + list(accident_types)[0] if len(accident_types) == 1 else "mixed" + ) adjacency: dict[str, set[str]] = defaultdict(set) valve_links: dict[str, tuple[str, str]] = {} @@ -76,11 +95,16 @@ def valve_isolation_analysis(network: str, accident_element: str) -> dict[str, A must_close_valves.sort() optional_valves.sort() - return { - "accident_element": accident_element, - "accident_type": accident_type, + result = { + "accident_elements": target_elements, + "accident_type": accident_type_str, "affected_nodes": sorted(affected_nodes), "must_close_valves": must_close_valves, "optional_valves": optional_valves, "isolatable": len(must_close_valves) > 0, } + + if len(target_elements) == 1: + result["accident_element"] = target_elements[0] + + return result diff --git a/app/api/v1/endpoints/simulation.py b/app/api/v1/endpoints/simulation.py index 0e4d137..9a5981a 100644 --- a/app/api/v1/endpoints/simulation.py +++ b/app/api/v1/endpoints/simulation.py @@ -238,7 +238,9 @@ async def fastapi_valve_close_analysis( @router.get("/valve_isolation_analysis/") -async def valve_isolation_endpoint(network: str, accident_element: str): +async def valve_isolation_endpoint( + network: str, accident_element: List[str] = Query(...) +): return analyze_valve_isolation(network, accident_element) diff --git a/app/services/valve_isolation.py b/app/services/valve_isolation.py index ae63571..0d3d101 100644 --- a/app/services/valve_isolation.py +++ b/app/services/valve_isolation.py @@ -3,5 +3,7 @@ from typing import Any from app.algorithms.valve_isolation import valve_isolation_analysis -def analyze_valve_isolation(network: str, accident_element: str) -> dict[str, Any]: +def analyze_valve_isolation( + network: str, accident_element: str | list[str] +) -> dict[str, Any]: return valve_isolation_analysis(network, accident_element)