Skip to content

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

Merge branch 'develop' into wwr

parents 9fc31e09 8663e19b
Hide whitespace changes
Inline Side-by-side
Showing with 697 additions and 609 deletions
unify_api/modules/elec_charge/dao/elec_charge_dao.py
View file @ 61200aa0
......@@ -19,7 +19,7 @@ async def point_day_power_dao(cid, start, end):
ORDER BY pp.create_time
"""
async with MysqlUtil() as conn:
data = await conn.fetchall(sql, args=(cid, ))
data = await conn.fetchall(sql, args=(cid,))
return data
......@@ -31,7 +31,7 @@ async def get_total_kwh_dao(cid, start, end):
and pp.create_time<='{end}' and m.demolished=0
"""
async with MysqlUtil() as conn:
total_kwh = await conn.fetchone(sql, args=(cid, ))
total_kwh = await conn.fetchone(sql, args=(cid,))
return total_kwh
......@@ -40,7 +40,7 @@ async def get_kwh_charge(table_name, name, value, start, end):
f"FROM {table_name} where create_time>='{start}' and " \
f"create_time<='{end}' and {name} = %s"
async with MysqlUtil() as conn:
datas = await conn.fetchone(sql, args=(value, ))
datas = await conn.fetchone(sql, args=(value,))
return datas
......@@ -73,7 +73,7 @@ async def query_charge_aggs(date_start, date_end, cid_list):
"""
start_es = convert_es_str(date_start)
end_es = convert_es_str(date_end)
query_body = {
"size": 0,
"query": {
......@@ -138,132 +138,46 @@ async def power_charge_p_aggs(date_start, date_end, cid_list, interval):
"""
date_histogram,
"""
start_es = convert_es_str(date_start)
end_es = convert_es_str(date_end)
query_body = {
"size": 0,
"query": {
"bool": {
"must": [
{
"terms": {
"cid": cid_list
}
},
{
"range": {
"quarter_time": {
"gte": start_es,
"lte": end_es
}
}
}
]
}
},
"aggs": {
"quarter_time": {
"date_histogram": {
"field": "quarter_time",
"interval": interval,
"time_zone": "+08:00",
"format": "yyyy-MM-dd HH:mm:ss"
},
"aggs": {
"kwh": {
"sum": {
"field": "kwh"
}
},
"charge": {
"sum": {
"field": "charge"
}
},
"p": {
"sum": {
"field": "p"
}
}
}
}
}
}
log.info(query_body)
async with EsUtil() as es:
es_re = await es.search_origin(body=query_body, index=index)
return es_re["aggregations"]["quarter_time"]["buckets"]
if interval == "hour":
time_fmt = "%%Y-%%m-%%d %%H"
elif interval == "day":
time_fmt = "%%Y-%%m-%%d"
else:
time_fmt = "%%Y-%%m-%%d %%H:%%i"
sql = f"""
select date_format(create_time,"{time_fmt}") as create_time,sum(kwh)
kwh,sum(charge) charge,sum(p) p
from company_15min_power
where cid in %s and create_time >= %s and create_time <= %s
group by date_format(create_time,"{time_fmt}")
"""
async with MysqlUtil() as conn:
results = await conn.fetchall(sql,
args=(cid_list, date_start, date_end))
return results or []
async def power_charge_p_cid_aggs(date_start, date_end, cid_list, interval):
"""
excel下载, 按照cid,date_histogram两次聚合,求电量电费
"""
start_es = convert_es_str(date_start)
end_es = convert_es_str(date_end)
query_body = {
"size": 0,
"query": {
"bool": {
"must": [
{
"terms": {
"cid": cid_list
}
},
{
"range": {
"quarter_time": {
"gte": start_es,
"lte": end_es
}
}
}
]
}
},
"aggs": {
"cids": {
"terms": {
"field": "cid",
"size": 1000
},
"aggs": {
"quarter_time": {
"date_histogram": {
"field": "quarter_time",
"interval": interval,
"time_zone": "+08:00",
"format": "yyyy-MM-dd HH:mm:ss"
},
"aggs": {
"kwh": {
"sum": {
"field": "kwh"
}
},
"charge": {
"sum": {
"field": "charge"
}
},
"p": {
"sum": {
"field": "p"
}
}
}
}
}
}
}
}
log.info(query_body)
async with EsUtil() as es:
es_re = await es.search_origin(body=query_body, index=index)
return es_re["aggregations"]["cids"]["buckets"]
if interval == "hour":
time_fmt = "%%Y-%%m-%%d %%H"
elif interval == "day":
time_fmt = "%%Y-%%m-%%d"
else:
time_fmt = "%%Y-%%m-%%d %%H:%%i"
sql = f"""
select cid,date_format(create_time,"{time_fmt}") as create_time,
sum(kwh) kwh,sum(charge) charge,sum(p) p
from company_15min_power
where cid in %s and create_time >= %s and create_time <= %s
group by cid,date_format(create_time,"{time_fmt}")
"""
async with MysqlUtil() as conn:
results = await conn.fetchall(sql,
args=(cid_list, date_start, date_end))
return results or []
async def query_charge_aggs_points(date_start, date_end, point_list):
......@@ -293,7 +207,7 @@ async def query_charge_aggs_points(date_start, date_end, point_list):
"""
start_es = convert_es_str(date_start)
end_es = convert_es_str(date_end)
query_body = {
"size": 0,
"query": {
......@@ -348,7 +262,7 @@ async def query_charge_aggs_points_new15(start, end, point_list):
f"where pid in %s and create_time BETWEEN '{start}' and '{end}' " \
f"GROUP BY pid"
async with MysqlUtil() as conn:
datas = await conn.fetchall(sql, args=(point_list, ))
datas = await conn.fetchall(sql, args=(point_list,))
return datas
......@@ -356,7 +270,7 @@ async def histogram_aggs_points(date_start, date_end, point_list, interval):
"""date_histogram"""
start_es = convert_es_str(date_start)
end_es = convert_es_str(date_end)
query_body = {
"size": 0,
"query": {
......@@ -418,7 +332,7 @@ async def power_charge_p_point_aggs(date_start, date_end, pid_list, interval):
"""
start_es = convert_es_str(date_start)
end_es = convert_es_str(date_end)
query_body = {
"size": 0,
"query": {
......@@ -495,7 +409,6 @@ async def point_aggs_kwh(point_list, start=None, end=None):
return data
async def point_aggs_kwh_new15(point_list, start=None, end=None):
"""1.5版本根据pid,求电量电费"""
if start and end:
......@@ -506,7 +419,7 @@ async def point_aggs_kwh_new15(point_list, start=None, end=None):
sql = "SELECT sum(kwh) kwh,sum(charge) charge FROM point_15min_power" \
" where pid in %s"
async with MysqlUtil() as conn:
data = await conn.fetchone(sql, args=(point_list, ))
data = await conn.fetchone(sql, args=(point_list,))
return data
......@@ -516,7 +429,7 @@ async def extended_bounds_agg(date_start, date_end, cid_list, interval):
"""
start_es = convert_es_str(date_start)
end_es = convert_es_str(date_end)
query_body = {
"size": 0,
"query": {
......
unify_api/modules/elec_charge/procedures/elec_charge_pds.py
View file @ 61200aa0
......@@ -217,7 +217,7 @@ async def power_aggs_cid_proxy(start, end, cid_list, date_type):
for info in es_re_this:
cid = info.get("key")
cid_name = com_dic[cid]["shortname"]
kwh = round_2(info.get("kwh")["value"])
if kwh == 0:
continue
......@@ -229,9 +229,10 @@ async def power_aggs_cid_proxy(start, end, cid_list, date_type):
continue
except Exception as e:
log.error(e)
log.info(f"本次有电量数据, 上周期没有电量数据, cid:{cid}, start:{start}, end:{end}")
log.info(
f"本次有电量数据, 上周期没有电量数据, cid:{cid}, start:{start}, end:{end}")
continue
charge = round_2(info.get("charge")["value"])
try:
charge_last = es_re_last_dic[cid]["charge"]["value"]
......@@ -241,7 +242,8 @@ async def power_aggs_cid_proxy(start, end, cid_list, date_type):
except Exception as e:
log.error(e)
log.info("本次有数据, 上周期没有数据")
log.info(f"本次有电费数据, 上周期没有电费数据, cid:{cid}, start:{start}, end:{end}")
log.info(
f"本次有电费数据, 上周期没有电费数据, cid:{cid}, start:{start}, end:{end}")
continue
price = round_2(charge / kwh)
price_last = round_2(charge_last / kwh_last)
......
unify_api/modules/elec_charge/procedures/elec_statis_proxy_pds.py
View file @ 61200aa0
......@@ -113,11 +113,11 @@ def by_slots(slots, es_re_dic):
for slot in slots:
if slot in es_re_dic:
# 1.每个时间点,电量信息
kwh_value = round_2(es_re_dic[slot].get("kwh").get("value"))
kwh_value = round_2(es_re_dic[slot].get("kwh"))
# 2.每个时间点,电费信息
charge_value = round_2(es_re_dic[slot].get("charge").get("value"))
charge_value = round_2(es_re_dic[slot].get("charge"))
# 3. 功率
p_value = round_2(es_re_dic[slot].get("p").get("value"))
p_value = round_2(es_re_dic[slot].get("p"))
# 4. 电价
try:
price_value = round_2(charge_value / kwh_value)
......@@ -141,16 +141,18 @@ async def power_charge_p_proxy(cid_list, start, end, date_type, interval):
# 1. 96个点数据
slots_96 = SLOTS_15MIN
es_re_96 = await power_charge_p_aggs(start, end, cid_list, "15m")
es_re_96_dic = es_process(es_re_96, fmat="HH:mm") # 为了es结果和slots对应
es_re_96_dic = es_process(es_re_96, fmat="HH:mm",
time_key="create_time") # 为了es结果和slots对应
kwh_24, charge_24, p_24, price_24 = by_slots(slots_96, es_re_96_dic)
# 96个点,工厂电量电费
# 2. 24个点数据
slots_24 = SLOTS[date_type]
es_re_24 = await power_charge_p_aggs(start, end, cid_list, "hour")
es_re_24_dic = es_process(es_re_24, fmat="HH:mm") # 为了es结果和slots对应
es_re_24_dic = es_process(es_re_24, fmat="HH:mm",
time_key="create_time") # 为了es结果和slots对应
kwh_24, charge_24, p_24, price_24 = by_slots(slots_24, es_re_24_dic)
# elif date_type == "month":
# intervel, slots = time_pick_transf(start, end)
# es_re = es_process(es_re, fmat="MM-DD") # 为了es结果和slots对应
unify_api/modules/elec_charge/service/elec_charge_service.py
View file @ 61200aa0
import pendulum
from unify_api.constants import POINT_LEVEL_MAP
from unify_api.modules.common.components.common_cps import LevelResp
from unify_api.modules.common.procedures.points import points_by_storeys
......@@ -6,6 +8,8 @@ from unify_api.modules.elec_charge.dao.elec_charge_dao import \
query_charge_aggs_points, query_charge_aggs_points_new15
from unify_api.modules.electric.dao.electric_dao import \
monitor_point_join_by_points
from unify_api.modules.home_page.procedures.count_info_pds import current_load, \
current_load_new15
from unify_api.utils.common_utils import round_2, division_two
......@@ -86,11 +90,11 @@ async def kwh_card_level_service(cid, point_list, start, end):
kwh = round_2(es_res[point_id]["kwh"])
charge = round_2(es_res[point_id]["charge"])
price = round_2(division_two(charge, kwh))
res_dic["kwh"] = kwh
res_dic["charge"] = charge
res_dic["price"] = price
ret_data[m_type].append(res_dic)
return LevelResp(
inline=ret_data["inline"],
......@@ -101,4 +105,11 @@ async def kwh_card_level_service(cid, point_list, start, end):
)
async def load_info_service(cid_list):
# 实时负荷
cur_load = await current_load_new15(cid_list)
yesterday_dt = pendulum.now(tz="Asia/Shanghai").subtract(days=1)
yes_load = await current_load_new15(cid_list, yesterday_dt)
load_percent = round((cur_load - yes_load) / yes_load,
2) if cur_load and yes_load else ""
return cur_load, yes_load, load_percent
unify_api/modules/elec_charge/views/elec_charge.py
View file @ 61200aa0
......@@ -16,14 +16,16 @@ from unify_api.modules.elec_charge.components.elec_charge_cps import \
power_overview_example, PricePolicyReq, PricePolicyResp, \
PricePolicy, AverPriceReq, PowerViewRes, Spvf, AverPriceResp, ChargeKwh, \
IndexChargeReq, IndexChargeResp, PopReq, PopResp, MtpResp, PspResp, \
IpspResp, KpReq, KpResp, KclReq, ProductProxyReq
IpspResp, KpReq, KpResp, KclReq, ProductProxyReq, LoadInfoReq, LoadInfoResp
from unify_api.modules.elec_charge.dao.elec_charge_dao import \
query_charge_aggs_points, get_kwh_charge, query_charge_aggs_points_new15
from unify_api.modules.elec_charge.procedures.elec_charge_pds import \
quarters_trans, power_overview_proxy, total_value, power_aggs_cid_proxy_new15, \
quarters_trans, power_overview_proxy, total_value, \
power_aggs_cid_proxy_new15, \
power_index_cid_proxy_new15, power_overview_proxy15
from unify_api.modules.elec_charge.service.elec_charge_service import \
kwh_points_service, kwh_card_level_service
kwh_points_service, kwh_card_level_service, load_info_service
from unify_api.modules.users.procedures.jwt_user import jwt_user
from unify_api.utils.common_utils import round_2, round_4, NumListHelper
from unify_api.utils.es_query_body import agg_statistics
from unify_api.utils.request_util import filed_value_from_list
......@@ -289,7 +291,7 @@ async def post_power_overview_proxy(req, body: PopReq) -> PopResp:
proxy_id = body.proxy_id
host = req.host
product = PRODUCT.get(host)
user_id = req.ctx.user_id
user_id = jwt_user(req)
# 全部工厂
if not cid_list:
log.info(f"power_overview_proxy根据用户userId:{user_id} "
......@@ -314,13 +316,14 @@ async def post_power_overview_proxy(req, body: PopReq) -> PopResp:
# 获取上一周期开始结束时间
start_last, end_last = last_time_str(start, end, date_type)
power, charge = await power_overview_proxy15(start, end, cid_list)
power_last, charge_last = await power_overview_proxy15(start_last, end_last,
cid_list)
power_last, charge_last = await power_overview_proxy15(start_last,
end_last,
cid_list)
if not all([power, charge, power_last, charge_last]):
return PopResp(power=Spvf(), charge=Spvf())
total_power = total_value(power)
total_charge = total_value(charge)
total_power_last = total_value(power_last)
total_charge_last = total_value(charge_last)
# 增长率
......@@ -351,7 +354,7 @@ async def post_month_today_proxy(req, body: ProductProxyReq) -> MtpResp:
# 1. 获取参数
host = req.host
product = PRODUCT.get(host)
user_id = req.ctx.user_id
user_id = jwt_user(req)
# cid_list = await get_cids(user_id, product)
proxy_id = body.proxy_id
cid_list = await get_proxy_cids(user_id, product, proxy_id) \
......@@ -364,7 +367,7 @@ async def post_month_today_proxy(req, body: ProductProxyReq) -> MtpResp:
today_start, today_end, month_start, month_end = today_month_date()
# 2. 本月/上月数据
last_month_start, last_month_end = last_time_str(month_start, month_end,
"month")
"month", True)
this_month_p, this_month_charge = await power_overview_proxy15(
month_start, month_end, cid_list)
last_month_p, last_month_charge = await power_overview_proxy15(
......@@ -541,3 +544,31 @@ async def post_kwh_card_level(req, body: KclReq) -> LevelResp:
start = body.start
end = body.end
return await kwh_card_level_service(cid, point_list, start, end)
@summary("获取知电管理版首页负荷信息")
async def post_load_info(request, body: LoadInfoReq) -> LoadInfoResp:
# 1. 获取company_id
# 1. 获取参数
product = PRODUCT.get(request.host)
user_id = jwt_user(request)
proxy_id = body.proxy_id
cid_list = await get_proxy_cids(user_id, product, proxy_id) \
if proxy_id else None
# 全部工厂
if not cid_list:
log.info(f"未查询到工厂, userId:{user_id} product:{product}")
return LoadInfoResp()
try:
# 实时负荷,昨日同时负荷,对比昨日
current_load, yesterday_load, load_percent = await load_info_service(
cid_list)
except Exception as e:
log.exception(e)
return LoadInfoResp().server_error()
return LoadInfoResp(
current_load=current_load,
yesterday_load=yesterday_load,
load_percent=load_percent
)
unify_api/modules/elec_charge/views/elec_statis_proxy.py
View file @ 61200aa0
import copy
import json
from itertools import groupby
from operator import itemgetter
from pot_libs.logger import log
from pot_libs.mysql_util.mysql_util import MysqlUtil
from pot_libs.sanic_api import summary
......@@ -50,37 +53,41 @@ async def post_power_statis_proxy(req,
# 2. 如果是日统计,则需要增加今日/昨日负荷曲线, 15min一个点
if date_type == "day":
# 电量电费
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id, start, end,
date_type)
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id, start,
end,
date_type)
# 需要增加15min电量电费, 渠道版首页不需要下载,暂时去掉
# 今日/昨日负荷曲线
today_p = await proxy_today_yesterday_p_new15(cid_list, start, end)
ysd_start, ysd_end = last_time_str(start, end, "day")
yesterday_p = await proxy_today_yesterday_p_new15(cid_list,
ysd_start,
ysd_end)
ysd_start,
ysd_end)
return PcStatiResp(kwh=kwh_sv, charge=charge_sv, today_p=today_p,
yesterday_p=yesterday_p)
elif date_type == "month":
# 本月电量电费, 平均电价
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id, start, end,
date_type)
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id, start,
end,
date_type)
this_aver_price = aver_price(kwh_sv, charge_sv)
# 上月电量电费, 平均电价
last_start, last_end = last_time_str(start, end, "month")
# 需要增加15min电量电费
last_kwh_sv, last_charge_sv = await power_charge_new15(cid_list, point_id,
last_start,
last_end,
date_type)
last_kwh_sv, last_charge_sv = await power_charge_new15(cid_list,
point_id,
last_start,
last_end,
date_type)
last_aver_price = aver_price(last_kwh_sv, last_charge_sv)
return PcStatiResp(kwh=kwh_sv, charge=charge_sv,
this_aver_price=this_aver_price,
last_aver_price=last_aver_price)
elif date_type == "year":
# 本月电量电费
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id, start, end,
date_type)
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id, start,
end,
date_type)
this_aver_price = aver_price(kwh_sv, charge_sv)
return PcStatiResp(kwh=kwh_sv, charge=charge_sv,
this_aver_price=this_aver_price)
......@@ -92,18 +99,20 @@ async def post_power_statis_proxy(req,
# 自定义选时范围，不需要最后时间的数据，解决bug
end = end_f.subtract(minutes=1).format("YYYY-MM-DD HH:mm:ss")
# 电量电费
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id, start,
end,
date_type)
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id,
start,
end,
date_type)
# 负荷曲线
this_p = await proxy_today_yesterday_p_new15(cid_list, start,
end)
end)
# 需要增加15min电量电费, 渠道版首页不需要下载,暂时去掉
return PcStatiResp(kwh=kwh_sv, charge=charge_sv, today_p=this_p)
else:
# 电量电费
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id, start,
end, date_type)
kwh_sv, charge_sv = await power_charge_new15(cid_list, point_id,
start,
end, date_type)
# 平均电价
this_aver_price = aver_price(kwh_sv, charge_sv)
return PcStatiResp(kwh=kwh_sv, charge=charge_sv,
......@@ -361,28 +370,28 @@ async def post_power_statist_opt(req, body: PopReq) -> PcStatiResp:
if date_type == "day":
# 电量电费
kwh_sv, charge_sv = await power_charge_new15(cid_list, -1, start, end,
date_type)
date_type)
# 需要增加15min电量电费, 渠道版首页不需要下载,暂时去掉
# 今日/昨日负荷曲线
today_p = await proxy_today_yesterday_p_new15(cid_list, start, end)
ysd_start, ysd_end = last_time_str(start, end, "day")
yesterday_p = await proxy_today_yesterday_p_new15(cid_list,
ysd_start,
ysd_end)
ysd_start,
ysd_end)
return PcStatiResp(kwh=kwh_sv, charge=charge_sv, today_p=today_p,
yesterday_p=yesterday_p)
elif date_type == "month":
# 本月电量电费, 平均电价
kwh_sv, charge_sv = await power_charge_new15(cid_list, -1, start, end,
date_type)
date_type)
this_aver_price = aver_price(kwh_sv, charge_sv)
# 上月电量电费, 平均电价
last_start, last_end = last_time_str(start, end, "month")
# 需要增加15min电量电费
last_kwh_sv, last_charge_sv = await power_charge_new15(cid_list, -1,
last_start,
last_end,
date_type)
last_start,
last_end,
date_type)
last_aver_price = aver_price(last_kwh_sv, last_charge_sv)
return PcStatiResp(kwh=kwh_sv, charge=charge_sv,
this_aver_price=this_aver_price,
......@@ -427,7 +436,8 @@ async def get_power_statist_download(req):
slots_24 = SLOTS[date_type]
# 1. 96个点, 电量\电费\平均功率\电价
es_re_96 = await power_charge_p_aggs(start, end, cid_list, "15m")
es_re_96_dic = es_process(es_re_96, fmat="HH:mm") # 为了es结果和slots对应
es_re_96_dic = es_process(es_re_96, "HH:mm", time_key="create_time")
# 为了es结果和slots对应
kwh_96, charge_96, p_96, price_96 = by_slots(slots_96, es_re_96_dic)
p_96.append("")
price_96.append("")
......@@ -445,32 +455,33 @@ async def get_power_statist_download(req):
power_96_dic = {"电量(万kWh)": slots_96_zj}
charge_96_dic = {"电费(万元)": slots_96_zj}
has_power_cids = [] # es查询有数据的工厂
for info in res_cid_96:
cid = info.get("key")
for cid, cid_info in groupby(res_cid_96, key=itemgetter("cid")):
cid_one = [one for one in cid_info]
has_power_cids.append(cid)
cid_name = com_dic[cid]["shortname"]
# 把slots作为key提出来
info_dic = es_process(info["quarter_time"]["buckets"],
fmat="HH:mm")
info_dic = es_process(cid_one, fmat="HH:mm",
time_key="create_time")
kwh_96, charge_96, p_96, price_96 = by_slots(slots_96,
info_dic)
power_96_dic[cid_name] = division_down(kwh_96)
charge_96_dic[cid_name] = division_down(charge_96)
# 没有电量数据的工厂, 设为空字符串
for cid in cid_list:
if cid not in has_power_cids:
cid_name = com_dic[cid]["shortname"]
power_96_dic[cid_name] = list(str(' ') * 97)
charge_96_dic[cid_name] = list(str(' ') * 97)
df1 = pd.DataFrame(power_96_dic)
df1.to_excel(writer, sheet_name="96个点电量", index=False)
df2 = pd.DataFrame(charge_96_dic)
df2.to_excel(writer, sheet_name="96个点电费", index=False)
# 3. 24个点, 电量\电费\平均功率\电价
es_re_24 = await power_charge_p_aggs(start, end, cid_list, "hour")
es_re_24_dic = es_process(es_re_24, fmat="HH:mm") # 为了es结果和slots对应
es_re_24_dic = es_process(es_re_24, fmat="HH:mm",
time_key="create_time") # 为了es结果和slots对应
kwh_24, charge_24, p_24, price_24 = by_slots(slots_24, es_re_24_dic)
slots_24_zj = copy.deepcopy(slots_24)
slots_24_zj.append("总计")
......@@ -491,25 +502,25 @@ async def get_power_statist_download(req):
power_24_dic = {"电量(万kWh)": slots_24_zj}
charge_24_dic = {"电费(万元)": slots_24_zj}
has_power_cids_24 = [] # es查询有数据的工厂
for info in res_cid_24:
cid = info.get("key")
for cid, cid_info in groupby(res_cid_24, key=itemgetter("cid")):
cid_one = [one for one in cid_info]
has_power_cids_24.append(cid)
cid_name = com_dic[cid]["shortname"]
# 把slots作为key提出来
info_dic_24 = es_process(info["quarter_time"]["buckets"],
fmat="HH:mm")
info_dic_24 = es_process(cid_one, fmat="HH:mm",
time_key="create_time")
kwh_24, charge_24, p_24, price_24 = by_slots(slots_24,
info_dic_24)
power_24_dic[cid_name] = division_down(kwh_24)
charge_24_dic[cid_name] = division_down(charge_24)
# 没有电量数据的工厂, 设为空字符串
for cid in cid_list:
if cid not in has_power_cids_24:
cid_name = com_dic[cid]["shortname"]
power_24_dic[cid_name] = list(str(' ') * 25)
charge_24_dic[cid_name] = list(str(' ') * 25)
df3 = pd.DataFrame(power_24_dic)
df3.to_excel(writer, sheet_name="24个点电量", index=False)
df4 = pd.DataFrame(charge_24_dic)
......@@ -521,7 +532,8 @@ async def get_power_statist_download(req):
slots_m_p = copy.deepcopy(slots_m)
slots_m_p.append("总计")
es_re_m = await power_charge_p_aggs(start, end, cid_list, "day")
es_re_m_dic = es_process(es_re_m, fmat="MM-DD") # 为了es结果和slots对应
es_re_m_dic = es_process(es_re_m, fmat="MM-DD",
time_key="create_time") # 为了es结果和slots对应
kwh_m, charge_m, p_m, price_m = by_slots(slots_m, es_re_m_dic)
price_m.append("")
dict_m = {
......@@ -537,24 +549,24 @@ async def get_power_statist_download(req):
power_m_dic = {"电量(万kWh)": slots_m_p}
charge_m_dic = {"电费(万元)": slots_m_p}
has_power_cids = [] # es查询有数据的工厂
for info in res_cid_m:
cid = info.get("key")
for cid, cid_info in groupby(res_cid_m, key=itemgetter("cid")):
cid_one = [one for one in cid_info]
has_power_cids.append(cid)
cid_name = com_dic[cid]["shortname"]
# 把slots作为key提出来
info_dic = es_process(info["quarter_time"]["buckets"],
fmat="MM-DD")
info_dic = es_process(cid_one, fmat="MM-DD",
time_key="create_time")
kwh_m, charge_m, p_m, price_m = by_slots(slots_m, info_dic)
power_m_dic[cid_name] = division_down(kwh_m)
charge_m_dic[cid_name] = division_down(charge_m)
# 没有电量数据的工厂, 设为空字符串
for cid in cid_list:
if cid not in has_power_cids:
cid_name = com_dic[cid]["shortname"]
power_m_dic[cid_name] = list(str(' ') * (len(slots_m) + 1))
charge_m_dic[cid_name] = list(str(' ') * (len(slots_m) + 1))
df1 = pd.DataFrame(power_m_dic)
df1.to_excel(writer, sheet_name="电量", index=False)
df2 = pd.DataFrame(charge_m_dic)
......@@ -583,7 +595,7 @@ async def get_power_company_download(req):
start = args.get("start")
end = args.get("end")
date_type = args.get("date_type")
# 参数为point_id, 转换为point_list
if point_id == -1: # 选的全部
# 1.找出工厂所有pid，point表add_to_company字段为1
......@@ -593,7 +605,7 @@ async def get_power_company_download(req):
point_list = [point.get("pid") for point in point_info]
else:
point_list = [point_id]
# 查询point
point_info_list = await point_by_points(point_list)
pid_dic = {i["pid"]: i for i in point_info_list}
......@@ -637,14 +649,14 @@ async def get_power_company_download(req):
info_dic)
power_96_dic[pid_name] = division_down(kwh_96)
charge_96_dic[pid_name] = division_down(charge_96)
# 没有电量数据的监测点, 设为空字符串
for pid in point_list:
if pid not in has_power_pids:
pid_name = pid_dic[pid]["name"]
power_96_dic[pid_name] = list(str(' ') * 97)
charge_96_dic[pid_name] = list(str(' ') * 97)
df1 = pd.DataFrame(power_96_dic)
df1.to_excel(writer, sheet_name="96个点电量", index=False)
df2 = pd.DataFrame(charge_96_dic)
......@@ -683,14 +695,14 @@ async def get_power_company_download(req):
info_dic_24)
power_24_dic[pid_name] = division_down(kwh_24)
charge_24_dic[pid_name] = division_down(charge_24)
# 没有电量数据的工厂, 设为空字符串
for pid in point_list:
if pid not in has_power_pids_24:
pid_name = pid_dic[pid]["name"]
power_24_dic[pid_name] = list(str(' ') * 25)
charge_24_dic[pid_name] = list(str(' ') * 25)
df3 = pd.DataFrame(power_24_dic)
df3.to_excel(writer, sheet_name="24个点电量", index=False)
df4 = pd.DataFrame(charge_24_dic)
......@@ -728,14 +740,14 @@ async def get_power_company_download(req):
kwh_m, charge_m, p_m, price_m = by_slots(slots_m, info_dic)
power_m_dic[pid_name] = division_down(kwh_m)
charge_m_dic[pid_name] = division_down(charge_m)
# 没有电量数据的工厂, 设为空字符串
for pid in point_list:
if pid not in has_power_pids:
pid_name = pid_dic[pid]["name"]
power_m_dic[pid_name] = list(str(' ') * (len(slots_m) + 1))
charge_m_dic[pid_name] = list(str(' ') * (len(slots_m) + 1))
df1 = pd.DataFrame(power_m_dic)
df1.to_excel(writer, sheet_name="电量", index=False)
df2 = pd.DataFrame(charge_m_dic)
......@@ -774,14 +786,14 @@ async def get_power_company_download(req):
kwh_m, charge_m, p_m, price_m = by_slots(slots_m, info_dic)
power_m_dic[pid_name] = division_down(kwh_m)
charge_m_dic[pid_name] = division_down(charge_m)
# 没有电量数据的point, 设为空字符串
for pid in point_list:
if pid not in has_power_pids:
pid_name = pid_dic[pid]["name"]
power_m_dic[pid_name] = list(str(' ') * (len(slots_m) + 1))
charge_m_dic[pid_name] = list(str(' ') * (len(slots_m) + 1))
df1 = pd.DataFrame(power_m_dic)
df1.to_excel(writer, sheet_name="电量", index=False)
df2 = pd.DataFrame(charge_m_dic)
......
unify_api/modules/electric/dao/electric_dao.py
View file @ 61200aa0
......@@ -5,7 +5,7 @@ async def monitor_point_join_by_points(points):
"""monitor和point关联"""
sql = "SELECT m.mtid,p.ctnum,m.name, m.m_type, p.pid,p.cid " \
"FROM monitor m inner join point p on m.mtid = p.mtid " \
"WHERE p.pid in %s and m.demolished = 0 order by field(p.pid,{})".\
"WHERE p.pid in %s and m.demolished = 0 order by field(p.pid,{})". \
format(str(points).replace("[", "").replace("]", ""))
async with MysqlUtil() as conn:
monitor_point_list = await conn.fetchall(sql, args=(tuple(points),))
......@@ -26,7 +26,7 @@ async def get_electric_datas_dao(table_name, pid, start, end):
sql = f"SELECT * FROM {table_name} where pid=%s and create_time " \
f"BETWEEN '{start}' and '{end}' ORDER BY create_time desc"
async with MysqlUtil() as conn:
electric_datas = await conn.fetchall(sql, args=(pid, ))
electric_datas = await conn.fetchall(sql, args=(pid,))
return electric_datas
......@@ -35,7 +35,7 @@ async def get_qual_history_dao(table_name, pid, start, end, date_format):
f"p.* FROM {table_name} p where p.pid=%s and p.create_time " \
f"BETWEEN '{start}' and '{end}' order by p.create_time"
async with MysqlUtil() as conn:
datas = await conn.fetchall(sql, args=(pid, ))
datas = await conn.fetchall(sql, args=(pid,))
return datas
......@@ -51,6 +51,12 @@ async def get_elec_history_dao(table_name, pid, start, end, date_format):
async def get_elec_mtid_sid_by_cid(cid):
if isinstance(cid, tuple):
cid_tuple = cid
elif isinstance(cid, list):
cid_tuple = tuple(cid)
else:
cid_tuple = (cid,)
sql = (
f"""
SELECT
......@@ -59,9 +65,9 @@ async def get_elec_mtid_sid_by_cid(cid):
FROM
monitor
WHERE
cid = {cid};
cid in %s;
"""
)
async with MysqlUtil() as conn:
datas = await conn.fetchall(sql)
datas = await conn.fetchall(sql, args=(cid_tuple,))
return datas if datas else []
unify_api/modules/electric/service/electric_service.py
View file @ 61200aa0
......@@ -212,7 +212,8 @@ async def elec_card_level_service(point_list):
# point_mid = await batch_get_wiring_type(point_list)
# # 3. 获取redis数据
# res_redis = await elec_current_data(point_mid)
mtids = [monitor["mtid"] for monitor in monitor_point_list if monitor["mtid"]]
mtids = [monitor["mtid"] for monitor in monitor_point_list if
monitor["mtid"]]
cid = monitor_point_list[0]['cid'] if len(monitor_point_list) > 0 else 0
results = await elec_current_data_new15(mtids, cid)
# 4. 返回数据
......@@ -286,7 +287,7 @@ async def elec_card_level_service(point_list):
"ub_dev": "",
"uc_dev": "",
}
ret_data[m_type].append(res_dic)
return EclResp(
inline=ret_data["inline"],
......@@ -326,15 +327,15 @@ async def qual_current_level_service(point_list):
# 初始化返回dic
if res.get(mtid):
time_str = res[mtid]["ts"][:-4]
fdia = round_2(res[mtid].get("fdia"))
fdib = round_2(res[mtid].get("fdib"))
fdic = round_2(res[mtid].get("fdic"))
thdia = round_4(res[mtid].get("thdia"))
thdib = round_4(res[mtid].get("thdib"))
thdic = round_4(res[mtid].get("thdic"))
res_dic = {
"name": m_name,
"point_id": point_id,
......@@ -344,7 +345,7 @@ async def qual_current_level_service(point_list):
"thdia": thdia,
"thdib": thdib,
"thdic": thdic,
"thdua": round_4(res[mtid].get("thdua")),
"thdub": round_4(res[mtid].get("thdub")),
"thduc": round_4(res[mtid].get("thduc")),
......@@ -385,7 +386,7 @@ async def qual_current_level_service(point_list):
"thdia": "",
"thdib": "",
"thdic": "",
"thdua": "",
"thdub": "",
"thduc": "",
......@@ -409,7 +410,7 @@ async def qual_current_level_service(point_list):
"thdib_virtual": "",
"thdic_virtual": "",
}
ret_data[m_type].append(res_dic)
return QclResp(
inline=ret_data["inline"],
......@@ -562,14 +563,14 @@ async def elec_index_service(cid, point_id, date_start, date_end):
"freq_dev_min",
"freq_dev_max",
]
query_body = EsQuery.aggr_index(page_request,
stats_items=common_items + elec_qual_items)
async with EsUtil() as es:
es_results = await es.search_origin(body=query_body,
index=constants.POINT_15MIN_INDEX)
aggregations = es_results.get("aggregations", {})
# 常规参数统计
common_indexes = []
......@@ -593,7 +594,7 @@ async def elec_index_service(cid, point_id, date_start, date_end):
else:
max_value = ""
max_value_time = ""
# 最小值
min_info = aggregations.get(f"{item}_min_min", {})
hits = min_info.get("hits", {}).get("hits")
......@@ -609,10 +610,10 @@ async def elec_index_service(cid, point_id, date_start, date_end):
else:
min_value = ""
min_value_time = ""
avg = aggregations.get(f"{item}_mean_avg", {}).get("value")
avg = round(avg, 2) if avg is not None else ""
elec_index = ElecIndex(
stats_index=item,
max=max_value,
......@@ -622,7 +623,7 @@ async def elec_index_service(cid, point_id, date_start, date_end):
avg=avg,
)
common_indexes.append(elec_index)
# 电能质量统计
elec_qual_indexes = []
_elec_qual_items = {i.rsplit("_", 1)[0] for i in elec_qual_items}
......@@ -641,7 +642,7 @@ async def elec_index_service(cid, point_id, date_start, date_end):
else:
max_value = ""
max_value_time = ""
# 最小值
min_info = aggregations.get(f"{item}_min_min", {})
hits = min_info.get("hits", {}).get("hits")
......@@ -656,10 +657,10 @@ async def elec_index_service(cid, point_id, date_start, date_end):
else:
min_value = ""
min_value_time = ""
avg = aggregations.get(f"{item}_mean_avg", {}).get("value")
avg = avg if avg is not None else ""
elec_index = ElecIndex(
stats_index=item,
max=max_value,
......@@ -669,7 +670,7 @@ async def elec_index_service(cid, point_id, date_start, date_end):
avg=avg,
)
elec_qual_indexes.append(elec_index)
if cid:
# 小程序需要这漏电流和温度
residual_current_map = await location_stats_statics(
......@@ -698,7 +699,7 @@ async def elec_index_service(cid, point_id, date_start, date_end):
else None,
)
)
temp_map = await location_stats_statics(
cid, point_id, start_tt, end_tt, _type="temperature"
)
......@@ -773,83 +774,101 @@ async def elec_index_service_new15(cid, point_id, start, end):
"ua_dev_mean", "ua_dev_min", "ua_dev_max",
"freq_dev_mean", "freq_dev_min", "freq_dev_max"]
datas = await get_electric_datas_dao(table_name, point_id, start, end)
if not datas:
return ElecIndexResponse(
ctnum=ctnum, common_indexes=[],
elec_qual_indexes=[]
)
# if not datas:
# return ElecIndexResponse(
# ctnum=ctnum, common_indexes=[],
# elec_qual_indexes=[]
# )
df = pd.DataFrame(list(datas))
# 常规参数统计
common_indexes = []
_common_items = {i.rsplit("_", 1)[0] for i in common_items}
for item in _common_items:
item_name = item.rsplit("_", 1)[0]
max_item_name = f"{item}_max"
max_value = df[max_item_name].max()
if not pd.isna(max_value):
max_datas = df.loc[df[max_item_name].idxmax()].to_dict()
max_time = max_datas.get(f"{item_name}_max_time")
max_time = '' if pd.isnull(max_time) else str(max_time)
else:
max_value, max_time = "", ""
min_item_name = f"{item}_min"
min_value = df[min_item_name].min()
if not pd.isna(min_value):
min_datas = df.loc[df[min_item_name].idxmin()].to_dict()
min_time = min_datas.get(f"{item_name}_min_time")
min_time = '' if pd.isnull(min_time) else str(min_time)
else:
min_value, min_time = "", ""
mean_item_name = f"{item}_mean"
avg_value = df[mean_item_name].mean()
if not pd.isna(avg_value):
avg_value = round(avg_value, 2) if avg_value else ""
if datas:
max_item_name = f"{item}_max"
max_value = df[max_item_name].max()
if not pd.isna(max_value):
max_datas = df.loc[df[max_item_name].idxmax()].to_dict()
max_time = max_datas.get(f"{item}_max_time")
max_time = '' if pd.isnull(max_time) else str(max_time)
else:
max_value, max_time = "", ""
min_item_name = f"{item}_min"
min_value = df[min_item_name].min()
if not pd.isna(min_value):
min_datas = df.loc[df[min_item_name].idxmin()].to_dict()
min_time = min_datas.get(f"{item}_min_time")
min_time = '' if pd.isnull(min_time) else str(min_time)
else:
min_value, min_time = "", ""
mean_item_name = f"{item}_mean"
avg_value = df[mean_item_name].mean()
if not pd.isna(avg_value):
avg_value = round(avg_value, 2) if avg_value else ""
else:
avg_value = ""
elec_index = ElecIndex(
stats_index=item,
max=max_value,
max_time=max_time or "",
min=min_value,
min_time=min_time or "",
avg=avg_value,
)
else:
avg_value = ""
elec_index = ElecIndex(
stats_index=item_name,
max=max_value,
max_time=max_time or "",
min=min_value,
min_time=min_time or "",
avg=avg_value,
)
elec_index = ElecIndex(
stats_index=item,
max="",
max_time="",
min="",
min_time="",
avg="",
)
common_indexes.append(elec_index)
# 电能质量统计
elec_qual_indexes = []
_elec_qual_items = {i.rsplit("_", 1)[0] for i in elec_qual_items}
for item in _elec_qual_items:
item_name = item.rsplit("_", 1)[0]
max_item_name = f"{item}_max"
max_value = df[max_item_name].max()
if not pd.isna(max_value):
max_datas = df.loc[df[max_item_name].idxmax()].to_dict()
max_time = max_datas.get(f"{item_name}_max_time")
max_time = '' if pd.isnull(max_time) else str(max_time)
else:
max_value, max_time = "", ""
min_item_name = f"{item}_min"
min_value = df[min_item_name].min()
if not pd.isna(min_value):
min_datas = df.loc[df[min_item_name].idxmin()].to_dict()
min_time = min_datas.get(f"{item_name}_min_time")
min_time = '' if pd.isnull(min_time) else str(min_time)
else:
min_value, min_time = "", ""
mean_item_name = f"{item}_mean"
avg_value = df[mean_item_name].mean()
if not pd.isna(avg_value):
avg_value = round(avg_value, 2) if avg_value else ""
if datas:
max_item_name = f"{item}_max"
max_value = df[max_item_name].max()
if not pd.isna(max_value):
max_datas = df.loc[df[max_item_name].idxmax()].to_dict()
max_time = max_datas.get(f"{item}_max_time")
max_time = '' if pd.isnull(max_time) else str(max_time)
else:
max_value, max_time = "", ""
min_item_name = f"{item}_min"
min_value = df[min_item_name].min()
if not pd.isna(min_value):
min_datas = df.loc[df[min_item_name].idxmin()].to_dict()
min_time = min_datas.get(f"{item}_min_time")
min_time = '' if pd.isnull(min_time) else str(min_time)
else:
min_value, min_time = "", ""
mean_item_name = f"{item}_mean"
avg_value = df[mean_item_name].mean()
if not pd.isna(avg_value):
avg_value = round(avg_value, 2) if avg_value else ""
else:
avg_value = ""
elec_index = ElecIndex(
stats_index=item,
max=max_value,
max_time=max_time,
min=min_value,
min_time=min_time,
avg=avg_value,
)
else:
avg_value = ""
elec_index = ElecIndex(
stats_index=item_name,
max=max_value,
max_time=max_time,
min=min_value,
min_time=min_time,
avg=avg_value,
)
elec_index = ElecIndex(
stats_index=item,
max="",
max_time="",
min="",
min_time="",
avg="",
)
elec_qual_indexes.append(elec_index)
# 小程序需要这漏电流和温度
if cid:
......@@ -877,6 +896,7 @@ async def elec_index_service_new15(cid, point_id, start, end):
elec_qual_indexes=elec_qual_indexes
)
async def elec_current_service_new15(point_id):
# 获取mtid
meter_info = await get_meter_by_point_new15(point_id)
......@@ -903,7 +923,7 @@ async def elec_current_service_new15(point_id):
time_str = str(res["ts"])[0:19]
else:
time_str = time_format.get_datetime_str(0)
return time_str,res
return time_str, res
def get_sdu_i_and_u(res, ctnum):
......
unify_api/modules/home_page/procedures/count_info_pds.py
View file @ 61200aa0
......@@ -3,6 +3,10 @@ import time
from datetime import datetime, timedelta
from math import sqrt
import pendulum
from pot_libs.settings import SETTING
from unify_api.modules.electric.dao.electric_dao import \
get_elec_mtid_sid_by_cid
from unify_api.utils.common_utils import round_2
from pot_libs.aredis_util.aredis_utils import RedisUtils
from pot_libs.common.components.query import Range, Equal, Filter, PageRequest
......@@ -35,7 +39,9 @@ from unify_api.modules.home_page.dao.count_info_dao import (
from unify_api.modules.electric_optimization.dao.power_index import (
price_policy_by_cid
)
from unify_api.utils.taos_new import parse_td_columns
from unify_api.utils.taos_new import parse_td_columns, get_td_table_name, \
td3_tbl_compate, get_td_engine_data
async def other_info(company_id):
"""
......@@ -52,7 +58,7 @@ async def other_info(company_id):
GROUP BY
DATE(pevent.event_datetime)
"""
now_time = datetime.now()
# 获取到工厂安装时间create_time
async with MysqlUtil() as conn:
......@@ -61,7 +67,7 @@ async def other_info(company_id):
company = await conn.fetchone(company_sql, (company_id,))
create_time_timestamp = company["create_time"]
create_time = datetime.fromtimestamp(create_time_timestamp)
today_alarm_count = 0
alarm_count = 0
if not alarm_data:
......@@ -72,7 +78,7 @@ async def other_info(company_id):
# 系统安全运行天数： 当前时间 - 工厂安装时间 + 1
safe_run_days = (now_time - create_time).days + 1
return today_alarm_count, safe_run_days, alarm_count
# 5. 构造返回
# 如果每天都有报警, 防止安全运行天数-1天, 所以total_days +2
total_days = (now_time - create_time).days + 2
......@@ -119,20 +125,20 @@ async def other_info_old(company_id):
}
},
}
async with EsUtil() as es:
es_result = await es.search_origin(body=query_body,
index=constants.POINT_1MIN_EVENT)
now_time = datetime.now()
# 获取到工厂安装时间create_time
async with MysqlUtil() as conn:
company_sql = "select create_time from company where cid = %s"
company = await conn.fetchone(company_sql, (company_id,))
create_time_timestamp = company["create_time"]
create_time = datetime.fromtimestamp(create_time_timestamp)
today_alarm_count = 0
alarm_count = 0
date_buckets = es_result.get("aggregations", {}).get("date_alarms",
......@@ -146,7 +152,7 @@ async def other_info_old(company_id):
# 系统安全运行天数： 当前时间 - 工厂安装时间 + 1
safe_run_days = (now_time - create_time).days + 1
return today_alarm_count, safe_run_days, alarm_count
# 5. 构造返回
# 如果每天都有报警, 防止安全运行天数-1天, 所以total_days +2
total_days = (now_time - create_time).days + 2
......@@ -214,7 +220,7 @@ async def electric_use_info(company_id):
end_timestamp = datetime_to_timestamp(now)
start_timestamp = datetime_to_timestamp(
datetime(now.year, now.month, now.day) - timedelta(30))
score_events = [
i
for i in EVENT_TYPE_MAP.keys()
......@@ -256,12 +262,12 @@ async def electric_use_info(company_id):
},
},
}
log.info("cal_score_safe_electric query_body={}".format(query_body))
async with EsUtil() as es:
es_result = await es.search_origin(body=query_body,
index=constants.POINT_1MIN_EVENT)
score_buckets = (
es_result.get("aggregations", {}).get("score_aggs", {}).get("types",
{}).get(
......@@ -277,7 +283,7 @@ async def electric_use_info(company_id):
second_alarm_cnt += bucket["doc_count"]
elif bucket["key"] == Importance.Third.value:
third_alarm_cnt += bucket["doc_count"]
company_point_map = await get_points([company_id])
point_len = len(company_point_map.get(company_id) or {})
alarm_score = (
......@@ -290,15 +296,15 @@ async def electric_use_info(company_id):
f"alarm_score:{alarm_score}")
if alarm_score >= 15:
alarm_score = 15
electric_use_score = get_electric_index(alarm_score)
log.info(
"point_len={} alarm_score={} electric_use_score={}".format(
point_len, alarm_score, electric_use_score
)
)
alarm_buckets = (
es_result.get("aggregations", {}).get("alarm_aggs", {}).get("types",
{}).get(
......@@ -337,7 +343,7 @@ async def electric_use_info_new15(cid):
f"GROUP BY importance"
first_score, second_score, third_score = 0, 0, 0
async with MysqlUtil() as conn:
score_datas = await conn.fetchall(score_sql, args=(cid, ))
score_datas = await conn.fetchall(score_sql, args=(cid,))
alarm_datas = await conn.fetchall(alarm_sql, args=(cid, score_events))
for data in score_datas:
if data["importance"] == Importance.First.value:
......@@ -389,27 +395,27 @@ async def normal_rate_of_location(company_id):
)
for location in locations:
location_map[location["id"]] = location
# todo批量hmget
count_info_map = {
"residual_current": {"total": 0, "normal": 0},
"temperature": {"total": 0, "normal": 0},
}
print(f"len(location_map)={len(location_map)}")
location_ids = list(location_map.keys())
adio_currents = []
if location_ids:
adio_currents = await RedisUtils().hmget("adio_current",
*location_ids)
*location_ids)
adio_info_map = {}
for index, item_byte in enumerate(adio_currents):
if item_byte:
item = json.loads(item_byte.decode())
adio_info_map[location_ids[index]] = item
for location_id, location_info in location_map.items():
audio_info = adio_info_map.get(location_id)
count_info_map[location_info["type"]]["total"] += 1
......@@ -420,7 +426,7 @@ async def normal_rate_of_location(company_id):
# 超过4小时的值不统计在normal里
log.warn(f"adio_current location_id={location_id} has expire!")
continue
print(
"threshold={} location_info['type'] = {} audio_info['value']={}".format(
location_info["threshold"], location_info["type"],
......@@ -447,7 +453,7 @@ async def normal_rate_of_location(company_id):
)
+ "%"
)
if count_info_map["residual_current"]["total"] == 0:
residual_current_qr = "100%"
else:
......@@ -463,7 +469,7 @@ async def normal_rate_of_location(company_id):
)
+ "%"
)
return temperature_qr, residual_current_qr
......@@ -488,13 +494,13 @@ async def normal_rate_of_location_new15(cid):
adio_currents = []
if location_ids:
adio_currents = await RedisUtils().hmget("adio_current",
*location_ids)
*location_ids)
adio_info_map = {}
for index, item_byte in enumerate(adio_currents):
if item_byte:
item = json.loads(item_byte.decode())
adio_info_map[location_ids[index]] = item
for location_id, location_info in location_map.items():
audio_info = adio_info_map.get(location_id)
count_info_map[location_info["type"]]["total"] += 1
......@@ -525,7 +531,7 @@ async def normal_rate_of_location_new15(cid):
)
+ "%"
)
if count_info_map["residual_current"]["total"] == 0:
residual_current_qr = "100%"
else:
......@@ -541,7 +547,7 @@ async def normal_rate_of_location_new15(cid):
)
+ "%"
)
return temperature_qr, residual_current_qr
......@@ -556,10 +562,10 @@ async def current_load(company_id):
"and add_to_company = 1"
points = await conn.fetchall(point_sql, args=(company_id,))
point_ids = [p["pid"] for p in points]
if not point_ids:
return ""
async with MysqlUtil() as conn:
meter_sql = (
"SELECT pid, mid FROM change_meter_record WHERE pid in %s ORDER BY pid, start_time"
......@@ -569,15 +575,15 @@ async def current_load(company_id):
# 正序排序，最后这个map存储的是按照start_time是最近的mid
change_meter_map = {m["pid"]: m["mid"] for m in change_meters if
m["mid"] is not None}
newest_mids = list(change_meter_map.values())
meterdata_currents = []
if newest_mids:
meterdata_currents = await RedisUtils().hmget(METERDATA_CURRENT_KEY,
*newest_mids)
*newest_mids)
now_tt = int(time.time())
if meterdata_currents:
total = 0
for item in meterdata_currents:
......@@ -598,39 +604,46 @@ async def current_load(company_id):
return ""
async def current_load_new15(cid):
async def current_load_new15(cid, end_dt=None):
"""实时负荷"""
import re
from pot_libs.settings import SETTING
from unify_api.modules.common.service.td_engine_service import \
get_td_engine_data
datas = await get_elec_mtid_sid_by_cid(cid)
td_mt_tables = tuple(
(get_td_table_name("electric", data["mtid"]) for data in datas if
data["mtid"]))
td_mt_tables = td3_tbl_compate(td_mt_tables)
if not end_dt:
end_dt = pendulum.now(tz="Asia/Shanghai")
start_dt = end_dt.subtract(minutes=2)
sql = f"select last_row(mdptime, pttl) from electric_stb " \
f"where TBNAME IN {td_mt_tables} and ts>='{str(start_dt)}' and ts " \
f"<='{str(end_dt)}' group by tbname"
url = f"{SETTING.stb_url}db_electric?tz=Asia/Shanghai"
sql = f"""
select last_row(*) from electric_stb
where cpyid={cid}
group by tbname
"""
is_succ, results = await get_td_engine_data(url, sql)
now_tt = int(time.time())
if is_succ:
head = parse_td_columns(results)
datas = []
for res in results["data"]:
datas.append(dict(zip(head, res)))
total = 0
for item in datas:
# 这里是有可能item为None的
if item:
mdptime_tt = None
if "mdptime" in item:
mdptime_dt = pendulum.parse(item["mdptime"])
item_tt = item.get("timestamp") or mdptime_dt.int_timestamp
if item_tt:
# 小于2分钟内的数据相加为实时负荷
if now_tt - item_tt <= 2 * 60:
total += item["pttl"]
return total
return ""
if not is_succ:
return ""
if not results["data"]: # 兼容：mt表（2.0架构）里面拿不到数据再从sid表（1.0架构）里面拿
td_s_tables = tuple(
(f"s{data['sid'].lower()}_e" for data in datas if data["sid"]))
td_s_tables = td3_tbl_compate(td_s_tables)
sql = f"select last_row(mdptime, pttl) from electric_stb " \
f"where TBNAME IN {td_s_tables} group by tbname"
is_succ, results = await get_td_engine_data(url, sql)
if not is_succ:
return ""
head = parse_td_columns(results)
datas = []
for res in results["data"]:
datas.append(dict(zip(head, res)))
total = 0
for item in datas:
if not item:
continue
total += item["pttl"]
return total
async def power_count_info(company_id):
......@@ -644,7 +657,7 @@ async def power_count_info(company_id):
now = datetime.now()
start_time = (now - timedelta(30)).strftime("%Y-%m-%d %H:%M:%S")
end_time = now.strftime("%Y-%m-%d %H:%M:%S")
max_30d_load, _time = await pttl_max(company_id, start_time, end_time, -1)
cur_load = await current_load(company_id)
return cur_load, max_30d_load
......@@ -655,7 +668,7 @@ async def power_count_info_new15(cid):
now = datetime.now()
start_time = (now - timedelta(30)).strftime("%Y-%m-%d 00:00:00")
end_time = now.strftime("%Y-%m-%d %H:%M:%S")
max_30d_load, _time = await pttl_max_new15(cid, start_time, end_time, -1)
cur_load = await current_load_new15(cid)
return round_2(cur_load), round_2(max_30d_load)
......@@ -691,7 +704,7 @@ async def get_max_aiao_of_filed(company_id, start_time, end_time,
index=constants.LOCATION_15MIN_AIAO)
value_max = es_results.get("aggregations", {}).get("value_max_max", {})
rc_max_hits = value_max.get("hits", {}).get("hits")
max_info, location_map = {}, {}
if rc_max_hits:
max_info = rc_max_hits[0]["_source"]
......@@ -706,7 +719,7 @@ async def get_max_aiao_of_filed(company_id, start_time, end_time,
if max_info
else None
)
return MaxResidualCurrent(
max=round(max_info["value_max"], 2) if max_info else None,
location_name=f"{location_map['group']}_{'漏电流' if location_map['item'] == 'default' else location_map['item']}"
......@@ -769,8 +782,9 @@ async def company_power_use_info_new15(company_id, start, end):
async def get_company_charge_price(company_id, es_time_start, es_time_end):
power_use_info = await company_power_use_info_new15(company_id, es_time_start,
es_time_end)
power_use_info = await company_power_use_info_new15(company_id,
es_time_start,
es_time_end)
if power_use_info["kwh"]:
unit_price = power_use_info["charge"] / power_use_info["kwh"]
else:
......@@ -790,7 +804,7 @@ async def power_charge_price(company_id):
yestoday_start = datetime(yestoday.year, yestoday.month, yestoday.day, 0,
0, 0)
yestoday_end = yestoday_start + timedelta(1)
es_yestoday_start = datetime.strftime(yestoday_start,
"%Y-%m-%dT%H:%M:%S+08:00")
es_yestoday_end = datetime.strftime(yestoday_end,
......@@ -798,7 +812,7 @@ async def power_charge_price(company_id):
yestoday_price = await get_company_charge_price(company_id,
es_yestoday_start,
es_yestoday_end)
if now.month == 1:
last_month = 12
year = now.year - 1
......@@ -813,7 +827,7 @@ async def power_charge_price(company_id):
last_month_price = await get_company_charge_price(
company_id, es_last_month_start, es_last_month_end
)
return yestoday_price, last_month_price
......@@ -835,7 +849,7 @@ async def power_charge_price_new15(cid):
year = now.year - 1
last_month_start = datetime(year=year, month=last_month, day=1)
else:
last_month_start = datetime(year=now.year, month=now.month-1, day=1)
last_month_start = datetime(year=now.year, month=now.month - 1, day=1)
last_month_end = datetime(year=now.year, month=now.month, day=1)
last_month_datas = await company_power_use_info_new15(
cid, str(last_month_start), str(last_month_end)
......@@ -859,19 +873,19 @@ async def power_factor(company_id):
)
points = await conn.fetchall(point_sql, args=(company_id, 1))
point_ids = [i["pid"] for i in points]
now = datetime.now()
if now.month == 1:
last_month_dt = datetime(year=now.year - 1, month=12, day=1)
else:
last_month_dt = datetime(year=now.year, month=now.month - 1, day=1)
# 首页功率因数取所有进线中最小的
async with MysqlUtil() as conn:
sql = "SELECT inlid, `name` FROM inline WHERE cid=%s"
inlines = await conn.fetchall(sql, args=(company_id,))
inline_ids = [inline["inlid"] for inline in inlines]
power_factor_results = []
sql = "SELECT inlid, save_charge pf_cost, `kpi_x`, `save_charge` " \
"FROM algo_power_factor_result WHERE inlid in %s and month=%s"
......@@ -883,7 +897,7 @@ async def power_factor(company_id):
type(i["kpi_x"]) in [int, float]
]
last_month_cos = min(pf_kpi_x_list) if len(pf_kpi_x_list) else ""
async with EsUtil() as es:
dt = pendulum.now(tz="Asia/Shanghai")
tstamp = dt.int_timestamp // (15 * 60) * (15 * 60)
......@@ -913,18 +927,18 @@ async def power_factor(company_id):
"qttl_mean": item["qttl_mean"],
}
)
total_pttl, total_qttl = 0, 0
for point_id, records in point_map.items():
total_pttl += records[0]["pttl_mean"]
total_qttl += records[0]["qttl_mean"]
# 计算实时功率的公式
cos_ttl = ""
l = sqrt(total_pttl * total_pttl + total_qttl * total_qttl)
if l:
cos_ttl = round(total_pttl / l, 2)
if type(last_month_cos) in [int, float]:
last_month_cos = round(last_month_cos, 2)
return cos_ttl, last_month_cos
......@@ -934,7 +948,7 @@ async def power_factor_new15(cid):
"""首页获取实时功率因数， 上月功率因数"""
point_sql = "select pid,inlid from point where cid=%s and add_to_company=1"
async with MysqlUtil() as conn:
points = await conn.fetchall(point_sql, args=(cid, ))
points = await conn.fetchall(point_sql, args=(cid,))
point_ids = [i["pid"] for i in points]
now = datetime.now()
if now.month == 1:
......@@ -957,11 +971,11 @@ async def power_factor_new15(cid):
type(i["kpi_x"]) in [int, float]
]
last_month_cos = min(pf_kpi_x_list) if len(pf_kpi_x_list) else ""
dt = pendulum.now(tz="Asia/Shanghai")
tstamp = dt.int_timestamp // (15 * 60) * (15 * 60)
dt = pendulum.from_timestamp(tstamp, tz="Asia/Shanghai")
end_dt = (dt-timedelta(minutes=15)).strftime("%Y-%m-%d %H:%M:%S")
end_dt = (dt - timedelta(minutes=15)).strftime("%Y-%m-%d %H:%M:%S")
str_dt = dt.strftime("%Y-%m-%d %H:%M:%S")
electric_sql = f"SELECT pid,create_time,pttl_mean,qttl_mean FROM " \
f"`point_15min_electric` where create_time in " \
......@@ -985,7 +999,7 @@ async def power_factor_new15(cid):
l = sqrt(total_pttl * total_pttl + total_qttl * total_qttl)
if l:
cos_ttl = round(total_pttl / l, 2)
if type(last_month_cos) in [int, float]:
last_month_cos = round(last_month_cos, 2)
return cos_ttl, last_month_cos
......@@ -997,12 +1011,12 @@ async def optimization_count_info(company_id: int):
:param company_id:
:return:
"""
async with MysqlUtil() as conn:
sql = "SELECT inlid, `name` FROM inline WHERE cid=%s"
inlines = await conn.fetchall(sql, args=(company_id,))
inline_ids = [inline["inlid"] for inline in inlines]
# 获取公司上月用电
# now = datetime.now()
# es_start_time = (
......@@ -1014,12 +1028,12 @@ async def optimization_count_info(company_id: int):
# "%Y-%m-%dT%H:%M:%S+08:00")
# power_use_info = await company_power_use_info(company_id, es_start_time,
# es_end_time)
now = datetime.now()
start_time = (
pendulum.datetime(now.year, now.month, 1)
.subtract(months=1)
.strftime("%Y-%m-%d %H:%M:%S")
.subtract(months=1)
.strftime("%Y-%m-%d %H:%M:%S")
)
end_time = pendulum.datetime(now.year, now.month, 1).strftime(
"%Y-%m-%d %H:%M:%S")
......@@ -1043,13 +1057,13 @@ async def optimization_count_info(company_id: int):
}
)
return count_info_map
now = datetime.now()
if now.month == 1:
last_month_dt = datetime(year=now.year - 1, month=12, day=1)
else:
last_month_dt = datetime(year=now.year, month=now.month - 1, day=1)
# 功率因数
async with MysqlUtil() as conn:
sql = "SELECT inlid, `cos`, save_charge pf_cost, kpi_x, save_charge " \
......@@ -1063,7 +1077,7 @@ async def optimization_count_info(company_id: int):
2,
)
total_pf_save = 0 if total_pf_save <= 0 else total_pf_save
pf_kpi_x_list = [
i["kpi_x"] for i in power_factor_results if
type(i["kpi_x"]) in [int, float]
......@@ -1079,20 +1093,20 @@ async def optimization_count_info(company_id: int):
pf_desc = "空间适中"
else:
pf_desc = "空间较大"
count_info_map["power_factor"] = {
"save_charge": total_pf_save if pf_kpi_x != "" else "",
"kpi_x": pf_kpi_x,
"desc": pf_desc,
}
# 移峰填谷指数
async with MysqlUtil() as conn:
sql = "select `score`, `cost_save` from `algo_plsi_result` " \
"where `inlid` in %s and `month` = %s"
last_month_str = datetime.strftime(last_month_dt, "%Y-%m")
pcvfs = await conn.fetchall(sql, args=(inline_ids, last_month_str))
pcvf_kpi_x_list = [i["score"] for i in pcvfs if
type(i["score"]) in [int, float]]
pcvf_kpi_x = min(pcvf_kpi_x_list) if len(pcvf_kpi_x_list) else ""
......@@ -1100,7 +1114,7 @@ async def optimization_count_info(company_id: int):
sum([i["cost_save"] for i in pcvfs if
i["cost_save"] and i["cost_save"] >= 0]), 2
)
if pcvf_kpi_x == "":
pcvf_desc = ""
elif pcvf_kpi_x >= 90:
......@@ -1111,14 +1125,14 @@ async def optimization_count_info(company_id: int):
pcvf_desc = "空间适中"
else:
pcvf_desc = "空间较大"
total_pcvf_save = 0 if total_pcvf_save <= 0 else total_pcvf_save
count_info_map["pcvf"] = {
"save_charge": total_pcvf_save if pcvf_kpi_x != "" else "",
"kpi_x": pcvf_kpi_x,
"desc": pcvf_desc,
}
# 经济运行
async with MysqlUtil() as conn:
sql = "select `kpi_x`, `save_charge`, `mean_load_factor` " \
......@@ -1155,13 +1169,13 @@ async def optimization_count_info(company_id: int):
economic_desc = "空间适中"
else:
economic_desc = "空间较大"
count_info_map["power_save"] = {
"save_charge": total_economic_save if economic_kpi_x != "" else "",
"kpi_x": economic_kpi_x,
"desc": economic_desc,
}
# 最大需量
async with MysqlUtil() as conn:
sql = (
......@@ -1173,7 +1187,7 @@ async def optimization_count_info(company_id: int):
)
last_month_str = datetime.strftime(last_month_dt, "%Y-%m")
md_spaces = await conn.fetchall(sql, args=(inline_ids, last_month_str))
md_space_kpi_x_list = [i["kpi_x"] for i in md_spaces if
type(i["kpi_x"]) in [int, float]]
md_space_kpi_x = max(md_space_kpi_x_list) if len(
......@@ -1202,7 +1216,7 @@ async def optimization_count_info(company_id: int):
"kpi_x": md_space_kpi_x,
"desc": md_space_desc,
}
total_save_cost = 0
for _, item in count_info_map.items():
total_save_cost += (
......@@ -1211,12 +1225,12 @@ async def optimization_count_info(company_id: int):
)
save_percent = total_save_cost / month_charge if month_charge else ""
count_info_map["save_percent"] = save_percent
# 计算最大需量
async with MysqlUtil() as conn:
sql = "select `price_md`,`price_tc` from `price_policy` where `cid`=%s"
price_policy = await conn.fetchone(sql, args=(company_id,))
total_md_space_charge = sum(
[i["inline_md_charge"] for i in md_spaces if i["inline_md_charge"]])
total_md_space_p = (
......@@ -1225,7 +1239,7 @@ async def optimization_count_info(company_id: int):
else ""
)
count_info_map["md_space_p"] = total_md_space_p
# 经济运行最低负载率
mean_load_factors = [
i["mean_load_factor"] for i in economic_operations if
......@@ -1260,12 +1274,12 @@ async def electric_use_info_sdu(cid):
}
}
}
log.info("cal_score_safe_electric query_body={}".format(query_body))
async with EsUtil() as es:
es_result = await es.search_origin(body=query_body,
index=constants.POINT_1MIN_EVENT)
score_buckets = (
es_result.get("aggregations", {}).get("alarm_aggs", {}).get("buckets",
[])
......@@ -1280,7 +1294,7 @@ async def electric_use_info_sdu(cid):
second_alarm_cnt += bucket["doc_count"]
elif bucket["key"] == Importance.Third.value:
third_alarm_cnt += bucket["doc_count"]
company_point_map = await get_points([cid])
point_len = len(company_point_map.get(cid) or {})
alarm_score = (
......@@ -1291,9 +1305,9 @@ async def electric_use_info_sdu(cid):
)
if alarm_score >= 15:
alarm_score = 15
electric_use_score = get_electric_index(alarm_score)
log.info(
"point_len={} alarm_score={} electric_use_score={}".format(
point_len, alarm_score, electric_use_score
......@@ -1366,12 +1380,12 @@ async def electric_use_info_points_sdu(start, end, points):
}
}
}
log.info("electric_use_info_points query_body={}".format(query_body))
async with EsUtil() as es:
es_result = await es.search_origin(body=query_body,
index=constants.POINT_1MIN_EVENT)
score_buckets = (
es_result.get("aggregations", {}).get("alarm_aggs", {}).get("buckets",
[])
......@@ -1386,15 +1400,15 @@ async def electric_use_info_points_sdu(start, end, points):
second_alarm_cnt += bucket["doc_count"]
elif bucket["key"] == Importance.Third.value:
third_alarm_cnt += bucket["doc_count"]
alarm_score = (first_alarm_cnt * 2 + second_alarm_cnt * 1 +
third_alarm_cnt * 0.5) / len(points)
if alarm_score >= 15:
alarm_score = 15
electric_use_score = get_electric_index(alarm_score)
log.info(
"point_len={} alarm_score={} electric_use_score={}".format(
len(points), alarm_score, electric_use_score
......@@ -1415,9 +1429,9 @@ async def electric_use_info_points_sdu_new15(start, end, points):
f"where pid in %s and event_datetime BETWEEN %s and %s " \
f"and event_type in %s group by importance"
async with MysqlUtil() as conn:
results = await conn.fetchall(sql,args=(points,start,end,
SDU_ALARM_LIST))
results = await conn.fetchall(sql, args=(points, start, end,
SDU_ALARM_LIST))
first_alarm_cnt = 0
second_alarm_cnt = 0
third_alarm_cnt = 0
......@@ -1428,15 +1442,15 @@ async def electric_use_info_points_sdu_new15(start, end, points):
second_alarm_cnt += result["doc_count"]
elif result["importance"] == Importance.Third.value:
third_alarm_cnt += result["doc_count"]
alarm_score = (first_alarm_cnt * 2 + second_alarm_cnt * 1 +
third_alarm_cnt * 0.5) / len(points)
if alarm_score >= 15:
alarm_score = 15
electric_use_score = get_electric_index(alarm_score)
log.info(
"point_len={} alarm_score={} electric_use_score={}".format(
len(points), alarm_score, electric_use_score
......@@ -1457,21 +1471,22 @@ async def optimization_count_info_new(company_id: int):
:param company_id:
:return:
"""
inlines = await get_inline_by_cid(company_id)
inline_ids = [inline["inlid"] for inline in inlines]
# 获取公司上月用电
now = datetime.now()
es_start_time = (
pendulum.datetime(now.year, now.month, 1)
.subtract(months=1)
.strftime("%Y-%m-%dT%H:%M:%S+08:00")
.subtract(months=1)
.strftime("%Y-%m-%dT%H:%M:%S+08:00")
)
es_end_time = pendulum.datetime(now.year, now.month, 1).strftime(
"%Y-%m-%dT%H:%M:%S+08:00")
power_use_info = await company_power_use_info_new15(company_id, es_start_time,
es_end_time)
power_use_info = await company_power_use_info_new15(company_id,
es_start_time,
es_end_time)
month_charge = power_use_info["charge"]
count_info_map = {
"avg_price": power_use_info["charge"] / power_use_info["kwh"]
......@@ -1497,7 +1512,7 @@ async def optimization_count_info_new(company_id: int):
last_month_dt = datetime(year=now.year, month=now.month - 1, day=1)
last_month_str = datetime.strftime(last_month_dt, "%Y-%m")
# 功率因数
power_factor_results = await get_power_factor_kpi(inline_ids,
last_month_dt)
total_pf_save = round(
......@@ -1510,7 +1525,7 @@ async def optimization_count_info_new(company_id: int):
(i["name"], i["kpi_x"]) for i in power_factor_results if
type(i["kpi_x"]) in [int, float]
]
if len(pf_kpi_x_list):
pf_kpi_x_num = [pf_kpi[1] for pf_kpi in pf_kpi_x_list]
pf_kpi_x = min(pf_kpi_x_num)
......@@ -1545,12 +1560,12 @@ async def optimization_count_info_new(company_id: int):
pcvfs = await get_pcvf_kpi(inline_ids, last_month_str)
pcvf_kpi_x_list = [(i["name"], i["score"]) for i in pcvfs if
type(i["score"]) in [int, float]]
if len(pcvf_kpi_x_list):
pcvf_kpi_x_num = [pcvf_kpi[1] for pcvf_kpi in pcvf_kpi_x_list]
pcvf_kpi_x = min(pcvf_kpi_x_num)
pcvf_kpi_x_name = []
if pcvf_kpi_x < 70:
for index, kpi_num in enumerate(pcvf_kpi_x_num):
if kpi_num < 70:
......@@ -1564,7 +1579,7 @@ async def optimization_count_info_new(company_id: int):
f"引入新能源，转移高峰电量至低谷"
else:
pcvf_desc = "平均电价处于较低水平，请继续保持"
else:
pcvf_kpi_x = ""
pcvf_desc = ""
......@@ -1582,7 +1597,7 @@ async def optimization_count_info_new(company_id: int):
sum([i["cost_save"] for i in pcvfs if
i["cost_save"] and i["cost_save"] >= 0]), 2
)
total_pcvf_save = 0 if total_pcvf_save <= 0 else total_pcvf_save
count_info_map["pcvf"] = {
"save_charge": total_pcvf_save if pcvf_kpi_x != "" else "",
......@@ -1590,7 +1605,7 @@ async def optimization_count_info_new(company_id: int):
"desc": pcvf_desc,
"space": pcvf_space
}
# 经济运行
economic_operations = await get_economic_kpi(inline_ids, last_month_str)
economic_kpi_x_list = [
......@@ -1651,14 +1666,14 @@ async def optimization_count_info_new(company_id: int):
"desc": economic_desc,
"space": econ_space
}
# 容量、需量价格
price_policy = await price_policy_by_cid(company_id)
price_md = price_policy["price_md"] if price_policy["price_md"] else 0
price_tc = price_policy["price_tc"] if price_policy["price_tc"] else 0
# 最大需量
md_spaces = await get_md_space(inline_ids, last_month_dt)
md_space_kpi_x_list = [i["kpi_x"] for i in md_spaces if
type(i["kpi_x"]) in [int, float]]
md_space_kpi_x = max(md_space_kpi_x_list) if len(
......@@ -1688,13 +1703,13 @@ async def optimization_count_info_new(company_id: int):
md_space_tc_runtimes[index]["tc_runtime"] * price_tc >= \
price_md * item["inline_md_predict"]:
md_space_name.append(md_space_tc_runtimes[index]["name"])
if len(md_space_name):
md_space_desc = f"若次月负荷无较大变动，建议{'、'.join(md_space_name)}" \
f"选择按最大需量计费"
else:
md_space_desc = "不存在容改需空间"
count_info_map["md_space"] = {
"save_charge": total_md_space_save if md_space_kpi_x != "" else "",
"kpi_x": md_space_kpi_x,
......@@ -1736,7 +1751,7 @@ async def cid_alarm_importance_count(cid, start, end):
point_list = [i["pid"] for i in monitor_point_list]
es_res = await sdu_alarm_importance_dao_new15(start, end, point_list)
es_res_key = {i["key"]: i for i in es_res}
res_list = []
for info in monitor_point_list:
name = info.get("name")
......@@ -1752,7 +1767,7 @@ async def cid_alarm_importance_count(cid, start, end):
tmp_dic["second"] += b["doc_count"]
elif b["key"] == Importance.Third.value:
tmp_dic["third"] += b["doc_count"]
tmp_dic["alarm_count"] = tmp_dic["first"] + tmp_dic["second"] + \
tmp_dic[
"third"]
......
unify_api/modules/load_analysis/dao/load_forecast_dao.py
View file @ 61200aa0
from pot_libs.mysql_util.mysql_util import MysqlUtil
async def get_kwh_p_dao(table_name, terms, start, end):
sql = f"SELECT create_time,kwh,p FROM {table_name} WHERE cid in %s " \
f"and create_time BETWEEN '{start}' and '{end}'"
async def get_kwh_p_dao(terms, start_time, end_time,
table_name="company_15min_power",
time_fmt="%%Y-%%m-%%d"):
"""
负荷实际数据
:param terms:
:param start_time:
:param end_time:
:param table_name:
:param time_fmt:
:return:
"""
sql = f"""
select p,kwh,DATE_FORMAT(create_time,"{time_fmt}") as cal_time
from {table_name} where cid in %s and create_time >= %s
and create_time <= %s
"""
async with MysqlUtil() as conn:
datas = await conn.fetchall(sql, args=(terms, ))
return datas
result = await conn.fetchall(sql, args=(terms, start_time, end_time))
return result or []
async def get_pred_p_dao(terms, start, end):
sql = f"SELECT create_time,p FROM company_day_ahead_predict " \
f"WHERE cid in %s and create_time BETWEEN '{start}' and '{end}'"
async def get_pred_p_dao(terms, start_time, end_time,
time_fmt="%%Y-%%m-%%d"):
"""
负荷预测数据
:param terms:
:param start_time:
:param end_time:
:param time_fmt:
:return:
"""
sql = f"""
select avg(p) p ,count(*) p_count,DATE_FORMAT(create_time,
"{time_fmt}") as cal_time
from company_day_ahead_predict
where cid in %s and create_time >= %s and create_time <= %s
group by cal_time
"""
async with MysqlUtil() as conn:
datas = await conn.fetchall(sql, args=(terms,))
return datas
\ No newline at end of file
result = await conn.fetchall(sql, args=(terms, start_time, end_time))
return result or []
unify_api/modules/load_analysis/service/load_forecast_service.py
View file @ 61200aa0
from datetime import datetime
import pendulum
from pot_libs.es_util.es_utils import EsUtil
from pot_libs.logger import log
from unify_api.constants import COMPANY_15MIN_POWER, COMPANY_DAY_AHEAD_PRE
......@@ -30,7 +33,7 @@ async def load_forecast_service(cid, cids, start, end):
trans_type = "15m"
interval = trans_type
format = "yyyy-MM-dd HH:mm:ss"
real_query["aggs"] = {
"quarter_time": {
"date_histogram": {
......@@ -39,7 +42,8 @@ async def load_forecast_service(cid, cids, start, end):
"time_zone": "+08:00",
"format": format,
},
"aggs": {"p": {"stats": {"field": "p"}}, "kwh": {"stats": {"field": "kwh"}}},
"aggs": {"p": {"stats": {"field": "p"}},
"kwh": {"stats": {"field": "kwh"}}},
}
}
real_index = COMPANY_15MIN_POWER
......@@ -70,14 +74,16 @@ async def load_forecast_service(cid, cids, start, end):
real_value = real_re[slot].get("p").get("sum")
real_power_val = (
real_re[slot].get("kwh").get("sum")
if type(real_re[slot].get("kwh").get("sum")) in [int, float]
if type(real_re[slot].get("kwh").get("sum")) in [int,
float]
else ""
)
else:
real_value = real_re[slot].get("p").get("sum")
real_power_val = (
real_re[slot].get("kwh").get("sum")
if type(real_re[slot].get("kwh").get("sum")) in [int, float]
if type(real_re[slot].get("kwh").get("sum")) in [int,
float]
else ""
)
real_power_list.append(real_power_val)
......@@ -160,8 +166,9 @@ async def load_forecast_service(cid, cids, start, end):
pred_power_list=LoadValue(slots=slots, value=pred_power_list),
deviation_power_list=[],
)
count, maxnum, minnum, average, max_index, min_index = choose_list(deviation_list_tmp, 4)
count, maxnum, minnum, average, max_index, min_index = choose_list(
deviation_list_tmp, 4)
# 求最大偏差和最小偏差时间
max_t = slots[max_index]
min_t = slots[min_index]
......@@ -183,7 +190,7 @@ async def load_forecast_service(cid, cids, start, end):
pred_tmp = [i for i in pred_tmp if i != ""]
total_deviation = abs((sum(pred_tmp) - sum(real_tmp)) / sum(real_tmp))
total_deviation = round(total_deviation, 4)
deviation_power_list = []
for index, real_power in enumerate(real_power_list):
if real_power == "" or pred_power_list[index] == "" or real_power == 0:
......@@ -192,16 +199,17 @@ async def load_forecast_service(cid, cids, start, end):
# (预测-实际)/实际 * 100%
deviation = (pred_power_list[index] - real_power) / real_power
deviation_power_list.append(abs(round(deviation, 4)))
if (
start.split(" ")[0].rsplit("-", 1)[0] == str(datetime.now().date()).rsplit("-", 1)[0]
and trans_type == "day"
start.split(" ")[0].rsplit("-", 1)[0] ==
str(datetime.now().date()).rsplit("-", 1)[0]
and trans_type == "day"
):
# 如果是本月的,那么当天的电量偏差没有意义, 置为空
today_str = str(datetime.now().date()).split("-", 1)[1]
index = slots.index(today_str)
deviation_power_list[index] = ""
return ForecastResp(
pred_data=lv_pred,
real_data=lv_real,
......@@ -216,150 +224,139 @@ async def load_forecast_service(cid, cids, start, end):
)
async def load_forecast_service_new15(cid, cids, start, end):
terms = cids if cids else [cid]
time_type = interval_type("range", start, end)
if time_type == "day":
table_name = "company_1day_power"
fmt = "%m-%d"
async def load_forecast_service_new15(cids, start_time, end_time, interval,
slots):
if interval == 15 * 60:
# 按日
time_fmt = "%%Y-%%m-%%d %%H:%%i"
now_time_fmt = "YYYY-MM-DD HH:mm"
to_fmt = "HH:mm"
real_table_name = "company_15min_power"
elif interval == 86400:
# 按月
time_fmt = "%%Y-%%m-%%d"
now_time_fmt = "YYYY-MM-DD"
to_fmt = now_time_fmt
real_table_name = "company_1day_power"
else:
table_name = "company_15min_power"
fmt = "%H:%M"
datas = await get_kwh_p_dao(table_name, terms, start, end)
if not datas:
return ForecastResp()
# 获取slots
intervel, slots = time_pick_transf(start, end)
real_re = sql_time_process(datas, "create_time", fmt)
lv_real = LoadValue()
lv_real.slots = slots
real_list = []
real_power_list, pred_power_list = [], []
# 按年
time_fmt = "%%Y-%%m"
now_time_fmt = "YYYY-MM"
to_fmt = now_time_fmt
real_table_name = "company_1day_power"
now_time = pendulum.now().format(now_time_fmt)
now_month = pendulum.now().format("YYYY-MM")
# 1,获取实际数据
real_result = await get_kwh_p_dao(cids, start_time, end_time,
real_table_name, time_fmt)
# if not real_result:
# return [], [], [], [], [], [], [], 0, "", 0, "", 0, 0
real_data = {i["cal_time"]: i for i in real_result}
real_list, real_power_list = [], []
for slot in slots:
if slot in real_re:
real_value = real_re[slot].get("p")
real_power_val = (
real_re[slot].get("kwh")
if type(real_re[slot].get("kwh")) in [int, float] else ""
)
real_power_list.append(real_power_val)
# 值为0是正常数据
if real_value in [0, 0.0]:
real_list.append(0.0)
if slot in real_data:
p = real_data[slot]["p"]
kwh = real_data[slot]["kwh"]
if slot > now_time:
value, kwh = "", ""
elif p in [0, 0.0]:
value, kwh = 0.0, 0.0
elif p:
value = round(p, 2)
kwh = round(kwh, 2)
else:
real_list.append(round(real_value, 2) if real_value else "")
value, kwh = "", ""
else:
real_list.append("")
real_power_list.append("")
lv_real.value = real_list
# 3. 预测数据
pred_re = await get_pred_p_dao(terms, start, end)
if not pred_re:
return ForecastResp()
pred_re = sql_time_process(pred_re, "create_time", fmt)
lv_pred = LoadValue()
lv_pred.slots = slots
pred_list = []
value, kwh = "", ""
real_list.append(value)
real_power_list.append(kwh)
# 2,获取预测数据
forecast_result = await get_pred_p_dao(cids, start_time, end_time,
time_fmt)
# if not forecast_result:
# return [], [], [], [], [], [], [], 0, "", 0, "", 0, 0
forecast_data = {i["cal_time"]: i for i in forecast_result}
forecast_list, forecast_power_list = [], []
for slot in slots:
if slot in pred_re:
pred_value = pred_re[slot].get("p")
pred_power_val = (
round(pred_re[slot].get("p") * 0.25, 2)
if type(pred_re[slot].get("p")) in [int, float] else ""
)
pred_power_list.append(pred_power_val)
# 值为0是正常数据
if pred_value in [0, 0.0]:
pred_list.append(0.0)
if slot in forecast_data:
p = forecast_data[slot]["p"]
if p in [0, 0.0]:
value = 0.0
elif p:
value = round(p, 2)
else:
pred_list.append(round(pred_value, 2) if pred_value else "")
value = ""
else:
pred_list.append("")
pred_power_list.append("")
lv_pred.value = pred_list
# 4.求偏差数据
deviation_list = [] # 偏差
deviation_list_abs = [] # 偏差取绝对值, 最大/最小/平均偏差都是绝对值后数据
value = ""
forecast_list.append(value)
if value or value == 0.0:
p_count = forecast_data[slot]["p_count"]
power_value = round(value * 0.25 * p_count, 2)
else:
power_value = ""
forecast_power_list.append(power_value)
# 3.求偏差数据
deviation_list, deviation_power_list = [], [] # 偏差取绝对值, 最大/最小/平均偏差都是绝对值后数据
for num, value in enumerate(real_list):
if not value or not pred_list[num]:
if not value or not forecast_list[num]:
deviation_list.append("")
else:
# (预测-实际)/实际 * 100%
deviation = (pred_list[num] - value) / value
deviation_list.append(round(deviation, 4))
deviation_list_abs.append(abs(round(deviation, 4)))
# 取绝对值,并保留
deviation_list_tmp = [i for i in deviation_list_abs if i != ""]
log.info(f"deviation_list_tmp:{deviation_list_tmp}, "
f"deviation_list_abs:{deviation_list_abs}")
if not deviation_list_tmp:
return ForecastResp(
pred_data=lv_pred,
real_data=lv_real,
deviation_list=deviation_list_abs,
max_deviation=[],
min_deviation=[],
avg_deviation="",
total_deviation="",
real_power_list=LoadValue(slots=slots, value=real_power_list),
pred_power_list=LoadValue(slots=slots, value=pred_power_list),
deviation_power_list=[],
)
count, maxnum, minnum, average, max_index, min_index = choose_list(
deviation_list_tmp, 4)
# 求最大偏差和最小偏差时间
max_t = slots[max_index]
min_t = slots[min_index]
if time_type == "day":
max_time = start[:4] + "-" + max_t
min_time = start[:4] + "-" + min_t
else:
max_time = start.split(" ")[0] + " " + max_t
min_time = start.split(" ")[0] + " " + min_t
# 最大偏差
max_deviation = [maxnum, max_time]
# 最小偏差
min_deviation = [minnum, min_time]
# 平均偏差
avg_deviation = average
# 总量偏差 = | (预测(总) - 实际(总)) / 实际(总) | * 100%
real_tmp = [i for i in real_list if i != ""]
pred_tmp = pred_list[: len(real_tmp)]
pred_tmp = [i for i in pred_tmp if i != ""]
total_deviation = abs((sum(pred_tmp) - sum(real_tmp)) / sum(real_tmp))
total_deviation = round(total_deviation, 4)
deviation_power_list = []
for index, real_power in enumerate(real_power_list):
if real_power == "" or pred_power_list[index] == "" or real_power == 0:
deviation_power_list.append("")
deviation = (forecast_list[num] - value) / value
deviation_list.append(abs(round(deviation, 4)))
for num, value in enumerate(real_power_list):
if not value or not forecast_power_list[num]:
deviation_power = ""
else:
# (预测-实际)/实际 * 100%
deviation = (pred_power_list[index] - real_power) / real_power
deviation_power_list.append(abs(round(deviation, 4)))
if (
start.split(" ")[0].rsplit("-", 1)[0] ==
str(datetime.now().date()).rsplit("-", 1)[0]
and time_type == "day"
):
deviation_power = (forecast_power_list[num] - value) / value
deviation_power = abs(round(deviation_power, 4))
# 如果是本月的,那么当天的电量偏差没有意义, 置为空
today_str = str(datetime.now().date()).split("-", 1)[1]
index = slots.index(today_str)
deviation_power_list[index] = ""
if interval == 86400 and now_month == start_time[:7] and \
num == slots.index(now_time):
deviation_power = ""
deviation_power_list.append(deviation_power)
# 4,求偏差统计值
deviation_list_tmp = [i for i in deviation_list if i != ""]
if deviation_list_tmp:
count, max_deviation, min_deviation, average, max_index, min_index = \
choose_list(deviation_list_tmp, 4)
# 求最大偏差和最小偏差时间
max_time = slots[max_index]
min_time = slots[min_index]
# 平均偏差
avg_deviation = average
# 总量偏差 = | (预测(总) - 实际(总)) / 实际(总) | * 100%
real_tmp = [i for i in real_list if i != ""]
pred_tmp = forecast_list[: len(real_tmp)]
pred_tmp = [i for i in pred_tmp if i != ""]
total_deviation = abs((sum(pred_tmp) - sum(real_tmp)) / sum(real_tmp))
total_deviation = round(total_deviation, 4)
else:
max_deviation, max_time, min_deviation, min_time = 0, "", 0, ""
avg_deviation, total_deviation = 0, 0
# 重置 slots
if interval == 15 * 60:
for num, slot in enumerate(slots):
slot = pendulum.parse(slot).format("HH:mm")
slots[num] = slot
return ForecastResp(
pred_data=lv_pred,
real_data=lv_real,
deviation_list=deviation_list_abs,
pred_data=LoadValue(slots=slots, value=forecast_list),
real_data=LoadValue(slots=slots, value=real_list),
deviation_list=deviation_list,
max_deviation=max_deviation,
min_deviation=min_deviation,
avg_deviation=avg_deviation,
total_deviation=total_deviation,
real_power_list=LoadValue(slots=slots, value=real_power_list),
pred_power_list=LoadValue(slots=slots, value=pred_power_list),
pred_power_list=LoadValue(slots=slots, value=forecast_power_list),
deviation_power_list=deviation_power_list,
)
unify_api/modules/load_analysis/views/load_forecast.py
View file @ 61200aa0
......@@ -5,6 +5,7 @@ from unify_api.modules.load_analysis.components.load_forecast_cps import (
)
from unify_api.modules.load_analysis.service.load_forecast_service import \
load_forecast_service, load_forecast_service_new15
from unify_api.utils.time_format import time_pick_transf_new
@summary("负荷预测")
......@@ -17,4 +18,8 @@ async def post_load_forecast(req, body: ForecastReq) -> ForecastResp:
# 管理版本多个工厂的情况, 兼容能力最强的参数cids， 保留旧有的cid:
cids = body.cids
# return await load_forecast_service(cid, cids, start, end)
return await load_forecast_service_new15(cid, cids, start, end)
terms = cids if cids else [cid]
# 获取时间差
interval, slots = time_pick_transf_new(start, end)
return await load_forecast_service_new15(terms, start, end,
interval, slots)
unify_api/utils/time_format.py
View file @ 61200aa0
......@@ -60,6 +60,41 @@ def time_pick_transf(start, end, is_range=0):
return intervel, slots
def time_pick_transf_new(start, end):
"""获取intervel和slots, 详细显示时间轴信息，新接口都使用这个来获取时间"""
start_f = my_pendulum.from_format(start, 'YYYY-MM-DD HH:mm:ss')
end_f = my_pendulum.from_format(end, 'YYYY-MM-DD HH:mm:ss')
diff = end_f.int_timestamp - start_f.int_timestamp
# 1. 计算intervel
# 1.1 区间48小时之内, 返回15min
if diff <= 48 * 3600:
intervel = 15 * 60
# 1.2 区间在60天以内, 返回1day
elif 48 * 3600 < diff <= 60 * 86400:
intervel = 86400
# 1.3 选择年, 返回1个月
else:
intervel = 30 * 86400
# 2. 计算slots
# 2.1 取到点的个数, 比如15min的96个点
slots = []
slot_num = round((end_f.int_timestamp - start_f.int_timestamp) / intervel)
for i in range(slot_num):
# 区间48小时之内
if diff <= 48 * 3600:
dt = start_f.add(minutes=15 * i).format("YYYY-MM-DD HH:mm")
dt_str = str(dt)
# 区间在60天以内
elif 48 * 3600 < diff <= 60 * 86400:
dt = start_f.add(days=1 * i).format("YYYY-MM-DD")
dt_str = str(dt)
else:
dt = start_f.add(months=1 * i).format("YYYY-MM")
dt_str = str(dt)
slots.append(dt_str)
return intervel, slots
def power_slots(start, end):
"""电量电费,用电统计,time=range slots计算"""
start_f = my_pendulum.from_format(start, 'YYYY-MM-DD HH:mm:ss')
......@@ -115,6 +150,7 @@ def proxy_power_slots(start, end, date_format="MM-DD", is_duration=False):
slots.append(dt)
return slots
def day_of_month(start):
"""这个月有几天"""
start_f = my_pendulum.from_format(start, 'YYYY-MM-DD HH:mm:ss')
......@@ -269,7 +305,7 @@ def start_end_date():
def time_str_to_str(date_str, format="HH:mm"):
"""YYYY-MM-DD HH:mm:ss格式转换为format格式"""
start_f = my_pendulum.from_format(date_str, 'YYYY-MM-DD HH:mm:ss')
start_f = my_pendulum.parse(date_str)
start_f = start_f.format(format)
return start_f
......@@ -311,7 +347,7 @@ def convert_to_es_str(str1, format="YYYY-MM-DD HH:mm:ss"):
return str(es_date)
def last_time_str(start, end, date_type):
def last_time_str(start, end, date_type, date_end=False):
"""年月日, 获取上一周期时间"""
if date_type not in ("day", "month", "year"):
return None
......@@ -319,13 +355,22 @@ def last_time_str(start, end, date_type):
end_f = my_pendulum.from_format(end, 'YYYY-MM-DD HH:mm:ss')
if date_type == "day":
start_last = start_f.subtract(days=1)
end_last = end_f.subtract(days=1)
if date_end:
end_last = start_last.end_of(unit=date_type)
else:
end_last = end_f.subtract(days=1)
elif date_type == "month":
start_last = start_f.subtract(months=1)
end_last = end_f.subtract(months=1)
if date_end:
end_last = start_last.end_of(unit=date_type)
else:
end_last = end_f.subtract(months=1)
else:
start_last = start_f.subtract(years=1)
end_last = end_f.subtract(years=1)
if date_end:
end_last = start_last.end_of(unit=date_type)
else:
end_last = end_f.subtract(years=1)
return start_last.format("YYYY-MM-DD HH:mm:ss"), end_last.format(
"YYYY-MM-DD HH:mm:ss")
......@@ -356,7 +401,7 @@ def get_datetime_str(timestamp):
'''
if not timestamp:
timestamp = time.time()
time_array = time.localtime(timestamp)
return time.strftime("%Y-%m-%d %H:%M:%S", time_array)
......@@ -671,7 +716,7 @@ def get_last_3month_start_end(date=None):
:param dt: datetime obj
:return:
"""
if not date:
dt = get_current_date_time()
else:
......@@ -733,7 +778,7 @@ def get_this_year_start_end(date=None):
:param dt: datetime obj
:return:
"""
if not date:
dt = get_current_date_time()
else:
......@@ -749,7 +794,7 @@ def get_last_year_start_end(date=None):
:param dt: datetime obj
:return:
"""
if not date:
dt = get_current_date_time()
else:
......@@ -765,7 +810,7 @@ def get_last_month_start_end(date=None):
:param dt: datetime obj
:return:
"""
if not date:
dt = get_current_date_time()
else:
......@@ -873,7 +918,7 @@ def get_current_start_n_end_dt(date_type, include_end_day=False):
start_year += 1
start_month = start_month - 12
start = datetime.datetime(start_year, start_month, 1)
elif date_type == "season":
season_number = int((current_dt.month - 1) / 3)
start = datetime.datetime(current_dt.year, season_number * 3 + 1, 1)
......@@ -984,7 +1029,7 @@ def get_previous_slot(slots, date_type, compare_type=None):
return get_slots_between_date(
start - datetime.timedelta(days=days),
end - datetime.timedelta(days=days), date_type)
elif date_type == "hour":
hours = int((end - start).total_seconds() / 3600) + 1
return get_slots_between_date(start - datetime.timedelta(hours=hours),
......@@ -1032,8 +1077,8 @@ def get_slots_between_date(start, end, interval_type, growth_type=None):
while start <= end:
date_dts.append([start, start + datetime.timedelta(days=1)])
start += datetime.timedelta(days=1)
elif interval_type == "month":
start_months = start.year * 12 + start.month
end_months = end.year * 12 + end.month
......@@ -1145,7 +1190,7 @@ def deco_convert_date_to_dt(f):
if kwarg in kwargs and kwargs[kwarg]:
kwargs[kwarg] = convert_to_dt(kwargs[kwarg])
return f(*args, **kwargs)
return deco
......@@ -1432,7 +1477,7 @@ def get_time_duration_by_str(duration_str):
days = int(duration_str / (60 * 60 * 24))
if days > 0:
return_str += "%s天" % str(days)
hours = int(duration_str % (60 * 60 * 24) / (60 * 60))
if hours > 0:
return_str += "%s时" % str(hours)
......
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