优化API文档,添加参数描述和示例

This commit is contained in:
2026-03-13 15:17:06 +08:00
parent 9a8d851275
commit b513d05611
38 changed files with 5846 additions and 1224 deletions
+88 -13
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -17,21 +17,54 @@ router = APIRouter()
# demand 9.[DEMANDS]
############################################################
@router.get("/getdemandschema")
async def fastapi_get_demand_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get(
"/getdemandschema",
summary="获取需水量属性架构",
description="获取指定水网中需水量(Demand)的属性架构定义"
)
async def fastapi_get_demand_schema(network: str = Query(..., description="管网名称(或数据库名称)")) -> dict[str, dict[str, Any]]:
"""
获取需水量属性架构。
返回指定水网的需水量属性架构,包括所有可配置的属性及其类型定义。
"""
return get_demand_schema(network)
@router.get("/getdemandproperties/")
async def fastapi_get_demand_properties(network: str, junction: str) -> dict[str, Any]:
@router.get(
"/getdemandproperties/",
summary="获取需水量属性",
description="获取指定水网中节点的需水量属性信息"
)
async def fastapi_get_demand_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点ID")
) -> dict[str, Any]:
"""
获取节点的需水量属性。
返回指定节点的所有需水量信息,包括需水量值、水压等级等。
"""
return get_demand(network, junction)
# example: set_demand(p, ChangeSet({'junction': 'j1', 'demands': [{'demand': 10.0, 'pattern': None, 'category': 'x'}, {'demand': 20.0, 'pattern': None, 'category': None}]}))
@router.post("/setdemandproperties/", response_model=None)
@router.post(
"/setdemandproperties/",
response_model=None,
summary="设置需水量属性",
description="设置指定水网中节点的需水量属性信息"
)
async def fastapi_set_demand_properties(
network: str, junction: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点ID"),
req: Request = Body(...)
) -> ChangeSet:
"""
设置节点的需水量属性。
修改指定节点的需水量信息。请求体应包含需水量值、水压等级等属性。
"""
props = await req.json()
ps = {"junction": junction} | props
return set_demand(network, ChangeSet(ps))
@@ -39,26 +72,68 @@ async def fastapi_set_demand_properties(
############################################################
# water distribution 36.[Water Distribution]
############################################################
@router.get("/calculatedemandtonodes/")
@router.get(
"/calculatedemandtonodes/",
summary="计算需水量到节点分配",
description="将总需水量按指定方式分配到多个节点"
)
async def fastapi_calculate_demand_to_nodes(
network: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> dict[str, float]:
"""
计算需水量到节点分配。
将指定的总需水量均匀或按比例分配到指定的节点列表中。
请求体格式:
{
"demand": 需水量值(float),
"nodes": 节点ID列表(list[str])
}
"""
props = await req.json()
demand = props["demand"]
nodes = props["nodes"]
return calculate_demand_to_nodes(network, demand, nodes)
@router.get("/calculatedemandtoregion/")
@router.get(
"/calculatedemandtoregion/",
summary="计算需水量到区域分配",
description="将总需水量按区域特征分配到该区域内的节点"
)
async def fastapi_calculate_demand_to_region(
network: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> dict[str, float]:
"""
计算需水量到区域分配。
根据区域内节点的特征(如面积、人口等)将总需水量分配到该区域的各个节点。
请求体格式:
{
"demand": 需水量值(float),
"region": 区域ID(str)
}
"""
props = await req.json()
demand = props["demand"]
region = props["region"]
return calculate_demand_to_region(network, demand, region)
@router.get("/calculatedemandtonetwork/")
@router.get(
"/calculatedemandtonetwork/",
summary="计算需水量到整网分配",
description="将需水量均匀分配到整个水网的所有需水节点"
)
async def fastapi_calculate_demand_to_network(
network: str, demand: float
network: str = Query(..., description="管网名称(或数据库名称)"),
demand: float = Query(..., description="总需水量(m³/h)", gt=0)
) -> dict[str, float]:
"""
计算需水量到整网分配。
将指定的需水量均匀分配到整个水网的所有需水节点。
"""
return calculate_demand_to_network(network, demand)
+281 -59
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -44,110 +44,291 @@ router = APIRouter()
# type
############################################################
@router.get("/isnode/")
async def fastapi_is_node(network: str, node: str) -> bool:
@router.get(
"/isnode/",
summary="检查节点有效性",
description="检查指定ID是否为水网中的有效节点"
)
async def fastapi_is_node(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> bool:
"""检查指定ID是否为节点。"""
return is_node(network, node)
@router.get("/isjunction/")
async def fastapi_is_junction(network: str, node: str) -> bool:
@router.get(
"/isjunction/",
summary="检查是否为接点",
description="检查指定ID是否为水网中的接点(需求点)"
)
async def fastapi_is_junction(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> bool:
"""检查指定ID是否为接点。"""
return is_junction(network, node)
@router.get("/isreservoir/")
async def fastapi_is_reservoir(network: str, node: str) -> bool:
@router.get(
"/isreservoir/",
summary="检查是否为水源",
description="检查指定ID是否为水网中的水源(水库/河流)"
)
async def fastapi_is_reservoir(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> bool:
"""检查指定ID是否为水源。"""
return is_reservoir(network, node)
@router.get("/istank/")
async def fastapi_is_tank(network: str, node: str) -> bool:
@router.get(
"/istank/",
summary="检查是否为蓄水池",
description="检查指定ID是否为水网中的蓄水池"
)
async def fastapi_is_tank(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> bool:
"""检查指定ID是否为蓄水池。"""
return is_tank(network, node)
@router.get("/islink/")
async def fastapi_is_link(network: str, link: str) -> bool:
@router.get(
"/islink/",
summary="检查管线有效性",
description="检查指定ID是否为水网中的有效管线"
)
async def fastapi_is_link(
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID")
) -> bool:
"""检查指定ID是否为管线。"""
return is_link(network, link)
@router.get("/ispipe/")
async def fastapi_is_pipe(network: str, link: str) -> bool:
@router.get(
"/ispipe/",
summary="检查是否为管道",
description="检查指定ID是否为水网中的管道"
)
async def fastapi_is_pipe(
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID")
) -> bool:
"""检查指定ID是否为管道。"""
return is_pipe(network, link)
@router.get("/ispump/")
async def fastapi_is_pump(network: str, link: str) -> bool:
@router.get(
"/ispump/",
summary="检查是否为泵",
description="检查指定ID是否为水网中的泵"
)
async def fastapi_is_pump(
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID")
) -> bool:
"""检查指定ID是否为泵。"""
return is_pump(network, link)
@router.get("/isvalve/")
async def fastapi_is_valve(network: str, link: str) -> bool:
@router.get(
"/isvalve/",
summary="检查是否为阀门",
description="检查指定ID是否为水网中的阀门"
)
async def fastapi_is_valve(
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID")
) -> bool:
"""检查指定ID是否为阀门。"""
return is_valve(network, link)
@router.get("/getnodetype/")
async def fastapi_get_node_type(network: str, node: str) -> str:
@router.get(
"/getnodetype/",
summary="获取节点类型",
description="获取指定节点的类型(接点/水源/蓄水池)"
)
async def fastapi_get_node_type(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> str:
"""获取节点的类型标识。"""
return get_node_type(network, node)
@router.get("/getlinktype/")
async def fastapi_get_link_type(network: str, link: str) -> str:
@router.get(
"/getlinktype/",
summary="获取管线类型",
description="获取指定管线的类型(管道/泵/阀门)"
)
async def fastapi_get_link_type(
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID")
) -> str:
"""获取管线的类型标识。"""
return get_link_type(network, link)
@router.get("/getelementtype/")
async def fastapi_get_element_type(network: str, element: str) -> str:
@router.get(
"/getelementtype/",
summary="获取元素类型",
description="获取指定元素的类型(节点或管线)"
)
async def fastapi_get_element_type(
network: str = Query(..., description="管网名称(或数据库名称)"),
element: str = Query(..., description="元素ID")
) -> str:
"""获取元素的类型标识。"""
return get_element_type(network, element)
@router.get("/getelementtypevalue/")
async def fastapi_get_element_type_value(network: str, element: str) -> int:
@router.get(
"/getelementtypevalue/",
summary="获取元素类型值",
description="获取指定元素的类型数值标识"
)
async def fastapi_get_element_type_value(
network: str = Query(..., description="管网名称(或数据库名称)"),
element: str = Query(..., description="元素ID")
) -> int:
"""获取元素的类型数值。"""
return get_element_type_value(network, element)
@router.get("/getnodes/")
async def fastapi_get_nodes(network: str) -> list[str]:
@router.get(
"/getnodes/",
summary="获取所有节点",
description="获取指定水网中的所有节点ID列表"
)
async def fastapi_get_nodes(network: str = Query(..., description="管网名称(或数据库名称)")) -> list[str]:
"""获取水网中所有节点的ID列表。"""
return get_nodes(network)
@router.get("/getlinks/")
async def fastapi_get_links(network: str) -> list[str]:
@router.get(
"/getlinks/",
summary="获取所有管线",
description="获取指定水网中的所有管线ID列表"
)
async def fastapi_get_links(network: str = Query(..., description="管网名称(或数据库名称)")) -> list[str]:
"""获取水网中所有管线的ID列表。"""
return get_links(network)
@router.get("/getnodelinks/")
def get_node_links_endpoint(network: str, node: str) -> list[str]:
@router.get(
"/getnodelinks/",
summary="获取节点的关联管线",
description="获取指定节点连接的所有管线ID列表"
)
def get_node_links_endpoint(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> list[str]:
"""获取节点关联的所有管线。"""
return get_node_links(network, node)
############################################################
# Node & Link properties
############################################################
@router.get("/getnodeproperties/")
async def fast_get_node_properties(network: str, node: str) -> dict[str, Any]:
@router.get(
"/getnodeproperties/",
summary="获取节点属性",
description="获取指定节点的所有属性信息"
)
async def fast_get_node_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> dict[str, Any]:
"""获取节点的完整属性信息。"""
return get_node_properties(network, node)
@router.get("/getlinkproperties/")
async def fast_get_link_properties(network: str, link: str) -> dict[str, Any]:
@router.get(
"/getlinkproperties/",
summary="获取管线属性",
description="获取指定管线的所有属性信息"
)
async def fast_get_link_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID")
) -> dict[str, Any]:
"""获取管线的完整属性信息。"""
return get_link_properties(network, link)
@router.get("/getscadaproperties/")
async def fast_get_scada_properties(network: str, scada: str) -> dict[str, Any]:
@router.get(
"/getscadaproperties/",
summary="获取SCADA点属性",
description="获取指定SCADA点的属性信息"
)
async def fast_get_scada_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
scada: str = Query(..., description="SCADA点ID")
) -> dict[str, Any]:
"""获取SCADA点的属性信息。"""
return get_scada_info(network, scada)
@router.get("/getallscadaproperties/")
async def fast_get_all_scada_properties(network: str) -> list[dict[str, Any]]:
@router.get(
"/getallscadaproperties/",
summary="获取所有SCADA点属性",
description="获取指定水网中所有SCADA点的属性信息"
)
async def fast_get_all_scada_properties(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""获取水网中所有SCADA点的属性列表。"""
return get_all_scada_info(network)
@router.get("/getelementpropertieswithtype/")
@router.get(
"/getelementpropertieswithtype/",
summary="获取指定类型元素属性",
description="获取指定类型的元素属性信息"
)
async def fast_get_element_properties_with_type(
network: str, elementtype: str, element: str
network: str = Query(..., description="管网名称(或数据库名称)"),
elementtype: str = Query(..., description="元素类型"),
element: str = Query(..., description="元素ID")
) -> dict[str, Any]:
"""获取指定类型元素的属性。"""
return get_element_properties_with_type(network, elementtype, element)
@router.get("/getelementproperties/")
async def fast_get_element_properties(network: str, element: str) -> dict[str, Any]:
@router.get(
"/getelementproperties/",
summary="获取元素属性",
description="获取指定元素的属性信息"
)
async def fast_get_element_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
element: str = Query(..., description="元素ID")
) -> dict[str, Any]:
"""获取元素的完整属性信息。"""
return get_element_properties(network, element)
############################################################
# title 1.[TITLE]
############################################################
@router.get("/gettitleschema/")
async def fast_get_title_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get(
"/gettitleschema/",
summary="获取标题属性架构",
description="获取指定水网的标题(标题)属性架构定义"
)
async def fast_get_title_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""获取水网标题的属性架构。"""
return get_title_schema(network)
@router.get("/gettitle/")
async def fast_get_title(network: str) -> dict[str, Any]:
@router.get(
"/gettitle/",
summary="获取水网标题属性",
description="获取指定水网的标题(Title)信息"
)
async def fast_get_title(network: str = Query(..., description="管网名称(或数据库名称)")) -> dict[str, Any]:
"""获取水网的标题属性。"""
return get_title(network)
@router.get("/settitle/", response_model=None)
async def fastapi_set_title(network: str, req: Request) -> ChangeSet:
@router.get(
"/settitle/",
response_model=None,
summary="设置水网标题属性",
description="设置指定水网的标题(Title)信息"
)
async def fastapi_set_title(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""设置水网的标题属性。"""
props = await req.json()
return set_title(network, ChangeSet(props))
@@ -155,18 +336,41 @@ async def fastapi_set_title(network: str, req: Request) -> ChangeSet:
# status 10.[STATUS]
############################################################
@router.get("/getstatusschema")
async def fastapi_get_status_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get(
"/getstatusschema",
summary="获取状态属性架构",
description="获取指定水网的状态(Status)属性架构定义"
)
async def fastapi_get_status_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""获取水网状态的属性架构。"""
return get_status_schema(network)
@router.get("/getstatus/")
async def fastapi_get_status(network: str, link: str) -> dict[str, Any]:
@router.get(
"/getstatus/",
summary="获取管线状态",
description="获取指定管线的状态信息"
)
async def fastapi_get_status(
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID")
) -> dict[str, Any]:
"""获取管线的状态属性。"""
return get_status(network, link)
@router.post("/setstatus/", response_model=None)
@router.post(
"/setstatus/",
response_model=None,
summary="设置管线状态",
description="设置指定管线的状态信息"
)
async def fastapi_set_status_properties(
network: str, link: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID"),
req: Request = Body(...)
) -> ChangeSet:
"""设置管线的状态属性。"""
props = await req.json()
ps = {"link": link} | props
return set_status(network, ChangeSet(ps))
@@ -175,8 +379,17 @@ async def fastapi_set_status_properties(
# General Deletion
############################################################
@router.post("/deletenode/", response_model=None)
async def fastapi_delete_node(network: str, node: str) -> ChangeSet:
@router.post(
"/deletenode/",
response_model=None,
summary="删除节点",
description="删除指定的节点(接点/水源/蓄水池)"
)
async def fastapi_delete_node(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> ChangeSet:
"""删除指定的节点。自动识别节点类型并调用相应的删除操作。"""
ps = {"id": node}
if is_junction(network, node):
return delete_junction(network, ChangeSet(ps))
@@ -186,8 +399,17 @@ async def fastapi_delete_node(network: str, node: str) -> ChangeSet:
return delete_tank(network, ChangeSet(ps))
return ChangeSet() # Should probably raise error or return empty
@router.post("/deletelink/", response_model=None)
async def fastapi_delete_link(network: str, link: str) -> ChangeSet:
@router.post(
"/deletelink/",
response_model=None,
summary="删除管线",
description="删除指定的管线(管道/泵/阀门)"
)
async def fastapi_delete_link(
network: str = Query(..., description="管网名称(或数据库名称)"),
link: str = Query(..., description="管线ID")
) -> ChangeSet:
"""删除指定的管线。自动识别管线类型并调用相应的删除操作。"""
ps = {"id": link}
if is_pipe(network, link):
return delete_pipe(network, ChangeSet(ps))
+59 -13
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request, Depends
from fastapi import APIRouter, Request, Depends, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -34,19 +34,44 @@ router = APIRouter()
# props = await req.json()
# return set_coord(network, ChangeSet(props))
@router.get("/getnodecoord/")
async def fastapi_get_node_coord(network: str, node: str) -> dict[str, float] | None:
@router.get(
"/getnodecoord/",
summary="获取节点坐标",
description="获取指定节点的地理坐标(X, Y)"
)
async def fastapi_get_node_coord(
network: str = Query(..., description="管网名称(或数据库名称)"),
node: str = Query(..., description="节点ID")
) -> dict[str, float] | None:
"""获取节点的地理坐标信息。"""
return get_node_coord(network, node)
# Additional geometry queries found in main.py logic (implicit or explicit)
@router.get("/getnetworkinextent/")
@router.get(
"/getnetworkinextent/",
summary="获取范围内的网络元素",
description="获取指定地理范围内的网络节点和管线"
)
async def fastapi_get_network_in_extent(
network: str, x1: float, y1: float, x2: float, y2: float
network: str = Query(..., description="管网名称(或数据库名称)"),
x1: float = Query(..., description="范围左下角X坐标", alias="x1"),
y1: float = Query(..., description="范围左下角Y坐标", alias="y1"),
x2: float = Query(..., description="范围右上角X坐标", alias="x2"),
y2: float = Query(..., description="范围右上角Y坐标", alias="y2")
) -> dict[str, Any]:
"""获取地理范围内的网络几何信息。"""
return get_network_in_extent(network, x1, y1, x2, y2)
@router.get("/getnetworkgeometries/", dependencies=[Depends(verify_token)])
async def fastapi_get_network_geometries(network: str) -> dict[str, Any] | None:
@router.get(
"/getnetworkgeometries/",
dependencies=[Depends(verify_token)],
summary="获取完整网络几何信息",
description="获取整个水网的所有节点、管线和SCADA点的几何信息(需要身份验证)"
)
async def fastapi_get_network_geometries(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, Any] | None:
"""获取完整的网络几何信息,包括所有节点、管线和SCADA点。结果从缓存返回。"""
cache_key = f"getnetworkgeometries_{network}"
data = redis_client.get(cache_key)
if data:
@@ -64,18 +89,39 @@ async def fastapi_get_network_geometries(network: str) -> dict[str, Any] | None:
redis_client.set(cache_key, msgpack.packb(results, default=encode_datetime))
return results
@router.get("/getmajornodecoords/")
@router.get(
"/getmajornodecoords/",
summary="获取主要节点坐标",
description="获取直径大于等于指定值的节点坐标"
)
async def fastapi_get_majornode_coords(
network: str, diameter: int
network: str = Query(..., description="管网名称(或数据库名称)"),
diameter: int = Query(..., description="最小直径(mm)", gt=0)
) -> dict[str, dict[str, float]]:
"""获取主要节点的坐标。只返回直径大于等于指定值的节点。"""
return get_major_node_coords(network, diameter)
@router.get("/getmajorpipenodes/")
async def fastapi_get_major_pipe_nodes(network: str, diameter: int) -> list[str] | None:
@router.get(
"/getmajorpipenodes/",
summary="获取主要管道节点",
description="获取直径大于等于指定值的管道的节点ID"
)
async def fastapi_get_major_pipe_nodes(
network: str = Query(..., description="管网名称(或数据库名称)"),
diameter: int = Query(..., description="最小直径(mm)", gt=0)
) -> list[str] | None:
"""获取主要管道节点。只返回直径大于等于指定值的管道。"""
return get_major_pipe_nodes(network, diameter)
@router.get("/getnetworklinknodes/")
async def fastapi_get_network_link_nodes(network: str) -> list[str] | None:
@router.get(
"/getnetworklinknodes/",
summary="获取网络管线节点",
description="获取指定水网所有管线的起点和终点节点"
)
async def fastapi_get_network_link_nodes(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[str] | None:
"""获取网络中所有管线的连接节点。"""
return get_network_link_nodes(network)
# @router.get("/getallcoords/")
+278 -37
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -13,108 +13,349 @@ from app.services.tjnetwork import (
router = APIRouter()
@router.get("/getjunctionschema")
async def fast_get_junction_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get("/getjunctionschema", summary="获取节点架构", description="获取指定项目的节点属性架构和数据类型定义。")
async def fast_get_junction_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""
获取节点架构信息。
返回指定项目的节点属性架构,包括所有属性的类型和约束信息。
Args:
network: 管网名称(或数据库名称)
"""
return get_junction_schema(network)
@router.post("/addjunction/", response_model=None)
@router.post("/addjunction/", response_model=None, summary="添加节点", description="在供水网络中添加新的节点,指定节点ID和空间坐标。")
async def fastapi_add_junction(
network: str, junction: str, x: float, y: float, z: float
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID"),
x: float = Query(..., description="X 坐标"),
y: float = Query(..., description="Y 坐标"),
z: float = Query(..., description="标高(海拔高度)")
) -> ChangeSet:
"""
添加新节点到供水网络。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
x: X 坐标值
y: Y 坐标值
z: 标高(海拔高度)
Returns:
ChangeSet: 包含变更信息的结果
"""
ps = {"id": junction, "x": x, "y": y, "elevation": z}
return add_junction(network, ChangeSet(ps))
@router.post("/deletejunction/", response_model=None)
async def fastapi_delete_junction(network: str, junction: str) -> ChangeSet:
@router.post("/deletejunction/", response_model=None, summary="删除节点", description="从供水网络中删除指定的节点。")
async def fastapi_delete_junction(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID")
) -> ChangeSet:
"""
删除指定的节点。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
Returns:
ChangeSet: 包含变更信息的结果
"""
ps = {"id": junction}
return delete_junction(network, ChangeSet(ps))
@router.get("/getjunctionelevation/")
async def fastapi_get_junction_elevation(network: str, junction: str) -> float:
@router.get("/getjunctionelevation/", summary="获取节点标高", description="获取指定节点的标高(海拔高度)。")
async def fastapi_get_junction_elevation(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID")
) -> float:
"""
获取节点的标高值。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
Returns:
float: 节点标高值
"""
ps = get_junction(network, junction)
return ps["elevation"]
@router.get("/getjunctionx/")
async def fastapi_get_junction_x(network: str, junction: str) -> float:
@router.get("/getjunctionx/", summary="获取节点 X 坐标", description="获取指定节点的 X 坐标值。")
async def fastapi_get_junction_x(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID")
) -> float:
"""
获取节点的 X 坐标。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
Returns:
float: 节点 X 坐标值
"""
ps = get_junction(network, junction)
return ps["x"]
@router.get("/getjunctiony/")
async def fastapi_get_junction_y(network: str, junction: str) -> float:
@router.get("/getjunctiony/", summary="获取节点 Y 坐标", description="获取指定节点的 Y 坐标值。")
async def fastapi_get_junction_y(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID")
) -> float:
"""
获取节点的 Y 坐标。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
Returns:
float: 节点 Y 坐标值
"""
ps = get_junction(network, junction)
return ps["y"]
@router.get("/getjunctioncoord/")
async def fastapi_get_junction_coord(network: str, junction: str) -> dict[str, float]:
@router.get("/getjunctioncoord/", summary="获取节点坐标", description="获取指定节点的 X 和 Y 坐标。")
async def fastapi_get_junction_coord(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID")
) -> dict[str, float]:
"""
获取节点的坐标信息。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
Returns:
dict: 包含 x 和 y 坐标的字典
"""
ps = get_junction(network, junction)
coord = {"x": ps["x"], "y": ps["y"]}
return coord
@router.get("/getjunctiondemand/")
async def fastapi_get_junction_demand(network: str, junction: str) -> float:
@router.get("/getjunctiondemand/", summary="获取节点需水量", description="获取指定节点的需水量。")
async def fastapi_get_junction_demand(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID")
) -> float:
"""
获取节点的需水量。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
Returns:
float: 节点的需水量值
"""
ps = get_junction(network, junction)
return ps["demand"]
@router.get("/getjunctionpattern/")
async def fastapi_get_junction_pattern(network: str, junction: str) -> str:
@router.get("/getjunctionpattern/", summary="获取节点需水模式", description="获取指定节点的需水模式标识。")
async def fastapi_get_junction_pattern(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID")
) -> str:
"""
获取节点的需水模式。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
Returns:
str: 节点的需水模式标识
"""
ps = get_junction(network, junction)
return ps["pattern"]
@router.post("/setjunctionelevation/", response_model=None)
@router.post("/setjunctionelevation/", response_model=None, summary="设置节点标高", description="设置指定节点的标高值。")
async def fastapi_set_junction_elevation(
network: str, junction: str, elevation: float
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID"),
elevation: float = Query(..., description="标高(海拔高度)")
) -> ChangeSet:
"""
设置节点的标高。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
elevation: 标高值
Returns:
ChangeSet: 包含变更信息的结果
"""
ps = {"id": junction, "elevation": elevation}
return set_junction(network, ChangeSet(ps))
@router.post("/setjunctionx/", response_model=None)
async def fastapi_set_junction_x(network: str, junction: str, x: float) -> ChangeSet:
@router.post("/setjunctionx/", response_model=None, summary="设置节点 X 坐标", description="设置指定节点的 X 坐标值。")
async def fastapi_set_junction_x(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID"),
x: float = Query(..., description="X 坐标值")
) -> ChangeSet:
"""
设置节点的 X 坐标。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
x: X 坐标值
Returns:
ChangeSet: 包含变更信息的结果
"""
ps = {"id": junction, "x": x}
return set_junction(network, ChangeSet(ps))
@router.post("/setjunctiony/", response_model=None)
async def fastapi_set_junction_y(network: str, junction: str, y: float) -> ChangeSet:
@router.post("/setjunctiony/", response_model=None, summary="设置节点 Y 坐标", description="设置指定节点的 Y 坐标值。")
async def fastapi_set_junction_y(
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID"),
y: float = Query(..., description="Y 坐标值")
) -> ChangeSet:
"""
设置节点的 Y 坐标。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
y: Y 坐标值
Returns:
ChangeSet: 包含变更信息的结果
"""
ps = {"id": junction, "y": y}
return set_junction(network, ChangeSet(ps))
@router.post("/setjunctioncoord/", response_model=None)
@router.post("/setjunctioncoord/", response_model=None, summary="设置节点坐标", description="设置指定节点的 X 和 Y 坐标。")
async def fastapi_set_junction_coord(
network: str, junction: str, x: float, y: float
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID"),
x: float = Query(..., description="X 坐标值"),
y: float = Query(..., description="Y 坐标值")
) -> ChangeSet:
"""
设置节点的坐标。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
x: X 坐标值
y: Y 坐标值
Returns:
ChangeSet: 包含变更信息的结果
"""
ps = {"id": junction, "x": x, "y": y}
return set_junction(network, ChangeSet(ps))
@router.post("/setjunctiondemand/", response_model=None)
@router.post("/setjunctiondemand/", response_model=None, summary="设置节点需水量", description="设置指定节点的需水量。")
async def fastapi_set_junction_demand(
network: str, junction: str, demand: float
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID"),
demand: float = Query(..., description="需水量值")
) -> ChangeSet:
"""
设置节点的需水量。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
demand: 需水量值
Returns:
ChangeSet: 包含变更信息的结果
"""
ps = {"id": junction, "demand": demand}
return set_junction(network, ChangeSet(ps))
@router.post("/setjunctionpattern/", response_model=None)
@router.post("/setjunctionpattern/", response_model=None, summary="设置节点需水模式", description="设置指定节点的需水模式标识。")
async def fastapi_set_junction_pattern(
network: str, junction: str, pattern: str
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID"),
pattern: str = Query(..., description="需水模式标识")
) -> ChangeSet:
"""
设置节点的需水模式。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
pattern: 需水模式标识
Returns:
ChangeSet: 包含变更信息的结果
"""
ps = {"id": junction, "pattern": pattern}
return set_junction(network, ChangeSet(ps))
@router.get("/getjunctionproperties/")
@router.get("/getjunctionproperties/", summary="获取节点属性", description="获取指定节点的所有属性信息。")
async def fastapi_get_junction_properties(
network: str, junction: str
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID")
) -> dict[str, Any]:
"""
获取节点的完整属性信息。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
Returns:
dict: 包含节点所有属性的字典
"""
return get_junction(network, junction)
@router.get("/getalljunctionproperties/")
async def fastapi_get_all_junction_properties(network: str) -> list[dict[str, Any]]:
@router.get("/getalljunctionproperties/", summary="获取所有节点属性", description="获取指定项目中所有节点的属性信息。")
async def fastapi_get_all_junction_properties(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""
获取所有节点的属性信息列表。
此端点返回指定项目中所有节点的详细属性。缓存查询结果以提高性能。
Args:
network: 管网名称(或数据库名称)
Returns:
list: 包含所有节点属性的列表
"""
# 缓存查询结果提高性能
# global redis_client # Redis logic removed for clean split, can be re-added if needed or imported
results = get_all_junctions(network)
return results
@router.post("/setjunctionproperties/", response_model=None)
@router.post("/setjunctionproperties/", response_model=None, summary="批量设置节点属性", description="批量设置指定节点的多个属性。")
async def fastapi_set_junction_properties(
network: str, junction: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
junction: str = Query(..., description="节点 ID"),
req: Request = Body(..., description="包含待设置属性的请求体")
) -> ChangeSet:
"""
批量设置节点属性。
允许一次性设置节点的多个属性,如坐标、标高、需水量等。
Args:
network: 管网名称(或数据库名称)
junction: 节点 ID
req: 包含属性和值的 JSON 请求体
Returns:
ChangeSet: 包含变更信息的结果
"""
props = await req.json()
ps = {"id": junction} | props
return set_junction(network, ChangeSet(ps))
+317 -49
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -14,22 +14,50 @@ from app.services.tjnetwork import (
router = APIRouter()
@router.get("/getpipeschema")
async def fastapi_get_pipe_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get("/getpipeschema", summary="获取管道模式", description="获取管道对象的模式定义,包含所有可用字段及其类型")
async def fastapi_get_pipe_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""
获取管道数据模式定义。
Args:
network: 管网名称(或数据库名称)
Returns:
包含管道模式信息的字典
"""
return get_pipe_schema(network)
@router.post("/addpipe/", response_model=None)
@router.post("/addpipe/", response_model=None, summary="添加管道", description="向网络中添加新的管道,需要提供管道的基本参数如长度、管径、粗糙度等")
async def fastapi_add_pipe(
network: str,
pipe: str,
node1: str,
node2: str,
length: float = 0,
diameter: float = 0,
roughness: float = 0,
minor_loss: float = 0,
status: str = PIPE_STATUS_OPEN,
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道标识符"),
node1: str = Query(..., description="管道起始节点ID"),
node2: str = Query(..., description="管道终止节点ID"),
length: float = Query(0, description="管道长度(单位:米)"),
diameter: float = Query(0, description="管道管径(单位:毫米)"),
roughness: float = Query(0, description="管道粗糙度"),
minor_loss: float = Query(0, description="管道局部阻力系数"),
status: str = Query(PIPE_STATUS_OPEN, description="管道状态(开启/关闭)"),
) -> ChangeSet:
"""
添加新管道到网络。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
node1: 起始节点ID
node2: 终止节点ID
length: 管道长度
diameter: 管道管径
roughness: 管道粗糙度
minor_loss: 局部阻力系数
status: 管道状态
Returns:
ChangeSet对象,包含本次操作的变更信息
"""
ps = {
"id": pipe,
"node1": node1,
@@ -42,102 +70,342 @@ async def fastapi_add_pipe(
}
return add_pipe(network, ChangeSet(ps))
@router.post("/deletepipe/", response_model=None)
async def fastapi_delete_pipe(network: str, pipe: str) -> ChangeSet:
@router.post("/deletepipe/", response_model=None, summary="删除管道", description="从网络中删除指定的管道")
async def fastapi_delete_pipe(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="要删除的管道ID")
) -> ChangeSet:
"""
删除管道。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
ChangeSet对象,包含本次删除操作的变更信息
"""
ps = {"id": pipe}
return delete_pipe(network, ChangeSet(ps))
@router.get("/getpipenode1/")
async def fastapi_get_pipe_node1(network: str, pipe: str) -> str | None:
@router.get("/getpipenode1/", summary="获取管道起始节点", description="获取指定管道的起始节点ID")
async def fastapi_get_pipe_node1(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID")
) -> str | None:
"""
获取管道的起始节点。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
起始节点ID,如果不存在则返回None
"""
ps = get_pipe(network, pipe)
return ps["node1"]
@router.get("/getpipenode2/")
async def fastapi_get_pipe_node2(network: str, pipe: str) -> str | None:
@router.get("/getpipenode2/", summary="获取管道终止节点", description="获取指定管道的终止节点ID")
async def fastapi_get_pipe_node2(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID")
) -> str | None:
"""
获取管道的终止节点。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
终止节点ID,如果不存在则返回None
"""
ps = get_pipe(network, pipe)
return ps["node2"]
@router.get("/getpipelength/")
async def fastapi_get_pipe_length(network: str, pipe: str) -> float | None:
@router.get("/getpipelength/", summary="获取管道长度", description="获取指定管道的长度")
async def fastapi_get_pipe_length(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID")
) -> float | None:
"""
获取管道长度。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
管道长度(单位:米),如果不存在则返回None
"""
ps = get_pipe(network, pipe)
return ps["length"]
@router.get("/getpipediameter/")
async def fastapi_get_pipe_diameter(network: str, pipe: str) -> float | None:
@router.get("/getpipediameter/", summary="获取管道管径", description="获取指定管道的管径")
async def fastapi_get_pipe_diameter(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID")
) -> float | None:
"""
获取管道管径。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
管道管径(单位:毫米),如果不存在则返回None
"""
ps = get_pipe(network, pipe)
return ps["diameter"]
@router.get("/getpiperoughness/")
async def fastapi_get_pipe_roughness(network: str, pipe: str) -> float | None:
@router.get("/getpiperoughness/", summary="获取管道粗糙度", description="获取指定管道的粗糙度")
async def fastapi_get_pipe_roughness(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID")
) -> float | None:
"""
获取管道粗糙度。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
管道粗糙度值,如果不存在则返回None
"""
ps = get_pipe(network, pipe)
return ps["roughness"]
@router.get("/getpipeminorloss/")
async def fastapi_get_pipe_minor_loss(network: str, pipe: str) -> float | None:
@router.get("/getpipeminorloss/", summary="获取管道局部阻力系数", description="获取指定管道的局部阻力系数")
async def fastapi_get_pipe_minor_loss(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID")
) -> float | None:
"""
获取管道局部阻力系数。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
局部阻力系数,如果不存在则返回None
"""
ps = get_pipe(network, pipe)
return ps["minor_loss"]
@router.get("/getpipestatus/")
async def fastapi_get_pipe_status(network: str, pipe: str) -> str | None:
@router.get("/getpipestatus/", summary="获取管道状态", description="获取指定管道的状态(开启或关闭)")
async def fastapi_get_pipe_status(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID")
) -> str | None:
"""
获取管道状态。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
管道状态(开启/关闭),如果不存在则返回None
"""
ps = get_pipe(network, pipe)
return ps["status"]
@router.post("/setpipenode1/", response_model=None)
async def fastapi_set_pipe_node1(network: str, pipe: str, node1: str) -> ChangeSet:
@router.post("/setpipenode1/", response_model=None, summary="设置管道起始节点", description="设置指定管道的起始节点")
async def fastapi_set_pipe_node1(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID"),
node1: str = Query(..., description="新的起始节点ID")
) -> ChangeSet:
"""
设置管道起始节点。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
node1: 新的起始节点ID
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pipe, "node1": node1}
return set_pipe(network, ChangeSet(ps))
@router.post("/setpipenode2/", response_model=None)
async def fastapi_set_pipe_node2(network: str, pipe: str, node2: str) -> ChangeSet:
@router.post("/setpipenode2/", response_model=None, summary="设置管道终止节点", description="设置指定管道的终止节点")
async def fastapi_set_pipe_node2(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID"),
node2: str = Query(..., description="新的终止节点ID")
) -> ChangeSet:
"""
设置管道终止节点。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
node2: 新的终止节点ID
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pipe, "node2": node2}
return set_pipe(network, ChangeSet(ps))
@router.post("/setpipelength/", response_model=None)
async def fastapi_set_pipe_length(network: str, pipe: str, length: float) -> ChangeSet:
@router.post("/setpipelength/", response_model=None, summary="设置管道长度", description="设置指定管道的长度")
async def fastapi_set_pipe_length(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID"),
length: float = Query(..., description="新的管道长度(单位:米)")
) -> ChangeSet:
"""
设置管道长度。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
length: 新的管道长度
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pipe, "length": length}
return set_pipe(network, ChangeSet(ps))
@router.post("/setpipediameter/", response_model=None)
@router.post("/setpipediameter/", response_model=None, summary="设置管道管径", description="设置指定管道的管径")
async def fastapi_set_pipe_diameter(
network: str, pipe: str, diameter: float
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID"),
diameter: float = Query(..., description="新的管道管径(单位:毫米)")
) -> ChangeSet:
"""
设置管道管径。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
diameter: 新的管道管径
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pipe, "diameter": diameter}
return set_pipe(network, ChangeSet(ps))
@router.post("/setpiperoughness/", response_model=None)
@router.post("/setpiperoughness/", response_model=None, summary="设置管道粗糙度", description="设置指定管道的粗糙度")
async def fastapi_set_pipe_roughness(
network: str, pipe: str, roughness: float
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID"),
roughness: float = Query(..., description="新的管道粗糙度值")
) -> ChangeSet:
"""
设置管道粗糙度。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
roughness: 新的管道粗糙度
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pipe, "roughness": roughness}
return set_pipe(network, ChangeSet(ps))
@router.post("/setpipeminorloss/", response_model=None)
@router.post("/setpipeminorloss/", response_model=None, summary="设置管道局部阻力系数", description="设置指定管道的局部阻力系数")
async def fastapi_set_pipe_minor_loss(
network: str, pipe: str, minor_loss: float
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID"),
minor_loss: float = Query(..., description="新的局部阻力系数值")
) -> ChangeSet:
"""
设置管道局部阻力系数。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
minor_loss: 新的局部阻力系数
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pipe, "minor_loss": minor_loss}
return set_pipe(network, ChangeSet(ps))
@router.post("/setpipestatus/", response_model=None)
async def fastapi_set_pipe_status(network: str, pipe: str, status: str) -> ChangeSet:
@router.post("/setpipestatus/", response_model=None, summary="设置管道状态", description="设置指定管道的状态(开启或关闭)")
async def fastapi_set_pipe_status(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID"),
status: str = Query(..., description="新的管道状态(开启/关闭)")
) -> ChangeSet:
"""
设置管道状态。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
status: 新的管道状态
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pipe, "status": status}
return set_pipe(network, ChangeSet(ps))
@router.get("/getpipeproperties/")
async def fastapi_get_pipe_properties(network: str, pipe: str) -> dict[str, Any]:
@router.get("/getpipeproperties/", summary="获取管道属性", description="获取指定管道的所有属性信息")
async def fastapi_get_pipe_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID")
) -> dict[str, Any]:
"""
获取管道的所有属性。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
Returns:
包含管道所有属性的字典
"""
return get_pipe(network, pipe)
@router.get("/getallpipeproperties/")
async def fastapi_get_all_pipe_properties(network: str) -> list[dict[str, Any]]:
@router.get("/getallpipeproperties/", summary="获取所有管道属性", description="获取网络中所有管道的属性信息列表")
async def fastapi_get_all_pipe_properties(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""
获取网络中所有管道的属性。
Args:
network: 管网名称(或数据库名称)
Returns:
包含所有管道属性的字典列表
"""
# 缓存查询结果提高性能
# global redis_client
results = get_all_pipes(network)
return results
@router.post("/setpipeproperties/", response_model=None)
@router.post("/setpipeproperties/", response_model=None, summary="设置管道属性", description="批量设置指定管道的多个属性")
async def fastapi_set_pipe_properties(
network: str, pipe: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
pipe: str = Query(..., description="管道ID"),
req: Request = Body(...)
) -> ChangeSet:
"""
批量设置管道属性。
Args:
network: 管网名称(或数据库名称)
pipe: 管道ID
req: 请求体,包含要设置的属性及其值
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
props = await req.json()
ps = {"id": pipe} | props
return set_pipe(network, ChangeSet(ps))
+155 -21
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -13,57 +13,191 @@ from app.services.tjnetwork import (
router = APIRouter()
@router.get("/getpumpschema")
async def fastapi_get_pump_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get("/getpumpschema", summary="获取水泵模式", description="获取水泵对象的模式定义,包含所有可用字段及其类型")
async def fastapi_get_pump_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""
获取水泵数据模式定义。
Args:
network: 管网名称(或数据库名称)
Returns:
包含水泵模式信息的字典
"""
return get_pump_schema(network)
@router.post("/addpump/", response_model=None)
@router.post("/addpump/", response_model=None, summary="添加水泵", description="向网络中添加新的水泵,需要提供水泵的基本参数如功率等")
async def fastapi_add_pump(
network: str, pump: str, node1: str, node2: str, power: float = 0.0
network: str = Query(..., description="管网名称(或数据库名称)"),
pump: str = Query(..., description="水泵标识符"),
node1: str = Query(..., description="水泵起始节点ID"),
node2: str = Query(..., description="水泵终止节点ID"),
power: float = Query(0.0, description="水泵功率(单位:千瓦)")
) -> ChangeSet:
"""
添加新水泵到网络。
Args:
network: 管网名称(或数据库名称)
pump: 水泵ID
node1: 起始节点ID
node2: 终止节点ID
power: 水泵功率
Returns:
ChangeSet对象,包含本次操作的变更信息
"""
ps = {"id": pump, "node1": node1, "node2": node2, "power": power}
return add_pump(network, ChangeSet(ps))
@router.post("/deletepump/", response_model=None)
async def fastapi_delete_pump(network: str, pump: str) -> ChangeSet:
@router.post("/deletepump/", response_model=None, summary="删除水泵", description="从网络中删除指定的水泵")
async def fastapi_delete_pump(
network: str = Query(..., description="管网名称(或数据库名称)"),
pump: str = Query(..., description="要删除的水泵ID")
) -> ChangeSet:
"""
删除水泵。
Args:
network: 管网名称(或数据库名称)
pump: 水泵ID
Returns:
ChangeSet对象,包含本次删除操作的变更信息
"""
ps = {"id": pump}
return delete_pump(network, ChangeSet(ps))
@router.get("/getpumpnode1/")
async def fastapi_get_pump_node1(network: str, pump: str) -> str | None:
@router.get("/getpumpnode1/", summary="获取水泵起始节点", description="获取指定水泵的起始节点ID")
async def fastapi_get_pump_node1(
network: str = Query(..., description="管网名称(或数据库名称)"),
pump: str = Query(..., description="水泵ID")
) -> str | None:
"""
获取水泵的起始节点。
Args:
network: 管网名称(或数据库名称)
pump: 水泵ID
Returns:
起始节点ID,如果不存在则返回None
"""
ps = get_pump(network, pump)
return ps["node1"]
@router.get("/getpumpnode2/")
async def fastapi_get_pump_node2(network: str, pump: str) -> str | None:
@router.get("/getpumpnode2/", summary="获取水泵终止节点", description="获取指定水泵的终止节点ID")
async def fastapi_get_pump_node2(
network: str = Query(..., description="管网名称(或数据库名称)"),
pump: str = Query(..., description="水泵ID")
) -> str | None:
"""
获取水泵的终止节点。
Args:
network: 管网名称(或数据库名称)
pump: 水泵ID
Returns:
终止节点ID,如果不存在则返回None
"""
ps = get_pump(network, pump)
return ps["node2"]
@router.post("/setpumpnode1/", response_model=None)
async def fastapi_set_pump_node1(network: str, pump: str, node1: str) -> ChangeSet:
@router.post("/setpumpnode1/", response_model=None, summary="设置水泵起始节点", description="设置指定水泵的起始节点")
async def fastapi_set_pump_node1(
network: str = Query(..., description="管网名称(或数据库名称)"),
pump: str = Query(..., description="水泵ID"),
node1: str = Query(..., description="新的起始节点ID")
) -> ChangeSet:
"""
设置水泵起始节点。
Args:
network: 管网名称(或数据库名称)
pump: 水泵ID
node1: 新的起始节点ID
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pump, "node1": node1}
return set_pump(network, ChangeSet(ps))
@router.post("/setpumpnode2/", response_model=None)
async def fastapi_set_pump_node2(network: str, pump: str, node2: str) -> ChangeSet:
@router.post("/setpumpnode2/", response_model=None, summary="设置水泵终止节点", description="设置指定水泵的终止节点")
async def fastapi_set_pump_node2(
network: str = Query(..., description="管网名称(或数据库名称)"),
pump: str = Query(..., description="水泵ID"),
node2: str = Query(..., description="新的终止节点ID")
) -> ChangeSet:
"""
设置水泵终止节点。
Args:
network: 管网名称(或数据库名称)
pump: 水泵ID
node2: 新的终止节点ID
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
ps = {"id": pump, "node2": node2}
return set_pump(network, ChangeSet(ps))
@router.get("/getpumpproperties/")
async def fastapi_get_pump_properties(network: str, pump: str) -> dict[str, Any]:
@router.get("/getpumpproperties/", summary="获取水泵属性", description="获取指定水泵的所有属性信息")
async def fastapi_get_pump_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
pump: str = Query(..., description="水泵ID")
) -> dict[str, Any]:
"""
获取水泵的所有属性。
Args:
network: 管网名称(或数据库名称)
pump: 水泵ID
Returns:
包含水泵所有属性的字典
"""
return get_pump(network, pump)
@router.get("/getallpumpproperties/")
async def fastapi_get_all_pump_properties(network: str) -> list[dict[str, Any]]:
@router.get("/getallpumpproperties/", summary="获取所有水泵属性", description="获取网络中所有水泵的属性信息列表")
async def fastapi_get_all_pump_properties(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""
获取网络中所有水泵的属性。
Args:
network: 管网名称(或数据库名称)
Returns:
包含所有水泵属性的字典列表
"""
# 缓存查询结果提高性能
# global redis_client
results = get_all_pumps(network)
return results
@router.post("/setpumpproperties/", response_model=None)
@router.post("/setpumpproperties/", response_model=None, summary="设置水泵属性", description="批量设置指定水泵的多个属性")
async def fastapi_set_pump_properties(
network: str, pump: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
pump: str = Query(..., description="水泵ID"),
req: Request = Body(...)
) -> ChangeSet:
"""
批量设置水泵属性。
Args:
network: 管网名称(或数据库名称)
pump: 水泵ID
req: 请求体,包含要设置的属性及其值
Returns:
ChangeSet对象,包含本次修改的变更信息
"""
props = await req.json()
ps = {"id": pump} | props
return set_pump(network, ChangeSet(ps))
+396 -75
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -7,9 +7,11 @@ from app.services.tjnetwork import (
add_region,
add_service_area,
add_virtual_district,
calculate_district_metering_area,
calculate_district_metering_area_for_network,
calculate_district_metering_area_for_nodes,
calculate_district_metering_area_for_region,
calculate_region,
calculate_service_area,
calculate_virtual_district,
delete_district_metering_area,
@@ -17,11 +19,13 @@ from app.services.tjnetwork import (
delete_service_area,
delete_virtual_district,
generate_district_metering_area,
generate_region,
generate_service_area,
generate_sub_district_metering_area,
generate_virtual_district,
get_all_district_metering_area_ids,
get_all_district_metering_areas,
get_all_regions,
get_all_service_areas,
get_all_virtual_districts,
get_district_metering_area,
@@ -44,41 +48,105 @@ router = APIRouter()
# region 32
############################################################
@router.get("/calculateregion/")
async def fastapi_calculate_region(network: str, time_index: int) -> dict[str, Any]:
@router.get(
"/calculateregion/",
summary="计算区域",
description="计算指定水网在指定时间步长的区域分区"
)
async def fastapi_calculate_region(
network: str = Query(..., description="管网名称(或数据库名称)"),
time_index: int = Query(..., description="时间步长索引", ge=0)
) -> dict[str, Any]:
"""计算区域分区。"""
return calculate_region(network, time_index)
@router.get("/getregionschema/")
async def fastapi_get_region_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get(
"/getregionschema/",
summary="获取区域属性架构",
description="获取指定水网的区域属性架构定义"
)
async def fastapi_get_region_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""获取区域的属性架构。"""
return get_region_schema(network)
@router.get("/getregion/")
async def fastapi_get_region(network: str, id: str) -> dict[str, Any]:
@router.get(
"/getregion/",
summary="获取区域信息",
description="获取指定ID的区域详细信息"
)
async def fastapi_get_region(
network: str = Query(..., description="管网名称(或数据库名称)"),
id: str = Query(..., description="区域ID")
) -> dict[str, Any]:
"""获取区域的详细信息。"""
return get_region(network, id)
@router.post("/setregion/", response_model=None)
async def fastapi_set_region(network: str, req: Request) -> ChangeSet:
@router.post(
"/setregion/",
response_model=None,
summary="设置区域属性",
description="修改指定区域的属性信息"
)
async def fastapi_set_region(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""设置区域属性。"""
props = await req.json()
return set_region(network, ChangeSet(props))
@router.post("/addregion/", response_model=None)
async def fastapi_add_region(network: str, req: Request) -> ChangeSet:
@router.post(
"/addregion/",
response_model=None,
summary="添加新区域",
description="向水网添加一个新的区域"
)
async def fastapi_add_region(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""添加新的区域。"""
props = await req.json()
return add_region(network, ChangeSet(props))
@router.post("/deleteregion/", response_model=None)
async def fastapi_delete_region(network: str, req: Request) -> ChangeSet:
@router.post(
"/deleteregion/",
response_model=None,
summary="删除区域",
description="删除指定的区域"
)
async def fastapi_delete_region(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""删除区域。"""
props = await req.json()
return delete_region(network, ChangeSet(props))
@router.get("/getallregions/")
async def fastapi_get_all_regions(network: str) -> list[dict[str, Any]]:
@router.get(
"/getallregions/",
summary="获取所有区域",
description="获取指定水网中的所有区域信息"
)
async def fastapi_get_all_regions(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""获取所有区域的信息列表。"""
return get_all_regions(network)
@router.post("/generateregion/", response_model=None)
@router.post(
"/generateregion/",
response_model=None,
summary="生成区域分区",
description="根据参数自动生成水网的区域分区"
)
async def fastapi_generate_region(
network: str, inflate_delta: float
network: str = Query(..., description="管网名称(或数据库名称)"),
inflate_delta: float = Query(..., description="膨胀参数")
) -> ChangeSet:
"""生成区域分区。"""
return generate_region(network, inflate_delta)
@@ -86,10 +154,25 @@ async def fastapi_generate_region(
# district_metering_area 33
############################################################
@router.get("/calculatedistrictmeteringarea/")
@router.get(
"/calculatedistrictmeteringarea/",
summary="计算DMA分区",
description="计算指定节点集的区域计量(DMA)分区方案"
)
async def fastapi_calculate_district_metering_area(
network: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> list[list[str]]:
"""
计算DMA分区。
请求体格式:
{
"nodes": 节点ID列表(list[str]),
"part_count": 分区数量(int),
"part_type": 分区类型(int)
}
"""
props = await req.json()
nodes = props["nodes"]
part_count = props["part_count"]
@@ -98,10 +181,25 @@ async def fastapi_calculate_district_metering_area(
network, nodes, part_count, part_type
)
@router.get("/calculatedistrictmeteringareaforregion/")
@router.get(
"/calculatedistrictmeteringareaforregion/",
summary="计算区域内DMA分区",
description="为指定区域计算区域计量(DMA)分区方案"
)
async def fastapi_calculate_district_metering_area_for_region(
network: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> list[list[str]]:
"""
计算区域内DMA分区。
请求体格式:
{
"region": 区域ID(str),
"part_count": 分区数量(int),
"part_type": 分区类型(int)
}
"""
props = await req.json()
region = props["region"]
part_count = props["part_count"]
@@ -110,32 +208,77 @@ async def fastapi_calculate_district_metering_area_for_region(
network, region, part_count, part_type
)
@router.get("/calculatedistrictmeteringareafornetwork/")
@router.get(
"/calculatedistrictmeteringareafornetwork/",
summary="计算整网DMA分区",
description="为整个水网计算区域计量(DMA)分区方案"
)
async def fastapi_calculate_district_metering_area_for_network(
network: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> list[list[str]]:
"""
计算整网DMA分区。
请求体格式:
{
"part_count": 分区数量(int),
"part_type": 分区类型(int)
}
"""
props = await req.json()
part_count = props["part_count"]
part_type = props["part_type"]
return calculate_district_metering_area_for_network(network, part_count, part_type)
@router.get("/getdistrictmeteringareaschema/")
@router.get(
"/getdistrictmeteringareaschema/",
summary="获取DMA属性架构",
description="获取指定水网的区域计量(DMA)属性架构定义"
)
async def fastapi_get_district_metering_area_schema(
network: str,
network: str = Query(..., description="管网名称(或数据库名称)"),
) -> dict[str, dict[str, Any]]:
"""获取DMA的属性架构。"""
return get_district_metering_area_schema(network)
@router.get("/getdistrictmeteringarea/")
async def fastapi_get_district_metering_area(network: str, id: str) -> dict[str, Any]:
@router.get(
"/getdistrictmeteringarea/",
summary="获取DMA信息",
description="获取指定ID的区域计量(DMA)详细信息"
)
async def fastapi_get_district_metering_area(
network: str = Query(..., description="管网名称(或数据库名称)"),
id: str = Query(..., description="DMA ID")
) -> dict[str, Any]:
"""获取DMA的详细信息。"""
return get_district_metering_area(network, id)
@router.post("/setdistrictmeteringarea/", response_model=None)
async def fastapi_set_district_metering_area(network: str, req: Request) -> ChangeSet:
@router.post(
"/setdistrictmeteringarea/",
response_model=None,
summary="设置DMA属性",
description="修改指定DMA的属性信息"
)
async def fastapi_set_district_metering_area(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""设置DMA属性。"""
props = await req.json()
return set_district_metering_area(network, ChangeSet(props))
@router.post("/adddistrictmeteringarea/", response_model=None)
async def fastapi_add_district_metering_area(network: str, req: Request) -> ChangeSet:
@router.post(
"/adddistrictmeteringarea/",
response_model=None,
summary="添加新DMA",
description="向水网添加一个新的区域计量(DMA)"
)
async def fastapi_add_district_metering_area(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""添加新的DMA。"""
props = await req.json()
# boundary should be [(x,y), (x,y)]
boundary = props.get("boundary", [])
@@ -146,33 +289,73 @@ async def fastapi_add_district_metering_area(network: str, req: Request) -> Chan
props["boundary"] = newBoundary
return add_district_metering_area(network, ChangeSet(props))
@router.post("/deletedistrictmeteringarea/", response_model=None)
@router.post(
"/deletedistrictmeteringarea/",
response_model=None,
summary="删除DMA",
description="删除指定的区域计量(DMA)"
)
async def fastapi_delete_district_metering_area(
network: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""删除DMA。"""
props = await req.json()
return delete_district_metering_area(network, ChangeSet(props))
@router.get("/getalldistrictmeteringareaids/")
async def fastapi_get_all_district_metering_area_ids(network: str) -> list[str]:
@router.get(
"/getalldistrictmeteringareaids/",
summary="获取所有DMA ID",
description="获取指定水网中所有DMA的ID列表"
)
async def fastapi_get_all_district_metering_area_ids(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[str]:
"""获取所有DMA的ID列表。"""
return get_all_district_metering_area_ids(network)
@router.get("/getalldistrictmeteringareas/")
async def getalldistrictmeteringareas(network: str) -> list[dict[str, Any]]:
@router.get(
"/getalldistrictmeteringareas/",
summary="获取所有DMA",
description="获取指定水网中所有DMA的详细信息"
)
async def getalldistrictmeteringareas(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""获取所有DMA的详细信息列表。"""
return get_all_district_metering_areas(network)
@router.post("/generatedistrictmeteringarea/", response_model=None)
@router.post(
"/generatedistrictmeteringarea/",
response_model=None,
summary="生成DMA分区",
description="根据参数自动生成水网的DMA分区方案"
)
async def fastapi_generate_district_metering_area(
network: str, part_count: int, part_type: int, inflate_delta: float
network: str = Query(..., description="管网名称(或数据库名称)"),
part_count: int = Query(..., description="分区数量", gt=0),
part_type: int = Query(..., description="分区类型"),
inflate_delta: float = Query(..., description="膨胀参数")
) -> ChangeSet:
"""生成DMA分区。"""
return generate_district_metering_area(
network, part_count, part_type, inflate_delta
)
@router.post("/generatesubdistrictmeteringarea/", response_model=None)
@router.post(
"/generatesubdistrictmeteringarea/",
response_model=None,
summary="生成DMA子分区",
description="为指定DMA生成子DMA分区"
)
async def fastapi_generate_sub_district_metering_area(
network: str, dma: str, part_count: int, part_type: int, inflate_delta: float
network: str = Query(..., description="管网名称(或数据库名称)"),
dma: str = Query(..., description="DMA ID"),
part_count: int = Query(..., description="分区数量", gt=0),
part_type: int = Query(..., description="分区类型"),
inflate_delta: float = Query(..., description="膨胀参数")
) -> ChangeSet:
"""生成DMA子分区。"""
return generate_sub_district_metering_area(
network, dma, part_count, part_type, inflate_delta
)
@@ -182,43 +365,105 @@ async def fastapi_generate_sub_district_metering_area(
# service_area 34
############################################################
@router.get("/calculateservicearea/")
@router.get(
"/calculateservicearea/",
summary="计算服务区",
description="计算指定水网在指定时间步长的服务区分区"
)
async def fastapi_calculate_service_area(
network: str, time_index: int
network: str = Query(..., description="管网名称(或数据库名称)"),
time_index: int = Query(..., description="时间步长索引", ge=0)
) -> dict[str, Any]:
"""计算服务区分区。"""
return calculate_service_area(network, time_index)
@router.get("/getserviceareaschema/")
async def fastapi_get_service_area_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get(
"/getserviceareaschema/",
summary="获取服务区属性架构",
description="获取指定水网的服务区属性架构定义"
)
async def fastapi_get_service_area_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""获取服务区的属性架构。"""
return get_service_area_schema(network)
@router.get("/getservicearea/")
async def fastapi_get_service_area(network: str, id: str) -> dict[str, Any]:
@router.get(
"/getservicearea/",
summary="获取服务区信息",
description="获取指定ID的服务区详细信息"
)
async def fastapi_get_service_area(
network: str = Query(..., description="管网名称(或数据库名称)"),
id: str = Query(..., description="服务区ID")
) -> dict[str, Any]:
"""获取服务区的详细信息。"""
return get_service_area(network, id)
@router.post("/setservicearea/", response_model=None)
async def fastapi_set_service_area(network: str, req: Request) -> ChangeSet:
@router.post(
"/setservicearea/",
response_model=None,
summary="设置服务区属性",
description="修改指定服务区的属性信息"
)
async def fastapi_set_service_area(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""设置服务区属性。"""
props = await req.json()
return set_service_area(network, ChangeSet(props))
@router.post("/addservicearea/", response_model=None)
async def fastapi_add_service_area(network: str, req: Request) -> ChangeSet:
@router.post(
"/addservicearea/",
response_model=None,
summary="添加新服务区",
description="向水网添加一个新的服务区"
)
async def fastapi_add_service_area(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""添加新的服务区。"""
props = await req.json()
return add_service_area(network, ChangeSet(props))
@router.post("/deleteservicearea/", response_model=None)
async def fastapi_delete_service_area(network: str, req: Request) -> ChangeSet:
@router.post(
"/deleteservicearea/",
response_model=None,
summary="删除服务区",
description="删除指定的服务区"
)
async def fastapi_delete_service_area(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""删除服务区。"""
props = await req.json()
return delete_service_area(network, ChangeSet(props))
@router.get("/getallserviceareas/")
async def fastapi_get_all_service_areas(network: str) -> list[dict[str, Any]]:
@router.get(
"/getallserviceareas/",
summary="获取所有服务区",
description="获取指定水网中的所有服务区信息"
)
async def fastapi_get_all_service_areas(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""获取所有服务区的信息列表。"""
return get_all_service_areas(network)
@router.post("/generateservicearea/", response_model=None)
@router.post(
"/generateservicearea/",
response_model=None,
summary="生成服务区分区",
description="根据参数自动生成水网的服务区分区"
)
async def fastapi_generate_service_area(
network: str, inflate_delta: float
network: str = Query(..., description="管网名称(或数据库名称)"),
inflate_delta: float = Query(..., description="膨胀参数")
) -> ChangeSet:
"""生成服务区分区。"""
return generate_service_area(network, inflate_delta)
@@ -226,52 +471,128 @@ async def fastapi_generate_service_area(
# virtual_district 35
############################################################
@router.get("/calculatevirtualdistrict/")
@router.get(
"/calculatevirtualdistrict/",
summary="计算虚拟分区",
description="根据指定的中心节点计算虚拟分区方案"
)
async def fastapi_calculate_virtual_district(
network: str, centers: list[str]
network: str = Query(..., description="管网名称(或数据库名称)"),
centers: list[str] = Query(..., description="中心节点ID列表")
) -> dict[str, list[Any]]:
"""计算虚拟分区。"""
return calculate_virtual_district(network, centers)
@router.get("/getvirtualdistrictschema/")
@router.get(
"/getvirtualdistrictschema/",
summary="获取虚拟分区属性架构",
description="获取指定水网的虚拟分区属性架构定义"
)
async def fastapi_get_virtual_district_schema(
network: str,
network: str = Query(..., description="管网名称(或数据库名称)"),
) -> dict[str, dict[str, Any]]:
"""获取虚拟分区的属性架构。"""
return get_virtual_district_schema(network)
@router.get("/getvirtualdistrict/")
async def fastapi_get_virtual_district(network: str, id: str) -> dict[str, Any]:
@router.get(
"/getvirtualdistrict/",
summary="获取虚拟分区信息",
description="获取指定ID的虚拟分区详细信息"
)
async def fastapi_get_virtual_district(
network: str = Query(..., description="管网名称(或数据库名称)"),
id: str = Query(..., description="虚拟分区ID")
) -> dict[str, Any]:
"""获取虚拟分区的详细信息。"""
return get_virtual_district(network, id)
@router.post("/setvirtualdistrict/", response_model=None)
async def fastapi_set_virtual_district(network: str, req: Request) -> ChangeSet:
@router.post(
"/setvirtualdistrict/",
response_model=None,
summary="设置虚拟分区属性",
description="修改指定虚拟分区的属性信息"
)
async def fastapi_set_virtual_district(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""设置虚拟分区属性。"""
props = await req.json()
return set_virtual_district(network, ChangeSet(props))
@router.post("/addvirtualdistrict/", response_model=None)
async def fastapi_add_virtual_district(network: str, req: Request) -> ChangeSet:
@router.post(
"/addvirtualdistrict/",
response_model=None,
summary="添加新虚拟分区",
description="向水网添加一个新的虚拟分区"
)
async def fastapi_add_virtual_district(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""添加新的虚拟分区。"""
props = await req.json()
return add_virtual_district(network, ChangeSet(props))
@router.post("/deletevirtualdistrict/", response_model=None)
async def fastapi_delete_virtual_district(network: str, req: Request) -> ChangeSet:
@router.post(
"/deletevirtualdistrict/",
response_model=None,
summary="删除虚拟分区",
description="删除指定的虚拟分区"
)
async def fastapi_delete_virtual_district(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""删除虚拟分区。"""
props = await req.json()
return delete_virtual_district(network, ChangeSet(props))
@router.get("/getallvirtualdistrict/")
async def fastapi_get_all_virtual_district(network: str) -> list[dict[str, Any]]:
@router.get(
"/getallvirtualdistrict/",
summary="获取所有虚拟分区",
description="获取指定水网中的所有虚拟分区信息"
)
async def fastapi_get_all_virtual_district(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""获取所有虚拟分区的信息列表。"""
return get_all_virtual_districts(network)
@router.post("/generatevirtualdistrict/", response_model=None)
@router.post(
"/generatevirtualdistrict/",
response_model=None,
summary="生成虚拟分区",
description="根据参数自动生成虚拟分区方案"
)
async def fastapi_generate_virtual_district(
network: str, inflate_delta: float, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
inflate_delta: float = Query(..., description="膨胀参数"),
req: Request = Body(...)
) -> ChangeSet:
"""生成虚拟分区。"""
props = await req.json()
return generate_virtual_district(network, props["centers"], inflate_delta)
@router.get("/calculatedistrictmeteringareafornodes/")
@router.get(
"/calculatedistrictmeteringareafornodes/",
summary="计算节点DMA分区",
description="为指定节点集计算区域计量(DMA)分区方案"
)
async def fastapi_calculate_district_metering_area_for_nodes(
network: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> list[list[str]]:
"""
计算节点DMA分区。
请求体格式:
{
"nodes": 节点ID列表(list[str]),
"part_count": 分区数量(int),
"part_type": 分区类型(int)
}
"""
props = await req.json()
nodes = props["nodes"]
part_count = props["part_count"]
+343 -35
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -13,102 +13,410 @@ from app.services.tjnetwork import (
router = APIRouter()
@router.get("/getreservoirschema")
async def fast_get_reservoir_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get(
"/getreservoirschema",
summary="获取水库模式",
description="获取指定供水网络中所有水库的模式/属性字段定义"
)
async def fast_get_reservoir_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""
获取水库模式定义。
该端点返回指定网络中水库对象的模式定义,包括所有可用的属性字段。
Args:
network: 管网名称(或数据库名称)
Returns:
水库属性的模式定义字典
"""
return get_reservoir_schema(network)
@router.post("/addreservoir/", response_model=None)
@router.post(
"/addreservoir/",
response_model=None,
summary="添加水库",
description="在指定供水网络中添加新的水库/水源节点"
)
async def fastapi_add_reservoir(
network: str, reservoir: str, x: float, y: float, head: float
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符"),
x: float = Query(..., description="水库的X坐标"),
y: float = Query(..., description="水库的Y坐标"),
head: float = Query(..., description="水库的水头/总水头(米)")
) -> ChangeSet:
"""
添加新的水库/水源节点。
在指定的供水网络中创建一个新的水库,并设置其坐标和水头参数。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
x: 水库的X坐标位置
y: 水库的Y坐标位置
head: 水库的供水水头(以米为单位)
Returns:
包含操作变更集的ChangeSet对象
"""
ps = {"id": reservoir, "x": x, "y": y, "head": head}
return add_reservoir(network, ChangeSet(ps))
@router.post("/deletereservoir/", response_model=None)
async def fastapi_delete_reservoir(network: str, reservoir: str) -> ChangeSet:
@router.post(
"/deletereservoir/",
response_model=None,
summary="删除水库",
description="从指定供水网络中删除指定的水库/水源节点"
)
async def fastapi_delete_reservoir(
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="要删除的水库的唯一标识符")
) -> ChangeSet:
"""
删除指定的水库节点。
从指定的供水网络中删除一个水库及其相关的所有连接关系。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
Returns:
包含操作变更集的ChangeSet对象
"""
ps = {"id": reservoir}
return delete_reservoir(network, ChangeSet(ps))
@router.get("/getreservoirhead/")
async def fastapi_get_reservoir_head(network: str, reservoir: str) -> float | None:
@router.get(
"/getreservoirhead/",
summary="获取水库水头",
description="获取指定水库的供水水头/总水头值"
)
async def fastapi_get_reservoir_head(
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符")
) -> float | None:
"""
获取水库的水头参数。
返回指定水库的供水水头(总水头),单位为米。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
Returns:
水库的水头值(米),如果水库不存在则返回None
"""
ps = get_reservoir(network, reservoir)
return ps["head"]
@router.get("/getreservoirpattern/")
async def fastapi_get_reservoir_pattern(network: str, reservoir: str) -> str | None:
@router.get(
"/getreservoirpattern/",
summary="获取水库模式",
description="获取指定水库的运行模式/供水模式"
)
async def fastapi_get_reservoir_pattern(
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符")
) -> str | None:
"""
获取水库的运行模式。
返回指定水库的供水模式,如固定水头模式、时间序列模式等。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
Returns:
水库的运行模式字符串,如果水库不存在则返回None
"""
ps = get_reservoir(network, reservoir)
return ps["pattern"]
@router.get("/getreservoirx/")
@router.get(
"/getreservoirx/",
summary="获取水库X坐标",
description="获取指定水库的X坐标位置"
)
async def fastapi_get_reservoir_x(
network: str, reservoir: str
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符")
) -> dict[str, float] | None:
"""
获取水库的X坐标。
返回指定水库的X轴坐标值。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
Returns:
水库的X坐标值,如果水库不存在则返回None
"""
ps = get_reservoir(network, reservoir)
return ps["x"]
@router.get("/getreservoiry/")
@router.get(
"/getreservoiry/",
summary="获取水库Y坐标",
description="获取指定水库的Y坐标位置"
)
async def fastapi_get_reservoir_y(
network: str, reservoir: str
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符")
) -> dict[str, float] | None:
"""
获取水库的Y坐标。
返回指定水库的Y轴坐标值。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
Returns:
水库的Y坐标值,如果水库不存在则返回None
"""
ps = get_reservoir(network, reservoir)
return ps["y"]
@router.get("/getreservoircoord/")
@router.get(
"/getreservoircoord/",
summary="获取水库坐标",
description="获取指定水库的平面坐标(X和Y坐标)"
)
async def fastapi_get_reservoir_coord(
network: str, reservoir: str
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符")
) -> dict[str, float] | None:
"""
获取水库的坐标。
返回指定水库的平面坐标,包含水库ID、X坐标和Y坐标。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
Returns:
包含water库ID和X、Y坐标的字典,如果水库不存在则返回None
"""
ps = get_reservoir(network, reservoir)
coord = {"id": reservoir, "x": ps["x"], "y": ps["y"]}
return coord
@router.post("/setreservoirhead/", response_model=None)
@router.post(
"/setreservoirhead/",
response_model=None,
summary="设置水库水头",
description="更新指定水库的供水水头/总水头值"
)
async def fastapi_set_reservoir_head(
network: str, reservoir: str, head: float
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符"),
head: float = Query(..., description="新的水头值(米)")
) -> ChangeSet:
"""
设置水库的水头参数。
更新指定水库的供水水头(总水头)值。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
head: 新的水头值(以米为单位)
Returns:
包含操作变更集的ChangeSet对象
"""
ps = {"id": reservoir, "head": head}
return set_reservoir(network, ChangeSet(ps))
@router.post("/setreservoirpattern/", response_model=None)
@router.post(
"/setreservoirpattern/",
response_model=None,
summary="设置水库模式",
description="更新指定水库的运行模式/供水模式"
)
async def fastapi_set_reservoir_pattern(
network: str, reservoir: str, pattern: str
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符"),
pattern: str = Query(..., description="新的运行模式")
) -> ChangeSet:
"""
设置水库的运行模式。
更新指定水库的供水模式,如固定水头模式、时间序列模式等。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
pattern: 新的运行模式字符串
Returns:
包含操作变更集的ChangeSet对象
"""
ps = {"id": reservoir, "pattern": pattern}
return set_reservoir(network, ChangeSet(ps))
@router.post("/setreservoirx/", response_model=None)
async def fastapi_set_reservoir_x(network: str, reservoir: str, x: float) -> ChangeSet:
@router.post(
"/setreservoirx/",
response_model=None,
summary="设置水库X坐标",
description="更新指定水库的X坐标位置"
)
async def fastapi_set_reservoir_x(
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符"),
x: float = Query(..., description="新的X坐标值")
) -> ChangeSet:
"""
设置水库的X坐标。
更新指定水库的X轴坐标值。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
x: 新的X坐标值
Returns:
包含操作变更集的ChangeSet对象
"""
ps = {"id": reservoir, "x": x}
return set_reservoir(network, ChangeSet(ps))
@router.post("/setreservoiry/", response_model=None)
async def fastapi_set_reservoir_y(network: str, reservoir: str, y: float) -> ChangeSet:
@router.post(
"/setreservoiry/",
response_model=None,
summary="设置水库Y坐标",
description="更新指定水库的Y坐标位置"
)
async def fastapi_set_reservoir_y(
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符"),
y: float = Query(..., description="新的Y坐标值")
) -> ChangeSet:
"""
设置水库的Y坐标。
更新指定水库的Y轴坐标值。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
y: 新的Y坐标值
Returns:
包含操作变更集的ChangeSet对象
"""
ps = {"id": reservoir, "y": y}
return set_reservoir(network, ChangeSet(ps))
@router.post("/setreservoircoord/", response_model=None)
@router.post(
"/setreservoircoord/",
response_model=None,
summary="设置水库坐标",
description="更新指定水库的平面坐标(X和Y坐标)"
)
async def fastapi_set_reservoir_coord(
network: str, reservoir: str, x: float, y: float
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符"),
x: float = Query(..., description="新的X坐标值"),
y: float = Query(..., description="新的Y坐标值")
) -> ChangeSet:
"""
设置水库的坐标。
更新指定水库的平面坐标,包括X和Y坐标。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
x: 新的X坐标值
y: 新的Y坐标值
Returns:
包含操作变更集的ChangeSet对象
"""
ps = {"id": reservoir, "x": x, "y": y}
return set_reservoir(network, ChangeSet(ps))
@router.get("/getreservoirproperties/")
@router.get(
"/getreservoirproperties/",
summary="获取水库属性",
description="获取指定水库的所有属性"
)
async def fastapi_get_reservoir_properties(
network: str, reservoir: str
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符")
) -> dict[str, Any]:
"""
获取水库的所有属性。
返回指定水库的完整属性信息,包括ID、坐标、水头、模式等所有属性。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
Returns:
包含水库所有属性的字典
"""
return get_reservoir(network, reservoir)
@router.get("/getallreservoirproperties/")
async def fastapi_get_all_reservoir_properties(network: str) -> list[dict[str, Any]]:
# 缓存查询结果提高性能
# global redis_client
@router.get(
"/getallreservoirproperties/",
summary="获取所有水库属性",
description="获取指定供水网络中所有水库的属性"
)
async def fastapi_get_all_reservoir_properties(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""
获取所有水库的属性。
返回指定供水网络中所有水库的完整属性信息列表。
Args:
network: 管网名称(或数据库名称)
Returns:
包含所有水库属性的字典列表
"""
results = get_all_reservoirs(network)
return results
@router.post("/setreservoirproperties/", response_model=None)
@router.post(
"/setreservoirproperties/",
response_model=None,
summary="设置水库属性",
description="批量更新指定水库的多个属性"
)
async def fastapi_set_reservoir_properties(
network: str, reservoir: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
reservoir: str = Query(..., description="水库的唯一标识符"),
req: Request = Body(..., description="包含要更新的属性的请求体")
) -> ChangeSet:
"""
设置水库的多个属性。
批量更新指定水库的属性。属性通过JSON请求体传递。
Args:
network: 管网名称(或数据库名称)
reservoir: 水库的唯一标识符
req: HTTP请求对象,包含JSON格式的属性数据
Returns:
包含操作变更集的ChangeSet对象
"""
props = await req.json()
ps = {"id": reservoir} | props
return set_reservoir(network, ChangeSet(ps))
+41 -9
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -15,20 +15,52 @@ router = APIRouter()
# tag 8.[TAGS]
############################################################
@router.get("/gettagschema/")
async def fastapi_get_tag_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get(
"/gettagschema/",
summary="获取标签属性架构",
description="获取指定水网的标签(Tag)属性架构定义"
)
async def fastapi_get_tag_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""获取标签的属性架构。"""
return get_tag_schema(network)
@router.get("/gettag/")
async def fastapi_get_tag(network: str, t_type: str, id: str) -> dict[str, Any]:
@router.get(
"/gettag/",
summary="获取标签信息",
description="获取指定类型和ID的标签信息"
)
async def fastapi_get_tag(
network: str = Query(..., description="管网名称(或数据库名称)"),
t_type: str = Query(..., description="标签类型"),
id: str = Query(..., description="元素ID")
) -> dict[str, Any]:
"""获取标签信息。"""
return get_tag(network, t_type, id)
@router.get("/gettags/")
async def fastapi_get_tags(network: str) -> list[dict[str, Any]]:
@router.get(
"/gettags/",
summary="获取所有标签",
description="获取指定水网中的所有标签信息"
)
async def fastapi_get_tags(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""获取水网中所有标签的列表。"""
tags = get_tags(network)
return tags
@router.post("/settag/", response_model=None)
async def fastapi_set_tag(network: str, req: Request) -> ChangeSet:
@router.post(
"/settag/",
response_model=None,
summary="设置标签",
description="为指定元素设置或修改标签信息"
)
async def fastapi_set_tag(
network: str = Query(..., description="管网名称(或数据库名称)"),
req: Request = Body(...)
) -> ChangeSet:
"""设置标签信息。"""
props = await req.json()
return set_tag(network, ChangeSet(props))
+435 -66
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -13,23 +13,50 @@ from app.services.tjnetwork import (
router = APIRouter()
@router.get("/gettankschema")
async def fast_get_tank_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get("/gettankschema", summary="获取水箱模式", description="获取指定网络的水箱数据结构模式定义")
async def fast_get_tank_schema(network: str = Query(..., description="管网名称(或数据库名称)")) -> dict[str, dict[str, Any]]:
"""
获取水箱的数据结构模式。
Args:
network: 管网名称(或数据库名称)
Returns:
包含水箱属性的模式定义字典
"""
return get_tank_schema(network)
@router.post("/addtank/", response_model=None)
@router.post("/addtank/", summary="新增水箱", description="向指定网络中新增一个水箱", response_model=None)
async def fastapi_add_tank(
network: str,
tank: str,
x: float,
y: float,
elevation: float,
init_level: float = 0,
min_level: float = 0,
max_level: float = 0,
diameter: float = 0,
min_vol: float = 0,
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
x: float = Query(..., description="X坐标"),
y: float = Query(..., description="Y坐标"),
elevation: float = Query(..., description="标高"),
init_level: float = Query(0, description="初始水位"),
min_level: float = Query(0, description="最小水位"),
max_level: float = Query(0, description="最大水位"),
diameter: float = Query(0, description="直径"),
min_vol: float = Query(0, description="最小体积"),
) -> ChangeSet:
"""
创建新水箱。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
x: X坐标
y: Y坐标
elevation: 水箱标高
init_level: 初始水位,默认为0
min_level: 最小水位,默认为0
max_level: 最大水位,默认为0
diameter: 水箱直径,默认为0
min_vol: 最小体积,默认为0
Returns:
包含变更信息的ChangeSet对象
"""
ps = {
"id": tank,
"x": x,
@@ -43,155 +70,497 @@ async def fastapi_add_tank(
}
return add_tank(network, ChangeSet(ps))
@router.post("/deletetank/", response_model=None)
async def fastapi_delete_tank(network: str, tank: str) -> ChangeSet:
@router.post("/deletetank/", summary="删除水箱", description="删除指定网络中的水箱", response_model=None)
async def fastapi_delete_tank(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> ChangeSet:
"""
删除指定的水箱。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank}
return delete_tank(network, ChangeSet(ps))
@router.get("/gettankelevation/")
async def fastapi_get_tank_elevation(network: str, tank: str) -> float | None:
@router.get("/gettankelevation/", summary="获取水箱标高", description="获取指定水箱的标高值")
async def fastapi_get_tank_elevation(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> float | None:
"""
获取水箱的标高。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱标高值,如果不存在则返回None
"""
ps = get_tank(network, tank)
return ps["elevation"]
@router.get("/gettankinitlevel/")
async def fastapi_get_tank_init_level(network: str, tank: str) -> float | None:
@router.get("/gettankinitlevel/", summary="获取水箱初始水位", description="获取指定水箱的初始水位值")
async def fastapi_get_tank_init_level(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> float | None:
"""
获取水箱的初始水位。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱初始水位值,如果不存在则返回None
"""
ps = get_tank(network, tank)
return ps["init_level"]
@router.get("/gettankminlevel/")
async def fastapi_get_tank_min_level(network: str, tank: str) -> float | None:
@router.get("/gettankminlevel/", summary="获取水箱最小水位", description="获取指定水箱的最小水位值")
async def fastapi_get_tank_min_level(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> float | None:
"""
获取水箱的最小水位。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱最小水位值,如果不存在则返回None
"""
ps = get_tank(network, tank)
return ps["min_level"]
@router.get("/gettankmaxlevel/")
async def fastapi_get_tank_max_level(network: str, tank: str) -> float | None:
@router.get("/gettankmaxlevel/", summary="获取水箱最大水位", description="获取指定水箱的最大水位值")
async def fastapi_get_tank_max_level(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> float | None:
"""
获取水箱的最大水位。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱最大水位值,如果不存在则返回None
"""
ps = get_tank(network, tank)
return ps["max_level"]
@router.get("/gettankdiameter/")
async def fastapi_get_tank_diameter(network: str, tank: str) -> float | None:
@router.get("/gettankdiameter/", summary="获取水箱直径", description="获取指定水箱的直径值")
async def fastapi_get_tank_diameter(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> float | None:
"""
获取水箱的直径。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱直径值,如果不存在则返回None
"""
ps = get_tank(network, tank)
return ps["diameter"]
@router.get("/gettankminvol/")
async def fastapi_get_tank_min_vol(network: str, tank: str) -> float | None:
@router.get("/gettankminvol/", summary="获取水箱最小体积", description="获取指定水箱的最小体积值")
async def fastapi_get_tank_min_vol(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> float | None:
"""
获取水箱的最小体积。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱最小体积值,如果不存在则返回None
"""
ps = get_tank(network, tank)
return ps["min_vol"]
@router.get("/gettankvolcurve/")
async def fastapi_get_tank_vol_curve(network: str, tank: str) -> str | None:
@router.get("/gettankvolcurve/", summary="获取水箱容积曲线", description="获取指定水箱的容积曲线标识")
async def fastapi_get_tank_vol_curve(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> str | None:
"""
获取水箱的容积曲线。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱容积曲线标识,如果不存在则返回None
"""
ps = get_tank(network, tank)
return ps["vol_curve"]
@router.get("/gettankoverflow/")
async def fastapi_get_tank_overflow(network: str, tank: str) -> str | None:
@router.get("/gettankoverflow/", summary="获取水箱溢流口", description="获取指定水箱的溢流口配置")
async def fastapi_get_tank_overflow(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> str | None:
"""
获取水箱的溢流口配置。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱溢流口配置,如果不存在则返回None
"""
ps = get_tank(network, tank)
return ps["overflow"]
@router.get("/gettankx/")
async def fastapi_get_tank_x(network: str, tank: str) -> float:
@router.get("/gettankx/", summary="获取水箱X坐标", description="获取指定水箱的X坐标值")
async def fastapi_get_tank_x(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> float:
"""
获取水箱的X坐标。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱X坐标值
"""
ps = get_tank(network, tank)
return ps["x"]
@router.get("/gettanky/")
async def fastapi_get_tank_y(network: str, tank: str) -> float:
@router.get("/gettanky/", summary="获取水箱Y坐标", description="获取指定水箱的Y坐标值")
async def fastapi_get_tank_y(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> float:
"""
获取水箱的Y坐标。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
水箱Y坐标值
"""
ps = get_tank(network, tank)
return ps["y"]
@router.get("/gettankcoord/")
async def fastapi_get_tank_coord(network: str, tank: str) -> dict[str, float]:
@router.get("/gettankcoord/", summary="获取水箱坐标", description="获取指定水箱的X和Y坐标")
async def fastapi_get_tank_coord(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> dict[str, float]:
"""
获取水箱的坐标。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
包含x和y坐标的字典
"""
ps = get_tank(network, tank)
coord = {"x": ps["x"], "y": ps["y"]}
return coord
@router.post("/settankelevation/", response_model=None)
@router.post("/settankelevation/", summary="设置水箱标高", description="设置指定水箱的标高值", response_model=None)
async def fastapi_set_tank_elevation(
network: str, tank: str, elevation: float
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
elevation: float = Query(..., description="新的标高值")
) -> ChangeSet:
"""
设置水箱的标高。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
elevation: 新的标高值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "elevation": elevation}
return set_tank(network, ChangeSet(ps))
@router.post("/settankinitlevel/", response_model=None)
@router.post("/settankinitlevel/", summary="设置水箱初始水位", description="设置指定水箱的初始水位值", response_model=None)
async def fastapi_set_tank_init_level(
network: str, tank: str, init_level: float
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
init_level: float = Query(..., description="新的初始水位值")
) -> ChangeSet:
"""
设置水箱的初始水位。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
init_level: 新的初始水位值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "init_level": init_level}
return set_tank(network, ChangeSet(ps))
@router.post("/settankminlevel/", response_model=None)
@router.post("/settankminlevel/", summary="设置水箱最小水位", description="设置指定水箱的最小水位值", response_model=None)
async def fastapi_set_tank_min_level(
network: str, tank: str, min_level: float
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
min_level: float = Query(..., description="新的最小水位值")
) -> ChangeSet:
"""
设置水箱的最小水位。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
min_level: 新的最小水位值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "min_level": min_level}
return set_tank(network, ChangeSet(ps))
@router.post("/settankmaxlevel/", response_model=None)
@router.post("/settankmaxlevel/", summary="设置水箱最大水位", description="设置指定水箱的最大水位值", response_model=None)
async def fastapi_set_tank_max_level(
network: str, tank: str, max_level: float
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
max_level: float = Query(..., description="新的最大水位值")
) -> ChangeSet:
"""
设置水箱的最大水位。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
max_level: 新的最大水位值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "max_level": max_level}
return set_tank(network, ChangeSet(ps))
@router.post("settankdiameter//", response_model=None)
@router.post("/settankdiameter/", summary="设置水箱直径", description="设置指定水箱的直径值", response_model=None)
async def fastapi_set_tank_diameter(
network: str, tank: str, diameter: float
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
diameter: float = Query(..., description="新的直径值")
) -> ChangeSet:
"""
设置水箱的直径。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
diameter: 新的直径值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "diameter": diameter}
return set_tank(network, ChangeSet(ps))
@router.post("/settankminvol/", response_model=None)
@router.post("/settankminvol/", summary="设置水箱最小体积", description="设置指定水箱的最小体积值", response_model=None)
async def fastapi_set_tank_min_vol(
network: str, tank: str, min_vol: float
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
min_vol: float = Query(..., description="新的最小体积值")
) -> ChangeSet:
"""
设置水箱的最小体积。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
min_vol: 新的最小体积值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "min_vol": min_vol}
return set_tank(network, ChangeSet(ps))
@router.post("/settankvolcurve/", response_model=None)
@router.post("/settankvolcurve/", summary="设置水箱容积曲线", description="设置指定水箱的容积曲线标识", response_model=None)
async def fastapi_set_tank_vol_curve(
network: str, tank: str, vol_curve: str
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
vol_curve: str = Query(..., description="新的容积曲线标识")
) -> ChangeSet:
"""
设置水箱的容积曲线。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
vol_curve: 新的容积曲线标识
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "vol_curve": vol_curve}
return set_tank(network, ChangeSet(ps))
@router.post("/settankoverflow/", response_model=None)
@router.post("/settankoverflow/", summary="设置水箱溢流口", description="设置指定水箱的溢流口配置", response_model=None)
async def fastapi_set_tank_overflow(
network: str, tank: str, overflow: str
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
overflow: str = Query(..., description="新的溢流口配置")
) -> ChangeSet:
"""
设置水箱的溢流口配置。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
overflow: 新的溢流口配置
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "overflow": overflow}
return set_tank(network, ChangeSet(ps))
@router.post("/settankx/", response_model=None)
async def fastapi_set_tank_x(network: str, tank: str, x: float) -> ChangeSet:
@router.post("/settankx/", summary="设置水箱X坐标", description="设置指定水箱的X坐标值", response_model=None)
async def fastapi_set_tank_x(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
x: float = Query(..., description="新的X坐标值")
) -> ChangeSet:
"""
设置水箱的X坐标。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
x: 新的X坐标值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "x": x}
return set_tank(network, ChangeSet(ps))
@router.post("/settanky/", response_model=None)
async def fastapi_set_tank_y(network: str, tank: str, y: float) -> ChangeSet:
@router.post("/settanky/", summary="设置水箱Y坐标", description="设置指定水箱的Y坐标值", response_model=None)
async def fastapi_set_tank_y(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
y: float = Query(..., description="新的Y坐标值")
) -> ChangeSet:
"""
设置水箱的Y坐标。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
y: 新的Y坐标值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "y": y}
return set_tank(network, ChangeSet(ps))
@router.post("/settankcoord/", response_model=None)
@router.post("/settankcoord/", summary="设置水箱坐标", description="设置指定水箱的X和Y坐标", response_model=None)
async def fastapi_set_tank_coord(
network: str, tank: str, x: float, y: float
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
x: float = Query(..., description="新的X坐标值"),
y: float = Query(..., description="新的Y坐标值")
) -> ChangeSet:
"""
设置水箱的坐标。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
x: 新的X坐标值
y: 新的Y坐标值
Returns:
包含变更信息的ChangeSet对象
"""
ps = {"id": tank, "x": x, "y": y}
return set_tank(network, ChangeSet(ps))
@router.get("/gettankproperties/")
async def fastapi_get_tank_properties(network: str, tank: str) -> dict[str, Any]:
@router.get("/gettankproperties/", summary="获取水箱属性", description="获取指定水箱的所有属性")
async def fastapi_get_tank_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID")
) -> dict[str, Any]:
"""
获取水箱的所有属性。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
Returns:
包含水箱所有属性的字典
"""
return get_tank(network, tank)
@router.get("/getalltankproperties/")
async def fastapi_get_all_tank_properties(network: str) -> list[dict[str, Any]]:
@router.get("/getalltankproperties/", summary="获取所有水箱属性", description="获取指定网络中所有水箱的属性")
async def fastapi_get_all_tank_properties(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""
获取网络中所有水箱的属性。
Args:
network: 管网名称(或数据库名称)
Returns:
包含所有水箱属性的字典列表
"""
# 缓存查询结果提高性能
# global redis_client
results = get_all_tanks(network)
return results
@router.post("/settankproperties/", response_model=None)
@router.post("/settankproperties/", summary="设置水箱属性", description="批量设置指定水箱的多个属性", response_model=None)
async def fastapi_set_tank_properties(
network: str, tank: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
tank: str = Query(..., description="水箱ID"),
req: Request = Body(..., description="包含要设置的属性的请求体")
) -> ChangeSet:
"""
批量设置水箱的属性。
Args:
network: 管网名称(或数据库名称)
tank: 水箱ID
req: 包含水箱属性的请求体(JSON格式)
Returns:
包含变更信息的ChangeSet对象
"""
props = await req.json()
ps = {"id": tank} | props
return set_tank(network, ChangeSet(ps))
+249 -42
View File
@@ -1,4 +1,4 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, Query, Path, Body
from typing import Any, List, Dict, Union
from app.services.tjnetwork import (
Any,
@@ -14,21 +14,42 @@ from app.services.tjnetwork import (
router = APIRouter()
@router.get("/getvalveschema")
async def fastapi_get_valve_schema(network: str) -> dict[str, dict[str, Any]]:
@router.get(
"/getvalveschema",
summary="获取阀门架构",
description="获取指定水网中所有阀门的架构和字段定义",
)
async def fastapi_get_valve_schema(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> dict[str, dict[str, Any]]:
"""
获取阀门架构。
返回指定水网中所有阀门类型的完整架构定义,包括字段名称、类型和默认值。
"""
return get_valve_schema(network)
@router.post("/addvalve/", response_model=None)
@router.post(
"/addvalve/",
response_model=None,
summary="添加阀门",
description="在指定的水网中添加新的阀门",
)
async def fastapi_add_valve(
network: str,
valve: str,
node1: str,
node2: str,
diameter: float = 0,
v_type: str = VALVES_TYPE_PRV,
setting: float = 0,
minor_loss: float = 0,
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
node1: str = Query(..., description="起点节点ID"),
node2: str = Query(..., description="终点节点ID"),
diameter: float = Query(0, description="阀门直径(mm"),
v_type: str = Query(VALVES_TYPE_PRV, description="阀门类型"),
setting: float = Query(0, description="阀门开度/设置值"),
minor_loss: float = Query(0, description="损失系数"),
) -> ChangeSet:
"""
添加新的阀门。
在指定的水网中创建一个新的阀门,设置其连接的两个节点、直径、类型、开度和损失系数。
"""
ps = {
"id": valve,
"node1": node1,
@@ -41,85 +62,271 @@ async def fastapi_add_valve(
return add_valve(network, ChangeSet(ps))
@router.post("/deletevalve/", response_model=None)
async def fastapi_delete_valve(network: str, valve: str) -> ChangeSet:
@router.post(
"/deletevalve/",
response_model=None,
summary="删除阀门",
description="从指定的水网中删除指定的阀门",
)
async def fastapi_delete_valve(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
) -> ChangeSet:
"""
删除阀门。
从指定的水网中删除指定ID的阀门。
"""
ps = {"id": valve}
return delete_valve(network, ChangeSet(ps))
@router.get("/getvalvenode1/")
async def fastapi_get_valve_node1(network: str, valve: str) -> str | None:
@router.get(
"/getvalvenode1/",
summary="获取阀门起点节点",
description="获取指定阀门连接的起点节点ID",
)
async def fastapi_get_valve_node1(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
) -> str | None:
"""
获取阀门的起点节点。
返回指定阀门连接的起点(第一个)节点的ID。
"""
ps = get_valve(network, valve)
return ps["node1"]
@router.get("/getvalvenode2/")
async def fastapi_get_valve_node2(network: str, valve: str) -> str | None:
@router.get(
"/getvalvenode2/",
summary="获取阀门终点节点",
description="获取指定阀门连接的终点节点ID",
)
async def fastapi_get_valve_node2(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
) -> str | None:
"""
获取阀门的终点节点。
返回指定阀门连接的终点(第二个)节点的ID。
"""
ps = get_valve(network, valve)
return ps["node2"]
@router.get("/getvalvediameter/")
async def fastapi_get_valve_diameter(network: str, valve: str) -> float | None:
@router.get(
"/getvalvediameter/",
summary="获取阀门直径",
description="获取指定阀门的直径",
)
async def fastapi_get_valve_diameter(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
) -> float | None:
"""
获取阀门的直径。
返回指定阀门的直径值(单位:mm)。
"""
ps = get_valve(network, valve)
return ps["diameter"]
@router.get("/getvalvetype/")
async def fastapi_get_valve_type(network: str, valve: str) -> str | None:
@router.get(
"/getvalvetype/",
summary="获取阀门类型",
description="获取指定阀门的类型",
)
async def fastapi_get_valve_type(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
) -> str | None:
"""
获取阀门的类型。
返回指定阀门的类型(例如:减压阀、调节阀等)。
"""
ps = get_valve(network, valve)
return ps["type"]
@router.get("/getvalvesetting/")
async def fastapi_get_valve_setting(network: str, valve: str) -> float | None:
@router.get(
"/getvalvesetting/",
summary="获取阀门开度",
description="获取指定阀门的开度/设置值",
)
async def fastapi_get_valve_setting(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
) -> float | None:
"""
获取阀门的开度。
返回指定阀门的开度/设置值。
"""
ps = get_valve(network, valve)
return ps["setting"]
@router.get("/getvalveminorloss/")
async def fastapi_get_valve_minor_loss(network: str, valve: str) -> float | None:
@router.get(
"/getvalveminorloss/",
summary="获取阀门损失系数",
description="获取指定阀门的损失系数",
)
async def fastapi_get_valve_minor_loss(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
) -> float | None:
"""
获取阀门的损失系数。
返回指定阀门的损失系数值,用于计算流体通过阀门的压力损失。
"""
ps = get_valve(network, valve)
return ps["minor_loss"]
@router.post("/setvalvenode1/", response_model=None)
async def fastapi_set_valve_node1(network: str, valve: str, node1: str) -> ChangeSet:
@router.post(
"/setvalvenode1/",
response_model=None,
summary="设置阀门起点节点",
description="设置指定阀门的起点节点",
)
async def fastapi_set_valve_node1(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
node1: str = Query(..., description="新的起点节点ID"),
) -> ChangeSet:
"""
设置阀门的起点节点。
更新指定阀门的起点节点连接。
"""
ps = {"id": valve, "node1": node1}
return set_valve(network, ChangeSet(ps))
@router.post("/setvalvenode2/", response_model=None)
async def fastapi_set_valve_node2(network: str, valve: str, node2: str) -> ChangeSet:
@router.post(
"/setvalvenode2/",
response_model=None,
summary="设置阀门终点节点",
description="设置指定阀门的终点节点",
)
async def fastapi_set_valve_node2(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
node2: str = Query(..., description="新的终点节点ID"),
) -> ChangeSet:
"""
设置阀门的终点节点。
更新指定阀门的终点节点连接。
"""
ps = {"id": valve, "node2": node2}
return set_valve(network, ChangeSet(ps))
@router.post("/setvalvenodediameter/", response_model=None)
@router.post(
"/setvalvenodediameter/",
response_model=None,
summary="设置阀门直径",
description="设置指定阀门的直径",
)
async def fastapi_set_valve_diameter(
network: str, valve: str, diameter: float
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
diameter: float = Query(..., description="新的直径值(mm"),
) -> ChangeSet:
"""
设置阀门的直径。
更新指定阀门的直径值。
"""
ps = {"id": valve, "diameter": diameter}
return set_valve(network, ChangeSet(ps))
@router.post("/setvalvetype/", response_model=None)
async def fastapi_set_valve_type(network: str, valve: str, type: str) -> ChangeSet:
@router.post(
"/setvalvetype/",
response_model=None,
summary="设置阀门类型",
description="设置指定阀门的类型",
)
async def fastapi_set_valve_type(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
type: str = Query(..., description="新的阀门类型"),
) -> ChangeSet:
"""
设置阀门的类型。
更新指定阀门的类型(例如:减压阀、调节阀等)。
"""
ps = {"id": valve, "type": type}
return set_valve(network, ChangeSet(ps))
@router.post("/setvalvesetting/", response_model=None)
@router.post(
"/setvalvesetting/",
response_model=None,
summary="设置阀门开度",
description="设置指定阀门的开度/设置值",
)
async def fastapi_set_valve_setting(
network: str, valve: str, setting: float
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
setting: float = Query(..., description="新的开度值"),
) -> ChangeSet:
"""
设置阀门的开度。
更新指定阀门的开度/设置值。
"""
ps = {"id": valve, "setting": setting}
return set_valve(network, ChangeSet(ps))
@router.get("/getvalveproperties/")
async def fastapi_get_valve_properties(network: str, valve: str) -> dict[str, Any]:
@router.get(
"/getvalveproperties/",
summary="获取阀门所有属性",
description="获取指定阀门的所有属性",
)
async def fastapi_get_valve_properties(
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
) -> dict[str, Any]:
"""
获取阀门的所有属性。
返回指定阀门的完整属性集合,包括ID、节点、直径、类型、开度和损失系数。
"""
return get_valve(network, valve)
@router.get("/getallvalveproperties/")
async def fastapi_get_all_valve_properties(network: str) -> list[dict[str, Any]]:
@router.get(
"/getallvalveproperties/",
summary="获取所有阀门属性",
description="获取指定水网中所有阀门的属性",
)
async def fastapi_get_all_valve_properties(
network: str = Query(..., description="管网名称(或数据库名称)")
) -> list[dict[str, Any]]:
"""
获取所有阀门的属性。
返回指定水网中所有阀门的完整属性列表。
"""
# 缓存查询结果提高性能
# global redis_client
results = get_all_valves(network)
return results
@router.post("/setvalveproperties/", response_model=None)
@router.post(
"/setvalveproperties/",
response_model=None,
summary="批量设置阀门属性",
description="批量设置指定阀门的多个属性",
)
async def fastapi_set_valve_properties(
network: str, valve: str, req: Request
network: str = Query(..., description="管网名称(或数据库名称)"),
valve: str = Query(..., description="阀门ID"),
req: Request = Body(..., description="请求体,包含要更新的属性"),
) -> ChangeSet:
"""
批量设置阀门的属性。
更新指定阀门的一个或多个属性,通过JSON请求体传递要更新的属性。
"""
props = await req.json()
ps = {"id": valve} | props
return set_valve(network, ChangeSet(ps))