import pendulum from datetime import datetime, date from unify_api.modules.electric_optimization.dao.power_index import \ tc_by_inline_id, price_policy_by_cid, pids_by_cid, get_economic_operations from unify_api.modules.electric_optimization.components.optimization_cps import ( PowerSaveResp, PeakCutValleyFillResp, MdSpaceResp, PowerFactorResp, ) from unify_api.modules.electric_optimization.procedures.optimization_pds import ( month_md_space ) from unify_api.modules.electric_optimization.procedures.optimization_pds import ( month_power_pcvf, month_power_factors, month_power_loadrate, load_inline_power ) from unify_api.constants import ADD_ELE_PRICE from unify_api.modules.elec_charge.dao.elec_charge_dao import ( point_kwh_charge ) # 需量管理 async def md_space_service(inline_id): dt = pendulum.now(tz="Asia/Shanghai") this_month_start_dt = pendulum.date(year=dt.year, month=dt.month, day=1) latest_12_months = [this_month_start_dt.subtract(months=i) for i in range(1, 13)] latest_12_months.sort() monthstr_list = [str(date(i.year, i.month, day=1))[:10] for i in latest_12_months] md_space_map = await month_md_space(inline_id, monthstr_list) inline = await tc_by_inline_id(inline_id) price_policy = await price_policy_by_cid(inline["cid"]) last_month = monthstr_list[-1] return MdSpaceResp( time_slots=[f"{str(i.year)[2:]}年{i.month}月" for i in latest_12_months], md_charge_values=[ round(md_space_map.get(i, {}).get("inline_md_charge"), 2) if type(md_space_map.get(i, {}).get("inline_md_charge")) in [int, float] else "" for i in monthstr_list ], tc_charge_values=[ round(md_space_map.get(i, {}).get("inline_tc_charge"), 2) if type(md_space_map.get(i, {}).get("inline_md_charge")) in [int, float] else "" for i in monthstr_list ], inline_md_predict=round( md_space_map.get(last_month, {}).get("inline_md_predict"), 2) if type(md_space_map.get(last_month, {}).get("inline_md_predict")) in [ int, float] else "", tc_runtime=inline["tc_runtime"] if inline else "", price_md=price_policy["price_md"] if price_policy else "", price_tc=price_policy["price_tc"] if price_policy else "", ) # 功率因素 async def power_factor_service(inline_id): dt = pendulum.now(tz="Asia/Shanghai") this_month_start_dt = pendulum.date(year=dt.year, month=dt.month, day=1) latest_12_months = [this_month_start_dt.subtract(months=i) for i in range(1, 13)] latest_12_months.sort() latest_12_months = [date(i.year, i.month, i.day) for i in latest_12_months] power_factors_map = await month_power_factors(inline_id, latest_12_months) last_month = latest_12_months[-1] save_charge = power_factors_map.get(last_month, {}).get("save_charge") last_month_pf = ( power_factors_map.get(last_month, {}).get("cos") if power_factors_map.get(last_month, {}).get("cos") else "" ) save_charge = save_charge if isinstance(save_charge, (int, float)) else "" if last_month_pf != "" and last_month_pf > 0.9: # 功率因数大于0.9不返回save_charge save_charge = None inline = await tc_by_inline_id(inline_id) point_list = await pids_by_cid(inline.get("cid")) charge_res = await point_kwh_charge(point_list) # total_charge = charge_res["aggregations"]["charge"]["value"] total_charge = charge_res.get("charge") or 0 if last_month_pf and 0.9 > last_month_pf >= 0: punish_money = total_charge * ADD_ELE_PRICE[last_month_pf] / 100 else: punish_money = 0.0 return PowerFactorResp( time_slots=[f"{str(i.year)[2:]}年{i.month}月" for i in latest_12_months], power_factor_values=[ power_factors_map.get(i, {}).get("cos") if power_factors_map.get(i, {}).get("cos") else "" for i in latest_12_months ], power_factor=last_month_pf, save_charge=save_charge, punish_money=round(punish_money, 2) ) # 移峰填谷 async def power_peakcut_service(inline_id): dt = pendulum.now(tz="Asia/Shanghai") this_month_start_dt = pendulum.date(year=dt.year, month=dt.month, day=1) latest_12_months = [this_month_start_dt.subtract(months=i) for i in range(1, 13)] latest_12_months.sort() monthstr_list = [str(date(i.year, i.month, day=1))[:10] for i in latest_12_months] power_pcvf_map = await month_power_pcvf(inline_id, monthstr_list) pcvf_indexs = [] for i in monthstr_list: score = power_pcvf_map.get(str(i), {}).get("score", "") if type(score) in [int, float]: pass else: score = "" pcvf_indexs.append(score) # 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") # ) # es_end_time = pendulum.datetime(now.year, now.month, 1).strftime( # "%Y-%m-%dT%H:%M:%S+08:00") # power_use_info = await inline_power_use_info(inline_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") ) end_time = pendulum.datetime(now.year, now.month, 1).strftime( "%Y-%m-%d %H:%M:%S") power_use_info = await load_inline_power(inline_id, start_time, end_time) month_kwh = power_use_info.get("kwh") or 0 month_charge = power_use_info.get("charge") or 0 avg_price = month_kwh / month_charge if month_charge else "" last_month = monthstr_list[-1] save_charge = power_pcvf_map.get(str(last_month), {}).get("cost_save") inline = await tc_by_inline_id(inline_id) price_policy = await price_policy_by_cid(inline["cid"]) if avg_price and price_policy["price_f"]: growth_percentage = round( ( (round(avg_price, 2) - round(price_policy["price_f"], 2)) / round(price_policy["price_f"], 2) ) * 100, 2, ) else: growth_percentage = "" return PeakCutValleyFillResp( time_slots=[f"{str(i.year)[2:]}年{i.month}月" for i in latest_12_months], pcvf_indexs=pcvf_indexs, avg_price=avg_price, save_charge=save_charge, pcvf_index=power_pcvf_map.get(str(last_month), {}).get("score", ""), price_f=price_policy["price_f"], growth_percentage=growth_percentage, ) # 经济运行 async def power_save_service(inline_id): dt = pendulum.now(tz="Asia/Shanghai") this_month_start_dt = pendulum.date(year=dt.year, month=dt.month, day=1) latest_12_months = [this_month_start_dt.subtract(months=i) for i in range(1, 13)] latest_12_months.sort() monthstr_list = [str(date(i.year, i.month, day=1))[:10] for i in latest_12_months] power_loadrate_map = await month_power_loadrate(inline_id, monthstr_list) load_rates = [] for i in monthstr_list: load_rate = power_loadrate_map.get(str(i), {}).get("mean_load_factor", "") if type(load_rate) in [int, float]: load_rate = round(load_rate / 0.01, 2) else: load_rate = "" load_rates.append(load_rate) economic_operations = await get_economic_operations(inline_id, monthstr_list[-1]) economic_kpi_x_list = [ i["kpi_x"] for i in economic_operations if type(i["kpi_x"]) in [int, float] ] economic_kpi_x = max(economic_kpi_x_list) if economic_kpi_x_list else "" total_economic_save = round( sum( [ i["save_charge"] for i in economic_operations if i["save_charge"] and i["save_charge"] >= 0 ] ), 2, ) save_charge = 0 if total_economic_save <= 0 else total_economic_save return PowerSaveResp( time_slots=[f"{str(i.year)[2:]}年{i.month}月" for i in latest_12_months], load_rates=load_rates, avg_load_rate=load_rates[-1], save_charge=save_charge if economic_kpi_x != "" else "", )