421 lines
14 KiB
Python
421 lines
14 KiB
Python
from fastapi import APIRouter, Request, Query, Path, Body
|
|
from typing import Any, List, Dict, Union
|
|
from app.services.tjnetwork import (
|
|
Any,
|
|
ChangeSet,
|
|
delete_junction,
|
|
delete_pipe,
|
|
delete_pump,
|
|
delete_reservoir,
|
|
delete_tank,
|
|
delete_valve,
|
|
get_all_scada_info,
|
|
get_element_properties,
|
|
get_element_properties_with_type,
|
|
get_element_type,
|
|
get_element_type_value,
|
|
get_link_properties,
|
|
get_link_type,
|
|
get_links,
|
|
get_node_links,
|
|
get_node_properties,
|
|
get_node_type,
|
|
get_nodes,
|
|
get_scada_info,
|
|
get_status,
|
|
get_status_schema,
|
|
get_title,
|
|
get_title_schema,
|
|
is_junction,
|
|
is_link,
|
|
is_node,
|
|
is_pipe,
|
|
is_pump,
|
|
is_reservoir,
|
|
is_tank,
|
|
is_valve,
|
|
set_status,
|
|
set_title,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
############################################################
|
|
# type
|
|
############################################################
|
|
|
|
@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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
summary="获取所有节点",
|
|
description="获取指定水网中的所有节点ID列表"
|
|
)
|
|
async def fastapi_get_nodes(network: str = Query(..., description="管网名称（或数据库名称）")) -> list[str]:
|
|
"""获取水网中所有节点的ID列表。"""
|
|
return get_nodes(network)
|
|
|
|
@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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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/",
|
|
summary="获取指定类型元素属性",
|
|
description="获取指定类型的元素属性信息"
|
|
)
|
|
async def fast_get_element_properties_with_type(
|
|
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/",
|
|
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/",
|
|
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/",
|
|
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,
|
|
summary="设置水网标题属性",
|
|
description="设置指定水网的标题(Title)信息"
|
|
)
|
|
async def fastapi_set_title(
|
|
network: str = Query(..., description="管网名称（或数据库名称）"),
|
|
req: Request = None
|
|
) -> ChangeSet:
|
|
"""设置水网的标题属性。"""
|
|
props = await req.json()
|
|
return set_title(network, ChangeSet(props))
|
|
|
|
############################################################
|
|
# status 10.[STATUS]
|
|
############################################################
|
|
|
|
@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/",
|
|
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,
|
|
summary="设置管线状态",
|
|
description="设置指定管线的状态信息"
|
|
)
|
|
async def fastapi_set_status_properties(
|
|
network: str = Query(..., description="管网名称（或数据库名称）"),
|
|
link: str = Query(..., description="管线ID"),
|
|
req: Request = None
|
|
) -> ChangeSet:
|
|
"""设置管线的状态属性。"""
|
|
props = await req.json()
|
|
ps = {"link": link} | props
|
|
return set_status(network, ChangeSet(ps))
|
|
|
|
############################################################
|
|
# General Deletion
|
|
############################################################
|
|
|
|
@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))
|
|
elif is_reservoir(network, node):
|
|
return delete_reservoir(network, ChangeSet(ps))
|
|
elif is_tank(network, node):
|
|
return delete_tank(network, ChangeSet(ps))
|
|
return ChangeSet() # Should probably raise error or return empty
|
|
|
|
@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))
|
|
elif is_pump(network, link):
|
|
return delete_pump(network, ChangeSet(ps))
|
|
elif is_valve(network, link):
|
|
return delete_valve(network, ChangeSet(ps))
|
|
return ChangeSet()
|