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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
# -*- coding:utf-8 -*-
#
# Author:jing
# Date: 5/27/20
from pot_libs.common.components.electric import Electric
from pot_libs.logger import Logger, log
from pot_libs.mysql_util.mysql_util import MysqlUtil
from pot_libs.qingstor_util.qs_helper import QsHelper
from pot_libs.sanic_api import summary, description
from unify_api.constants import U_WHITE_LIST_SID, PRODUCT
from unify_api.modules.uassistant.components.assistant import DebugRequest, \
DebugResponse, DebugValue, DebugValueElectric, DebugValueAdio, AssiResponse
from unify_api.modules.uassistant.procedures.assistant import calc_result
from unify_api.modules.uassistant.procedures.check_meter_connection import \
MeterConnector, meter_result
from unify_api.modules.users.procedures.jwt_user import jwt_user
logger = Logger.get_logger(__name__)
@summary('u工助手')
@description('用于装置调试时查看数据,调整接线等。')
async def post_assistant(request, body: DebugRequest) -> DebugResponse:
# 1.获取参数
sid = body.sid
meter_sn = body.meter_sn
user_id = jwt_user(request)
conclusions = {}
# 2.白名单判断
# 不在白名单
if sid not in U_WHITE_LIST_SID:
# 3.超管判断
sql = "SELECT uassistant_auth from user_product_auth " \
"where product = %s and user_id = %s"
async with MysqlUtil() as conn:
auth_info = await conn.fetchone(sql=sql,
args=(PRODUCT["anDianU"], user_id))
if not auth_info:
return DebugResponse.user_error()
# {'uassistant_auth': 1}
uassistant_auth = auth_info.get("uassistant_auth")
# 不是超管权限,需要判断装置是否已安装
if not uassistant_auth:
sql_mid = "SELECT mtid FROM monitor WHERE demolished=0 and sid=%s"
async with MysqlUtil() as conn:
mid_info = await conn.fetchone(sql=sql_mid, args=(sid,))
# 装置已经存在
if mid_info:
return DebugResponse.sid_exits(sid)
# sql_mid = "SELECT mid from meter WHERE sid=%s " \
# "ORDER BY create_time DESC limit 1;"
# async with MysqlUtil() as conn:
# mid_info = await conn.fetchone(sql=sql_mid, args=(sid,))
# # 装置已安装逻辑:有mid且ptr、ctr有值
# if mid_info:
# sql_me = "SELECT ptr,ctr from meter_param_record " \
# "where mid = %s ORDER BY start_time DESC limit 1"
# async with MysqlUtil() as conn:
# record_info = await conn.fetchone(sql=sql_me, args=(
# mid_info.get("mid"),))
# # 装置已经存在
# if record_info:
# return DebugResponse.sid_exits(sid)
if sid == U_WHITE_LIST_SID[0]:
sid = "A1904000095"
if sid == U_WHITE_LIST_SID[1]:
sid = "A1911000294"
# 4. 业务后台处理
values = []
async with QsHelper() as helper:
# 获取electrics量数据
electrics = await helper.get_recent_electric(sid, meter_sn=meter_sn,
items=4)
# 获取adios量数据
adios = await helper.get_recent_adio(sid, meter_sn=meter_sn, items=4)
if not electrics:
logger.info(f'no electric data in qingstor of sid {sid}')
return DebugResponse.sid_data_missing(sid)
for i, (pd, data) in enumerate(electrics):
# electric = Electric.from_dict(data['data'])
if i == 0 and meter_sn not in ("A", "B", "C"):
# 图的结果,按照新的算法计算
# conclusions = calc_result(electric)
log.info(data["data"])
conclusions = meter_result(data["data"])
if len(adios) > i:
adio_pd, adio = adios[i]
adio_data = adio['data']
adio = {**adio_data['aiao'], **adio_data['dido']}
else:
adio_pd, adio = pd, {}
value = DebugValue(
electric=DebugValueElectric.from_dict(data['data']),
adio=DebugValueAdio(adio),
date_time=str(pd),
adio_date=str(adio_pd)
)
values.append(value)
resp = DebugResponse(values=values, conclusions=conclusions)
logger.info(f'response of assistant : {resp}, sid: {sid}')
return resp