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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import pendulum
from pot_libs.logger import log
from unify_api.modules.common.dao.common_dao import (
get_point_monitor_dao, get_location_monitor_dao, sid_by_pid
)
from uuid import uuid4
async def get_point_soe_config(event_data, with_params=False):
point_id = event_data.get('point_id')
event_settings = {k: v for k, v in event_data.items() if k in [
'threshold', 'duration', 'enable'
]}
datas = await get_point_monitor_dao(point_id)
sid = datas.get("sid")
meter_sn = datas.get("meter_no")
config = {
'soe': {
"electric": {
meter_sn: {event_data['event_type']: event_settings}
}
}
}
if with_params:
keys_li = ["ctr", "ptr", "ctnum", "vc", "tc", "imax"]
params = {k: v for k, v in datas.items() if k in keys_li
and (v is not None)}
params['point_id'] = point_id
config['params'] = {'electric': {meter_sn: params}}
return sid, config
async def get_location_soe_config(event, with_params):
location_id = event.get('location_id')
event_settings = {k: v for k, v in event.items() if k in [
'threshold', 'duration', 'enable'
]}
datas = await get_location_monitor_dao(location_id)
sid = datas.get("sid")
field = datas.get("ad_field")
if field == "residual_current":
config = {'soe': {"adio": {event['event_type']: event_settings}}}
else:
config = {
'soe': {"adio": {field: {event['event_type']: event_settings}}}}
if with_params:
config['params'] = {'adio': {field: {'location_id': location_id}}}
return sid, config
async def get_point_scope_config(event_data, method='config'):
'''
获取录波配置需要的参数
'''
point_id = event_data.get('point_id')
del event_data['point_id']
# 根据pid获取sid
monitor_dic = await sid_by_pid(point_id)
if monitor_dic is None:
log.info(f'=======no monitor of point id: {point_id}')
return None, None
sid = monitor_dic.get("sid")
scope_type = event_data.get('scope_type')
del event_data['scope_type']
scope_trans_rule = {'0.25ms': 'scope', '0.2s': 'wave_200ms',
'2s': 'electric_2s'}
scope_param = scope_trans_rule.get(scope_type)
if method != 'get' and not scope_param:
log.info(f'=======no scope_param of scope_type: {scope_type}')
return None, None
if method == 'config':
config = {
scope_param: event_data
}
else:
config = scope_param
return sid, config
async def change_param_to_config(req_json, method, type='soe'):
"""
web参数, 转换为与装置通信的报文
soe method: config
"""
sid, config = None, None
if type == 'scope':
sid, config = await get_point_scope_config(req_json, method)
else:
point_id = req_json.get("point_id")
if point_id is not None:
sid, config = await get_point_soe_config(req_json,
with_params=True)
location_id = req_json.get('location_id')
if location_id is not None:
sid, config = await get_location_soe_config(req_json,
with_params=True)
request_data = {}
if sid:
request_id = str(uuid4())
request_data = dict(
request_id=request_id,
time=pendulum.now().to_datetime_string(),
method=method,
sid=sid)
if method == 'get-config':
request_data['key'] = config
elif method == 'config':
request_data['data'] = config
return request_data
async def switch_control_param_to_config(sid, field, switch,
command="switch_control",
method="command"):
"""
联动控制, 转换为与装置通信的报文
soe method: command
"""
# 展龙反馈, 装置不识别00
if switch == "00":
switch = "10"
request_id = str(uuid4())
request_data = dict(request_id=request_id,
time=pendulum.now().to_datetime_string(),
method=method,
sid=sid,
data={'command': command,
'params': {'field': field, 'value': switch}}
)
return request_data