Skip to content

U
unify_api2
Commit dcef5559 authored by wang.wenrong's avatar wang.wenrong
Browse files
Options

Merge branch 'wwr' into 'develop' 

识别记录

See merge request !39
parents 677ae78f 9b8ceb84
Hide whitespace changes
Inline Side-by-side
Showing with 340 additions and 92 deletions
unify_api/modules/anshiu/components/scope_operations_cps.py
View file @ dcef5559
......@@ -67,15 +67,15 @@ class ScopeItemDownload(Model):
ic: float = Float('C相电流')
pttl: float = Float('功率')
lc: float = Float('漏电流')
@dataclass
class ScopeListDownloadResp(Model):
"""
录波下载数据
"""
rows: list = Opt(List("数据列表").items(ScopeItemDownload))
@dataclass
class ScopeContent(Model):
......@@ -92,7 +92,7 @@ class ScopeDetailRep(Model):
识别详情-请求格式
'''
id: str = Opt(Str("doc_id").eg("423_over_gap_i__1655991008"))
@classmethod
def example(cls):
return {
......@@ -117,7 +117,7 @@ class ScopeDetailsResp(Model, DbErr):
v: list = List("电压事件录波").items(ScopeContent)
residual_current: list = Opt(List("2s录波漏电流").items(ScopeContent))
p: list = Opt(List("2s录波功率").items(ScopeContent))
@classmethod
def example(cls):
return {
......@@ -185,7 +185,7 @@ class GetScopeConfigResp(Model, ServerErr):
'''
pid: int = Int("监测点id").eg(20)
rows: list = List("配置内容").items(SetScopeConfigReq)
@classmethod
def example(cls):
return {
......@@ -335,7 +335,7 @@ class InitScopeConfigReq(Model):
初始化配置信息--请求
'''
pids: list = List('pids')
@classmethod
def example(cls):
return {
......@@ -351,7 +351,7 @@ class FlushScopeEsDataReq(Model):
scope_type_list: list = List('录波类型')
start_time: str = Str('开始时间')
end_time: str = Str('结束时间')
@classmethod
def example(cls):
return {
......@@ -399,7 +399,7 @@ scope_list_req_example = {
"end": "2022-06-22 23:59:59",
"pids": [260, 261, 268]
}
}
'''
......@@ -507,3 +507,30 @@ init_scope_config_example = {
}
}
}
@dataclass
class ScopeDetail(Model):
name: str = Str("名称").eg("ia")
value: list = List("值")
@dataclass
class ScopeDetailResp(Model):
name: str = Str("监测点名称").eg("华侨新村270栋101")
ctnum: int = Int("接线方式 2-两表法 3-三表法").eg(1)
sid: str = Str("硬件编号").eg("A2270457094")
event_datetime: str = Str("触发时间").eg("2022-05-26 15:55:03")
duration: str = Str("录波时长").eg("400ms")
message: str = Str("触发原因").eg("漏电流越限")
scope_g: str = Str("录波颗粒度 0.25ms/0.2s/2s").eg("0.25ms")
scope_type: str = Str("录波类型").eg("over_res_cur")
conclusion: str = Str("分析结论").eg("xxxxxxx")
u_list: list = List("电压曲线").items(ScopeDetail)
u_slots: list = List("电压横坐标")
i_list: list = List("电流曲线").items(ScopeDetail)
i_slots: list = List("电流横坐标")
lc_list: list = List("漏电流曲线").items(ScopeDetail)
lc_slots: list = List("漏电流横坐标")
power_list: list = List("功率曲线").items(ScopeDetail)
power_slots: list = List("功率横坐标")
unify_api/modules/anshiu/dao/fine_monitor_dao.py
View file @ dcef5559
......@@ -183,10 +183,10 @@ async def get_mtid_by_pid_dao(pid):
FROM
point
WHERE
pid = {pid}
pid = %s
"""
async with MysqlUtil() as conn:
data = await conn.fetchone(sql, )
data = await conn.fetchone(sql, args=(pid,))
return data
......@@ -200,7 +200,8 @@ async def get_mtids_by_pids_dao(pid):
pid in %s
"""
async with MysqlUtil() as conn:
data = await conn.fetchone(sql, args=(pid,))
data = await conn.fetchall(sql, args=(pid,))
data = [i['mtid'] for i in data if i['mtid']]
return data
......
unify_api/modules/anshiu/dao/scope_operations_dao.py 0 → 100644
View file @ dcef5559
from pot_libs.mysql_util.mysql_util import MysqlUtil
async def get_scope_event_by_event_id(event_id):
"""
获取录波报警详情
:param event_id:
:return:
"""
sql = f"""
select pe.*,m.sid from point_1min_event pe
left join monitor m on pe.mtid = m.mtid
where pe.event_id =%s
"""
async with MysqlUtil() as conn:
result = await conn.fetchone(sql, args=(event_id,))
return result
async def get_scope_detail_by_pid(pid, event_datetime):
"""
获取录波详情
:param pid:
:param event_datetime:
:return:
"""
sql = f"""
select * from point_1min_scope
where pid =%s and create_time=%s
"""
async with MysqlUtil() as conn:
result = await conn.fetchone(sql, args=(pid, event_datetime))
return result
async def get_threshold_by_mtid(mtid):
"""
根据mtid获取阈值
:param mtid:
:return:
"""
sql = f"""
select threshold from soe_config_record scr
left join location l on scr.lid = l.lid
where l.ad_type='residual_current' and scr.etype='overResidualCurrent'
and l.mtid=%s
"""
async with MysqlUtil() as conn:
threshold = await conn.fetch_value(sql, args=(mtid,))
return threshold
unify_api/modules/anshiu/service/fine_monitor_serv.py
View file @ dcef5559
from operator import itemgetter
from itertools import groupby
from pot_libs.common.components.query import Range, Equal, Filter, InGroup, \
Sort
from unify_api.modules.common.dao.common_dao import get_fields_by_mtid
from pot_libs.es_util import es_helper
from pot_libs.es_util.es_query import EsQuery
from pot_libs.logger import log
from pot_libs.common.components.query import PageRequest
from pot_libs.utils.pendulum_wrapper import my_pendulum
from unify_api.modules.anshiu.dao.fine_monitor_dao import get_aiao_1min_dao, \
get_aiao_1day_dao, get_aiao_15min_dao, get_point_15min_chart_dao, \
get_point_1day_chart_dao, get_point_1min_chart_dao, get_point_monitor_dao, \
electric_index_list_dao, electric_index_location_dao
from unify_api.utils import time_format
from unify_api.modules.electric.procedures.electric_util import (
get_wiring_type
)
from unify_api.modules.anshiu.components.fine_monitor_cps import (
Statistics, Chart, ElectricIndexParam
ElectricIndexParam
)
from unify_api.modules.anshiu.procedures.fine_monitor_pds import (
get_es_point_15min_data,
get_es_aiao_15min_data, get_threshold_by_location, get_aiao_1min_pds,
get_threshold_by_location, get_aiao_1min_pds,
get_aiao_data_pds, get_point_1min_chart_pds, get_point_data_chart_pds,
GENERAL_PARAM_FIELD_2, ELECTRIC_QUALITY_FIELD_2,
GENERAL_PARAM_FIELD_3, ELECTRIC_QUALITY_FIELD_3, cal_electic_value,
cal_aiao_value, cal_pt_value
)
from unify_api.utils.time_format import convert_timestamp_to_str, get_time_diff
from unify_api.utils.time_format import get_time_diff
async def get_adio_chart_data(location_group, location_info, date_start,
......
unify_api/modules/anshiu/service/scope_operations_serv.py
View file @ dcef5559
import json
import time
import math
from pot_libs.qingstor_util.qs_client import QsClient
import pandas as pd
from pandas.core.dtypes.inference import is_number
from unify_api.modules.zhiwei_u.fault_foreast.actionFile import actionFilemin
from unify_api.modules.zhiwei_u.fault_foreast.test import leakage_reg
from unify_api.utils.log_utils import LOGGER
from pot_libs.aiomqtt_util.hbmqtt_utils import MqttUtil
from pot_libs.aredis_util import aredis_utils
from pot_libs.common.components.query import PageRequest, Filter, InGroup, \
......@@ -17,7 +20,9 @@ from pot_libs.utils.time_format import convert_dt_to_timestr, \
convert_to_es_str, time_str_to_str
from unify_api.modules.anshiu.components.scope_operations_cps import \
ScopeListItem, ScopeContent, ScopeDetailsResp, GetScopeConfigList, \
init_scope_config_example, ScopeItemDownload
init_scope_config_example, ScopeItemDownload, ScopeDetail
from unify_api.modules.anshiu.dao.scope_operations_dao import \
get_scope_event_by_event_id, get_scope_detail_by_pid, get_threshold_by_mtid
from unify_api.modules.anshiu.procedures.scope_operations_pds import \
get_scope_config_by_pid, set_scope_config_by_pid, add_scope_config_by_pid, \
get_scope_list_by_pid
......@@ -25,6 +30,8 @@ from unify_api.modules.common.dao.common_dao import point_by_points, \
points_by_cid
from unify_api.modules.device_cloud.procedures.mqtt_helper import \
change_param_to_config
from unify_api.modules.electric.procedures.electric_util import \
get_wiring_type_new15
from unify_api.modules.zhiwei_u import config
from unify_api.modules.zhiwei_u.dao.data_es_dao import query_search_scope_pids, \
query_search_scope
......@@ -40,43 +47,13 @@ async def search_scope_service(pids, cid, page_num, page_size, start, end,
'''
获取录波记录
'''
if len(pids) == 0:
datas = await query_search_scope([cid], '', page_num, page_size,
start, end, scope_g)
points = await points_by_cid([cid])
else:
datas = await query_search_scope_pids(pids, page_num, page_size,
start, end, scope_g)
points = await point_by_points(list(set(pids)))
if not datas["hits"]["hits"]:
return [], 0
total = datas["hits"]["total"]
# 获取监测点数据
points = {point['pid']: point for point in points}
# 获取录波明细数据以获取颗粒度
# scope_type_items = await get_scope_type(pids, cid, start, end)
rows = []
for info in datas["hits"]["hits"]:
pid = info["_source"]["point_id"]
message = info["_source"]["message"]
check_dt = info["_source"]["datetime"]
# scope_type_key = "%s_%s" % (pid, info["_source"]["time"])
# scope_type = scope_type_items.get(scope_type_key, 0)
scope_type = info["_source"].get("scope_g", '')
if scope_type == '200ms':
scope_type = '0.2s'
dt = time_format.convert_to_dt(check_dt)
check_dt = time_format.convert_dt_to_timestr(dt)
scope_list_item = ScopeListItem(check_dt=check_dt,
scope_type=scope_type,
point=points[pid].get("name"),
message=message,
doc_id=info["_id"]
)
rows.append(scope_list_item)
return rows, total
datas = await query_search_scope(cid, pids, page_num, page_size,
start, end, scope_g)
total = len(datas)
return datas, total
async def scope_list_download_data(pids, start, end):
......@@ -350,7 +327,7 @@ async def get_scope_config_serv(pid):
'value':
[config.get('start_time_III'),
config.get('stop_time_III')]}
one_data = GetScopeConfigList(state=state,
configs=one_config)
return_data[scope_type] = one_data
......@@ -397,7 +374,7 @@ async def get_mqtt_scope_config(pid):
res_data = await mqtt_func(pub_json, pid)
if not res_data.get('data'):
raise Exception('请求数据为空')
sid = pub_json.get("sid")
# 转换数据字段
trans_fields = {"0.25ms": {"scopeEnable": "en_scope",
......@@ -492,7 +469,7 @@ async def flush_scope_es_data(scope_g, start_time, end_time):
in_groups=[])
page_request = PageRequest(page_size=10000, page_num=1, sort=sort,
filter=filter)
query_body = EsQuery.query(page_request)
query_body['_source'] = ['_id']
async with EsUtil() as es:
......@@ -532,3 +509,120 @@ async def flush_scope_es_data(scope_g, start_time, end_time):
finally:
# 递增
start_dt = every_end_dt
# 1min_event 里面的event_datetime 对应 tdengine 里面scope的ts_origin
async def scope_detail_service(event_id):
# 获取报警信息
event_data = await get_scope_event_by_event_id(event_id)
if not event_data:
return {}, [], [], [], [], [], [], [], []
event_datetime = str(event_data.get("event_datetime"))
pid = event_data.get("pid")
mtid = event_data.get("mtid")
sid = event_data.get("sid")
event_type = event_data.get("event_type")
# 获取录播数据
scope_data = await get_scope_detail_by_pid(pid, event_datetime)
if not scope_data or not scope_data.get("url"):
return {}, [], [], [], [], [], [], [], []
# 接线法:二表/三表
ctnum = await get_wiring_type_new15(pid) or 3
# 查询录波详细数据 还不确定在哪儿拿，现在脚本是minio里边，但是家义说未来要在青云里面
# 测试数据存储
# filedata/electric_ops/scope/A2004000519/2022/6/27/17/1656321462.json
# 获取录波曲线数据
try:
async with QsClient() as qs:
url = scope_data.get("url")
wave_data = await qs.get_object(url)
except Exception as e:
LOGGER.error(f"scope_detail_service error message:{str(e)}")
return {}, [], [], [], [], [], [], [], []
# 录波颗粒度
scope_g = scope_data.get("scope_g")
if ctnum == 2:
i_fields = ['ia', 'ic']
v_fields = ['uab', 'ucb']
else:
i_fields = ['ia', 'ib', 'ic']
v_fields = ['ua', 'ub', 'uc']
u_list, i_list = [], []
residual_current, power = [], []
u_count, i_count, lc_count, power_count = 0, 0, 0, 0
for k, v in wave_data.items():
v = [value if not math.isnan(value) else '' for value in v]
if k in i_fields:
i_list.append(ScopeDetail(name=k, value=v))
u_count = len(v)
if k in v_fields:
u_list.append(ScopeDetail(name=k, value=v))
i_count = len(v)
# 2s颗粒度的会有漏电流及功率
if k in ("lc", "ileak_rms"):
residual_current.append(ScopeDetail(name='漏电流', value=v))
lc_count = len(v)
if k == "pttl":
power.append(ScopeDetail(name='总有功功率', value=v))
power_count = len(v)
u_slots = [i for i in range(1, u_count + 1)]
i_slots = [i for i in range(1, i_count + 1)]
lc_slots = [i for i in range(1, lc_count + 1)]
power_slots = [i for i in range(1, power_count + 1)]
# 结论分析
result = await get_scope_conclusion(wave_data, event_type, mtid, sid,
ctnum)
# 返回信息
data = dict()
data["name"] = event_data.get("name")
data["sid"] = sid
data["event_datetime"] = str(event_data.get("event_datetime"))
# data["duration"] = event_data.get("duration")
if scope_g == '0.25ms':
# data["duration"] = "400ms"
data["duration"] = str(int(i_count * 0.25)) + "ms"
elif scope_g == '200ms':
data["duration"] = "10s"
else:
data["duration"] = "10min"
data["message"] = event_data.get("message")
data["conclusion"] = result
data["scope_g"] = scope_g
data["scope_type"] = event_data.get("event_type")
data["ctnum"] = ctnum
return data or {}, u_list, i_list, residual_current, power, u_slots, \
i_slots, lc_slots, power_slots
async def get_scope_conclusion(wave_data, event_type, mtid, sid, ctnum):
"""
结论分析
:param wave_data:
:param event_type:
:param mtid:
:param sid:
:param ctnum:
:return:
"""
try:
lc_data = wave_data.get("ileak_rms") or wave_data.get("lc")
LOGGER.info(f"wave_data:{wave_data}")
if event_type == "over_res_cur":
threshold = await get_threshold_by_mtid(mtid) or 30
result = leakage_reg(ileak_rms=lc_data,
leak_hold=threshold)
LOGGER.info(f"actionFile 漏电流 sid:{sid}结论:{result}")
else:
res = actionFilemin(wave_data, ctnum)
LOGGER.info(f"结论分析 res:{res}")
if res == "nofault":
result = "不存在故障"
else:
result = ""
for r in res:
result += "%s : %0.0f%%;" % (r[0], r[1] * 100)
except Exception as e:
result = ""
LOGGER.error(f"actionFile error：{e}")
return result
unify_api/modules/anshiu/views/scope_operations.py
View file @ dcef5559
......@@ -7,11 +7,11 @@ from unify_api.modules.anshiu.components.scope_operations_cps import \
SetScopeConfigReq, SetScopeConfigResp, \
scope_list_req_example, ScopeDetailsResp, ScopeDetailRep, \
set_scope_config_example, InitScopeConfigReq, FlushScopeEsDataReq, \
ScopeListDownloadReq, ScopeListDownloadResp
ScopeListDownloadReq, ScopeListDownloadResp, ScopeDetailResp
from unify_api.modules.anshiu.service.scope_operations_serv import \
search_scope_service, scope_detail_data, get_scope_config_serv, \
set_scope_config_serv, init_scope_config, flush_scope_es_data, \
scope_list_download_data
scope_list_download_data, scope_detail_service
@summary("识别记录-列表")
......@@ -67,7 +67,7 @@ async def post_scope_list_download(request,
@summary("识别记录-详情")
async def post_scope_detail(request,
async def post_scope_detail1(request,
body: ScopeDetailRep) -> ScopeDetailsResp:
'''
识别详情
......@@ -76,6 +76,37 @@ async def post_scope_detail(request,
return await scope_detail_data(doc_id)
@summary('数据统计-录波查询-录波详情')
async def post_scope_detail(req, body: ScopeDetailRep) -> ScopeDetailResp:
# 1，获取参数
event_id = body.id
# 2，获取信息
data, u_list, i_list, residual_current, power, u_slots, \
i_slots, lc_slots, power_slots = await scope_detail_service(event_id)
# 3，返回信息
return ScopeDetailResp(
name=data.get("name"),
ctnum=data.get("ctnum"),
sid=data.get("sid"),
event_datetime=data.get("event_datetime"),
duration=data.get("duration"),
message=data.get("message"),
scope_g=data.get("scope_g"),
scope_type=data.get("scope_type"),
conclusion=data.get("conclusion"),
u_list=u_list,
u_slots=u_slots,
i_list=i_list,
i_slots=i_slots,
lc_list=residual_current,
lc_slots=lc_slots,
power_list=power,
power_slots=power_slots,
)
@summary("识别设置-获取配置信息")
async def post_get_scope_config(request,
body: GetScopeConfigReq) -> GetScopeConfigResp:
......
unify_api/modules/zhiwei_u/dao/data_es_dao.py
View file @ dcef5559
......@@ -74,18 +74,18 @@ async def query_location_1min_index(l_database, date_start, date_end,
async def get_search_scope(cid, pid, start, end):
where_list = [f"pid = {pid}"]
if cid:
where_list.append(f"cid in %s")
if start and end:
where_list.append(f"event_datetime >= '{start}' and event_datetime <='{end}' ")
where_list.append(
f"event_datetime >= '{start}' and event_datetime <='{end}' ")
where_str = " and ".join(where_list)
sql = f"select * from point_1min_event where {where_str} " \
f"ORDER BY event_datetime desc limit 5000"
async with MysqlUtil() as conn:
if cid:
data = await conn.fetchall(sql, args=(cid, ))
data = await conn.fetchall(sql, args=(cid,))
else:
data = await conn.fetchall(sql)
return data
......@@ -98,24 +98,33 @@ async def query_search_scope(cid, pid, page_num, page_size,
"""
if len(pid) > 1:
mtid = get_mtids_by_pids_dao(pid)
mtid = await get_mtids_by_pids_dao(pid)
else:
mtid = get_mtid_by_pid_dao(pid)
mtid = mtid['mtid']
mtid = await get_mtid_by_pid_dao(pid)
where = ""
if cid:
where += f" and pe.cid={cid} "
if start_time:
where += f" and event_datetime >= '{start_time}'"
where += f" and pt.event_datetime >= '{start_time}' "
if end_time:
where += f" and event_datetime <= '{end_time}'"
where += f" and pe.event_datetime <= '{end_time}' "
if mtid:
where += f" and pe.mtid={mtid}"
join_sql = ""
if len(mtid) == 1:
where += f" and pe.mtid = {mtid['mtid']} "
else:
where += f" and pe.mtid in {tuple(mtid)} "
if scope_g:
if len(scope_g) == 1:
where += f" AND pe.scope_g = {scope_g[0]} "
else:
where += f" AND pe.scope_g in {tuple(scope_g)} "
sql = f"""
SELECT
pt.event_id doc_id,
pt.event_datetime check_dt,
DATE_FORMAT(pt.event_datetime, '%Y-%m-%d %H:%i:%s') check_dt,
pt.`name` point,
pt.message message,
pe.scope_g scope_type
......@@ -124,19 +133,16 @@ async def query_search_scope(cid, pid, page_num, page_size,
LEFT JOIN point_1min_scope pe ON pt.mtid = pe.mtid
AND pe.create_time = pt.event_datetime
WHERE
pt.event_mode = 'scope'
{where}
pe.cid = 183
AND pt.event_mode = '{scope_g}'
AND pe.pid in (1463, 2248)
ORDER BY
pe.create_time DESC
LIMIT limit {page_num} offset {page_size} """
LIMIT {page_num} , {page_size} """
async with MysqlUtil() as conn:
if cid:
data = await conn.fetchall(sql, args=(cid, ))
else:
data = await conn.fetchall(sql)
data = await conn.fetchall(sql, )
return data
......
unify_api/modules/zhiwei_u/views/scope_operations.py
View file @ dcef5559
......@@ -20,7 +20,7 @@ async def post_search_scope(req, body: SearchScopeRep) -> SearchScopeResq:
@summary('录波查询-详情')
async def post_scope_detail(req, body: ScopeDetailRep) -> SearchScopeResq:
async def post_scope_detai1(req, body: ScopeDetailRep) -> SearchScopeResq:
id = body.id
wave_range = body.wave_range
return await scope_detail_service(id, wave_range)
......
unify_api/utils/log_utils.py 0 → 100644
View file @ dcef5559
import traceback
import re
from pot_libs.logger import Logger
from pot_libs.settings import SETTING
class CLog(object):
def info(self, msg, save_file=None,
group_name=SETTING.mysql_db): # group_name组名一般为项目名，日志存放在此文件夹下
if not save_file:
save_file = get_save_file()
log_obj = Logger.get_logger(save_file, group_name)
log_obj.info(msg, stacklevel=2)
def error(self, msg, save_file=None, group_name=SETTING.mysql_db):
if not save_file:
save_file = get_save_file()
log_obj = Logger.get_logger(save_file, group_name)
log_obj.error(msg, stacklevel=2)
def warning(self, msg, save_file=None, group_name=SETTING.mysql_db):
if not save_file:
save_file = get_save_file()
log_obj = Logger.get_logger(save_file, group_name)
log_obj.warning(msg, stacklevel=2)
def get_save_file():
"""获取日志存储文件路径，以model_name作为存储文件名"""
stack_list = traceback.extract_stack()
filename = stack_list[-3].filename # 调用位置路径
re_str = f"(.*?){SETTING.mysql_db}/{SETTING.mysql_db}(.*?$)"
res_obj = re.search(re_str, filename)
if not res_obj:
return "default"
tmp_dir = res_obj.group(2)
file_name_list = tmp_dir.split("/")
if not file_name_list or len(file_name_list) == 1:
return "default"
save_file = file_name_list[1] # save_file = modules,script,services,tests,utils
if save_file != "modules":
return save_file.split(".")[0] # 防止出现save_file=xxx.py的情况
return file_name_list[2].split(".")[0]
LOGGER = CLog()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment