from unify_api.modules.elec_charge.common.utils import \ max_min_time, load_power_charge from pot_libs.sanic_api import summary from pot_libs.utils.pendulum_wrapper import my_pendulum from unify_api.modules.common.procedures.pttl_max import load_pttl_max from unify_api.modules.elec_charge.components.elec_statistics_cps import \ PcStatiReq, PcStatiResp, MaxpReq, MaxpResp, PcmResp from unify_api.utils.common_utils import round_2 from unify_api.modules.elec_charge.service.elec_statistics_service import \ power_charge_stats_srv @summary('电量电费统计曲线') async def post_power_charge_stati(req, body: PcStatiReq) -> PcStatiResp: # 1.获取参数 cid = body.cid point_id = body.point_id start = body.start end = body.end date_type = body.date_type return await power_charge_stats_srv(cid, point_id, start, end, date_type) @summary("小程序最高负荷/web最大需量") async def post_max_p(req, body: MaxpReq) -> MaxpResp: cid = body.cid pid = body.point_id start = body.start end = body.end max_val, max_val_time = await load_pttl_max(cid, start, end, point_id=pid) return MaxpResp(maxp=max_val, date_time=max_val_time) @summary('电量电费/最多最少最高负荷') async def post_power_charge_min_max(req, body: PcStatiReq) -> PcmResp: # 1.获取参数 cid = body.cid pid = body.point_id start = body.start end = body.end date_type = body.date_type return await power_charge_min_max_srv(cid, pid, start, end, date_type) async def power_charge_min_max_srv(cid, pid, start, end, date_type): # 初始化返回值 max_p = {"value": "", "time": ""} max_kwh = {"value": "", "time": ""} min_kwh = {"value": "", "time": ""} max_charge = {"value": "", "time": ""} min_charge = {"value": "", "time": ""} max_val, max_val_time = await load_pttl_max(cid, start, end, point_id=pid) max_p["value"] = round_2(max_val) max_p["time"] = max_val_time # 2. 如果是日统计,则需要增加今日/昨日负荷曲线, 15min一个点 if date_type == "day": kwh_sv, charge_sv = await load_power_charge([cid], pid, start, end, date_type) k_max_v, k_max_time, k_min_v, k_min_time = \ max_min_time(kwh_sv, add_one_index=True) c_max_v, c_max_time, c_min_v, c_min_time = \ max_min_time(charge_sv, add_one_index=True) elif date_type == "month": # 本月电量电费, 平均电价 kwh_sv, charge_sv = await load_power_charge([cid], pid, start, end, date_type) k_max_v, k_max_time, k_min_v, k_min_time = max_min_time(kwh_sv) c_max_v, c_max_time, c_min_v, c_min_time = max_min_time(charge_sv) elif date_type == "year": # 本月电量电费 kwh_sv, charge_sv = await load_power_charge([cid], pid, start, end, date_type) k_max_v, k_max_time, k_min_v, k_min_time = max_min_time(kwh_sv) c_max_v, c_max_time, c_min_v, c_min_time = max_min_time(charge_sv) else: 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_mm = (end_f - start_f).in_minutes() if diff_mm <= 48 * 60: # 自定义选时范围,不需要最后时间的数据,解决bug end = end_f.subtract(minutes=1).format("YYYY-MM-DD HH:mm:ss") # 电量电费 kwh_sv, charge_sv = await load_power_charge([cid], pid, start, end, date_type) k_max_v, k_max_time, k_min_v, k_min_time = \ max_min_time(kwh_sv, add_one_index=True, in_2_day=True) c_max_v, c_max_time, c_min_v, c_min_time = \ max_min_time(charge_sv, add_one_index=True, in_2_day=True) else: # 电量电费 kwh_sv, charge_sv = await load_power_charge([cid], pid, start, end, date_type) k_max_v, k_max_time, k_min_v, k_min_time = max_min_time(kwh_sv) c_max_v, c_max_time, c_min_v, c_min_time = max_min_time(charge_sv) max_kwh["value"] = round_2(k_max_v) max_kwh["time"] = k_max_time min_kwh["value"] = round_2(k_min_v) min_kwh["time"] = k_min_time max_charge["value"] = round_2(c_max_v) max_charge["time"] = c_max_time min_charge["value"] = round_2(c_min_v) min_charge["time"] = c_min_time return PcmResp(max_p=max_p, max_kwh=max_kwh, min_kwh=min_kwh, max_charge=max_charge, min_charge=min_charge)