1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
from pot_libs.logger import log
from pot_libs.sanic_api import summary
from pot_libs.mysql_util.mysql_util import MysqlUtil
from unify_api.modules.common.components.common_cps import CidReq
from unify_api.modules.common.procedures.points import point_to_mid
from unify_api.modules.common.service.list_point_service import \
list_storey_service, list_tsp_point_service, list_point_level_service, \
list_point_inline_service
from unify_api.modules.common.components.list_points_cps import (
ListPointRequest,
ListPointResponse,
CommonPoint,
CommonLocation,
Inline, LsRep, LtpRep, LplResp, LpiResp)
@summary('获取监测点,进线列表')
async def post_list_point(req, body: ListPointRequest) -> ListPointResponse:
cid = body.cid
list_point, inline_list, points = [], [], {}
sql = "SELECT pid,name,mtid,inlid,add_to_company FROM point where cid=%s"
async with MysqlUtil() as conn:
result = await conn.fetchall(sql, args=(cid,))
add_to_company = []
for res in result:
# mtids.append(res["mtid"])
inline_list.append({"inline_id": res["inlid"], "name": res["name"]})
points[res["mtid"]] = res
add_to_company.append(res["add_to_company"])
if points:
if len(points) == 1:
l_sql = f"SELECT mtid,lid,ad_field from location " \
f"where mtid = {list(points.keys())[0]}"
else:
l_sql = f"SELECT mtid,lid,ad_field from location " \
f"where mtid in {tuple(points.keys())}"
async with MysqlUtil() as conn:
datas = await conn.fetchall(l_sql)
for data in datas:
if data["mtid"] in points.keys():
temp_dict = {"temp1": "A相", "temp2": "B相", "temp3": "C相",
"temp4": "N相", "residual_current": "default"}
flag = temp_dict.get(data["ad_field"]) or ""
locations_dict = {
"item": flag,
"location_id": data.get("lid")
}
if "locations" not in points[data["mtid"]]:
points[data["mtid"]]["locations"] = [locations_dict]
else:
points[data["mtid"]]["locations"].append(locations_dict)
for k, v in points.items():
list_point.append({
"name": v["name"], "point_id": v["pid"],
"locations": v.get("locations") or [],
"add_to_company": v["add_to_company"]
})
return ListPointResponse(
points=list_point,
inlines=inline_list,
power_show_all=1 if any(add_to_company) else 0
)
@summary('获取楼层-识电u')
async def post_list_storey(req, body: ListPointRequest) -> LsRep:
cid = body.cid
return await list_storey_service(cid)
@summary('获取TSP监测点-扬尘')
async def post_list_tsp_point(req, body: ListPointRequest) -> LtpRep:
cid = body.cid
return await list_tsp_point_service(cid)
@summary('新版监测点-包含进线变压器层级')
async def post_list_point_level(req, body: CidReq) -> LplResp:
cid = body.cid
return await list_point_level_service(cid)
@summary('新版监测点-进线')
async def post_list_point_inline(req, body: CidReq) -> LpiResp:
cid = body.cid
return await list_point_inline_service(cid)