energy_opt.py 6.31 KB
Newer Older
lcn's avatar
lcn committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146
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
                   )