proxy_optimization.py 6.9 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 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207
from dataclasses import fields

from pot_libs.sanic_api import summary
from unify_api.modules.electric_optimization.components.proxy_optimization_cps import (
    ProxyOpReq,
    ProxyPowerFactorListResp,
    ProxyPowerFactorPageItem,
    ProxyPowerFactorSummarryResp,
    ProxyMdSpaceListResp,
    ProxyMdSpacePageItem,
    ProxyMdSpaceSummarryResp,
    ProxyPowerSaveListResp,
    ProxyPowerSaveItem,
    ProxyPowerSaveSummaryResp,
    ProxyPcvfListResp,
    ProxyPcvfPageItem,
    ProxyElectricOpSummaryResp,
    OptimizationInfo,
)
from unify_api.modules.electric_optimization.procedures.proxy_optimization_pds import (
    proxy_power_factor_list,
    proxy_power_factor_summary,
    proxy_md_space_list,
    proxy_md_space_summary,
    proxy_power_save_list,
    proxy_power_save_summary,
    proxy_pcvf_list,
    proxy_pcvf_summary,
    proxy_electric_optimization_summary,
)


@summary("知电U管理版本用电优化-功率因素-列表数据")
async def post_power_factor_list(req, body: ProxyOpReq) -> ProxyPowerFactorListResp:
    cids, month_str = body.cids, body.month
    page_size, page_num = body.page_size or 10, body.page_num or 1
    sort_field, sort_direction = (
        body.sort_field or "company_name",
        body.sort_direction or "asc",
    )
    status_list = body.status
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    pf_page_map = await proxy_power_factor_list(
        cids,
        month_str,
        page_size=page_size,
        page_num=page_num,
        sort_field=sort_field,
        sort_direction=sort_direction,
        status_list=status_list,
    )

    return ProxyPowerFactorListResp(
        rows=[ProxyPowerFactorPageItem(**i) for i in pf_page_map["rows"]],
        total=pf_page_map["total"],
    )


@summary("知电U管理版本用电优化-功率因素-统计数据")
async def post_power_factor_summary(req, body: ProxyOpReq) -> ProxyPowerFactorSummarryResp:
    cids, month_str = body.cids, body.month
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    pf_summary_map = await proxy_power_factor_summary(cids, month_str)
    return ProxyPowerFactorSummarryResp(
        **{
            k: v
            for k, v in pf_summary_map.items()
            if k in [field.name for field in fields(ProxyPowerFactorSummarryResp)]
        }
    )


@summary("知电U管理版本用电优化-需量管理-列表数据")
async def post_md_space_list(req, body: ProxyOpReq) -> ProxyMdSpaceListResp:
    cids, month_str = body.cids, body.month
    page_size, page_num = body.page_size or 10, body.page_num or 1
    sort_field, sort_direction = (
        body.sort_field or "company_name",
        body.sort_direction or "asc",
    )
    status_list = body.status
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    md_space_map = await proxy_md_space_list(
        cids,
        month_str,
        page_size=page_size,
        page_num=page_num,
        sort_field=sort_field,
        sort_direction=sort_direction,
        status_list=status_list,
    )

    return ProxyMdSpaceListResp(
        rows=[ProxyMdSpacePageItem(**i) for i in md_space_map["rows"]], total=md_space_map["total"],
    )


@summary("知电U管理版本用电优化-需量管理-统计数据")
async def post_md_space_summary(req, body: ProxyOpReq) -> ProxyMdSpaceSummarryResp:
    cids, month_str = body.cids, body.month
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    md_summary_map = await proxy_md_space_summary(cids, month_str)
    return ProxyMdSpaceSummarryResp(
        **{
            k: v
            for k, v in md_summary_map.items()
            if k in [field.name for field in fields(ProxyMdSpaceSummarryResp)]
        }
    )


@summary("知电U管理版本用电优化-经济运行-列表数据")
async def post_power_save_list(req, body: ProxyOpReq) -> ProxyPowerSaveListResp:
    cids, month_str = body.cids, body.month
    page_size, page_num = body.page_size or 10, body.page_num or 1
    sort_field, sort_direction = (
        body.sort_field or "company_name",
        body.sort_direction or "asc",
    )
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    power_save_map = await proxy_power_save_list(
        cids,
        month_str,
        page_size=page_size,
        page_num=page_num,
        sort_field=sort_field,
        sort_direction=sort_direction,
    )

    return ProxyPowerSaveListResp(
        rows=[ProxyPowerSaveItem(**i) for i in power_save_map["rows"]],
        total=power_save_map["total"],
    )


@summary("知电U管理版本用电优化-经济运行-统计数据")
async def post_power_save_summary(req, body: ProxyOpReq) -> ProxyPowerSaveSummaryResp:
    cids, month_str = body.cids, body.month
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    save_summary_map = await proxy_power_save_summary(cids, month_str)
    return ProxyPowerSaveSummaryResp(
        **{
            k: v
            for k, v in save_summary_map.items()
            if k in [field.name for field in fields(ProxyPowerSaveSummaryResp)]
        }
    )


@summary("知电U管理版本用电优化-移峰填谷-列表数据")
async def post_pcvf_list(req, body: ProxyOpReq) -> ProxyPcvfListResp:
    cids, month_str = body.cids, body.month
    page_size, page_num = body.page_size or 10, body.page_num or 1
    sort_field, sort_direction = (
        body.sort_field or "company_name",
        body.sort_direction or "asc",
    )
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    pcvf_map = await proxy_pcvf_list(
        cids,
        month_str,
        page_size=page_size,
        page_num=page_num,
        sort_field=sort_field,
        sort_direction=sort_direction,
    )

    return ProxyPcvfListResp(
        rows=[ProxyPcvfPageItem(**i) for i in pcvf_map["rows"]], total=pcvf_map["total"],
    )


@summary("知电U管理版本用电优化-移峰填谷-统计数据")
async def post_pcvf_summary(req, body: ProxyOpReq) -> ProxyPowerSaveSummaryResp:
    cids, month_str = body.cids, body.month
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    pcvf_summary_map = await proxy_pcvf_summary(cids, month_str)
    return ProxyPowerSaveSummaryResp(
        **{
            k: v
            for k, v in pcvf_summary_map.items()
            if k in [field.name for field in fields(ProxyPowerSaveSummaryResp)]
        }
    )


@summary("知电U管理版本-首页用电优化-统计数据")
async def post_electric_op_summary(req, body: ProxyOpReq) -> \
        ProxyElectricOpSummaryResp:
    cids, month_str = body.cids, body.month
    if len(month_str) == 7:
        month_str = f"{month_str}-01"
    summary_map = await proxy_electric_optimization_summary(cids, month_str)
    return ProxyElectricOpSummaryResp(
        power_factor=OptimizationInfo(**summary_map["power_factor"]),
        md_space=OptimizationInfo(**summary_map["md_space"]),
        pcvf=OptimizationInfo(**summary_map["pcvf"]),
        power_save=OptimizationInfo(**summary_map["power_save"]),
    )