from pot_libs.sanic_api import summary from unify_api.constants import USER_SIDE, DEFAULT_PV_PARAMS from unify_api.modules.energy_optimize.components.energy_opt_cps import \ UsaReq, UseResp, UsaResp, DpResp, DpaReq, DpaResp from unify_api.modules.energy_optimize.procedures.energy_opt_pds import \ process_capacity, process_opt_analysis, get_user_energy_storage_curve, \ get_user_energy_storage_economic_evaluate, pv_process_opt_analysis from unify_api.modules.energy_optimize.procedures.energy_utils import \ NO_DATA_EXPRESS, get_slots_between_date, get_current_date, format_value, \ format_chart_data, DATA_FORMAT from unify_api.modules.energy_optimize.service.energy_store_optimize import \ ess_out_result from unify_api.modules.energy_optimize.service.pv_distributed_photovoltaic import \ pv_out_result @summary("用户侧储能-默认数据") async def post_user_side_energy(req) -> UseResp: return UseResp( cell_price=USER_SIDE.get("cell_price"), pcs_price=USER_SIDE.get("pcs_price"), other_ttl_charge=USER_SIDE.get("other_ttl_charge"), pcs_efficiency=USER_SIDE.get("pcs_efficiency"), bat_efficiency=USER_SIDE.get("bat_efficiency"), decay_rate=USER_SIDE.get("decay_rate"), dod=USER_SIDE.get("dod"), year_use_days=USER_SIDE.get("year_use_days"), charge_c_rate=USER_SIDE.get("charge_c_rate") ) @summary("用户侧储能-查看分析结果") async def post_user_side_analyzed(req, body: UsaReq) -> UsaResp: inlid = body.inline_id params = { "cell_price": body.cell_price, "pcs_price": body.pcs_price, "other_ttl_charge": body.other_ttl_charge, "pcs_efficiency": body.pcs_efficiency, "bat_efficiency": body.bat_efficiency, "decay_rate": body.decay_rate, "DOD": body.dod, "year_use_days": body.year_use_days, "charge_C_rate": body.charge_c_rate } storage_optimize_tools_data = await ess_out_result(inlid, params) # 没有储能结果, 返回message信息 if storage_optimize_tools_data.get("message"): return UsaResp(message=storage_optimize_tools_data.get("message")) # 有储能分析结果 capacity = storage_optimize_tools_data["capacity"] capacity = process_capacity(capacity) opt_analysis = storage_optimize_tools_data["opt_analysis"] opt_analysis = process_opt_analysis(opt_analysis) opt_curve = get_user_energy_storage_curve(inlid, storage_optimize_tools_data) economic_evaluate = get_user_energy_storage_economic_evaluate( inlid, storage_optimize_tools_data) return UsaResp(capacity=capacity, economic_evaluate=economic_evaluate, opt_analysis=opt_analysis, opt_curve=opt_curve) @summary("分布式光伏-默认数据") async def post_distributed_photovoltaic(req) -> DpResp: return DpResp( self_use_ratio=DEFAULT_PV_PARAMS.get("self_use_ratio"), rmb_per_wp=DEFAULT_PV_PARAMS.get("rmb_per_wp"), sel_use_per_kwh=DEFAULT_PV_PARAMS.get("sel_use_per_kwh"), self_use_price_discout=DEFAULT_PV_PARAMS.get("self_use_price_discout"), first_install_subsidy=DEFAULT_PV_PARAMS.get("first_install_subsidy"), local_subsidy=DEFAULT_PV_PARAMS.get("local_subsidy"), local_subsidy_year=DEFAULT_PV_PARAMS.get("local_subsidy_year") ) @summary("分布式光伏-查看分析结果") async def post_distributed_photovoltaic_analyzed(req, body: DpaReq) -> DpaResp: inlid = body.inline_id params = { "install_space": body.install_space, "self_use_ratio": body.self_use_ratio, "rmb_per_wp": body.rmb_per_wp, "sel_use_per_kwh": body.sel_use_per_kwh, "self_use_price_discout": body.self_use_price_discout, "first_install_subsidy": body.first_install_subsidy, "local_subsidy": body.local_subsidy, "local_subsidy_year": body.local_subsidy_year } # defalut_params = pv_optimize_tool.get_pv_params(inlid) # defalut_params["install_space"] = install_space pv_optimize_data = await pv_out_result(inlid, params) # 没有储能结果, 返回message信息 if pv_optimize_data.get("message"): return DpaResp(message=pv_optimize_data.get("message")) install_cap = pv_optimize_data["install_cap"] invest_evaluate = pv_optimize_data["invest_evaluate"] invest_evaluate["i_and_r"]["ttl_invest"] = \ format_value(invest_evaluate["i_and_r"]["ttl_invest"], "big_money4")[ 0] invest_evaluate["i_and_r"]["first_year_income"] = \ format_value(invest_evaluate["i_and_r"]["first_year_income"], "big_money4")[0] invest_evaluate["i_and_r"]["self_use_ratio"] = \ format_value(invest_evaluate["i_and_r"]["self_use_ratio"], "percentage2")[0] invest_evaluate["i_and_r"]["first_year_income_rate"] = \ format_value( invest_evaluate["i_and_r"]["first_year_income_rate"], "percentage2")[0] for k in ("coal", "CO2", "SO2", "NOx"): invest_evaluate["carbon_footprint"][k] = \ format_value(invest_evaluate["carbon_footprint"][k], "big_weight4")[0] opt_analysis = pv_optimize_data["opt_analysis"] opt_analysis = pv_process_opt_analysis(opt_analysis) # 优化曲线 opt_curve = pv_optimize_data["opt_curve"] slots = get_slots_between_date(get_current_date(), get_current_date(), "hour") curve_datas = {} for slot in slots: curve_datas[slot[0]] = {} for item in ["load_curve", "load_pv_curve", "pv_curve"]: curve_datas[slot[0]][item] = NO_DATA_EXPRESS for value in opt_curve: if value["quarter_time"] == slot[0].strftime("%H:%M:%S"): curve_datas[slot[0]]["load_curve"] = value["load_curve"] curve_datas[slot[0]]["load_pv_curve"] = value["load_pv_curve"] curve_datas[slot[0]]["pv_curve"] = value["pv_curve"] capacity_scale_chart = { "data": format_chart_data(curve_datas, date_type="hour"), "unit": DATA_FORMAT["p"]["unit"] } return DpaResp(install_cap=install_cap, invest_evaluate=invest_evaluate, opt_analysis=opt_analysis, capacity_scale_chart=capacity_scale_chart )