list_point.py 3.29 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
from pot_libs.logger import log
from pot_libs.sanic_api import summary
from pot_libs.mysql_util.mysql_util import MysqlUtil
from unify_api.modules.common.components.common_cps import CidReq
from unify_api.modules.common.procedures.points import point_to_mid
from unify_api.modules.common.service.list_point_service import \
    list_storey_service, list_tsp_point_service, list_point_level_service, \
    list_point_inline_service
from unify_api.modules.common.components.list_points_cps import (
    ListPointRequest,
    ListPointResponse,
    CommonPoint,
    CommonLocation,
    Inline, LsRep, LtpRep, LplResp, LpiResp)


@summary('获取监测点,进线列表')
async def post_list_point(req, body: ListPointRequest) -> ListPointResponse:
    cid = body.cid
    list_point, inline_list, points = [], [], {}
    sql = "SELECT pid,name,mtid,inlid,add_to_company FROM point where cid=%s"
    async with MysqlUtil() as conn:
        result = await conn.fetchall(sql, args=(cid,))
    add_to_company = []
    for res in result:
        # mtids.append(res["mtid"])
        inline_list.append({"inline_id": res["inlid"], "name": res["name"]})
        points[res["mtid"]] = res
        add_to_company.append(res["add_to_company"])
    if points:
        if len(points) == 1:
            l_sql = f"SELECT mtid,lid,ad_field from location " \
                    f"where mtid = {list(points.keys())[0]}"
        else:
            l_sql = f"SELECT mtid,lid,ad_field from location " \
                    f"where mtid in {tuple(points.keys())}"
        async with MysqlUtil() as conn:
            datas = await conn.fetchall(l_sql)
        for data in datas:
            if data["mtid"] in points.keys():
                temp_dict = {"temp1": "A相", "temp2": "B相", "temp3": "C相",
                             "temp4": "N相", "residual_current": "default"}
                flag = temp_dict.get(data["ad_field"]) or ""
                locations_dict = {
                    "item": flag,
                    "location_id": data.get("lid")
                }
                if "locations" not in points[data["mtid"]]:
                    points[data["mtid"]]["locations"] = [locations_dict]
                else:
                    points[data["mtid"]]["locations"].append(locations_dict)
        for k, v in points.items():
            list_point.append({
                "name":  v["name"], "point_id": v["pid"],
                "locations": v.get("locations") or [],
                "add_to_company": v["add_to_company"]
            })
    return ListPointResponse(
        points=list_point,
        inlines=inline_list,
        power_show_all=1 if any(add_to_company) else 0
    )


@summary('获取楼层-识电u')
async def post_list_storey(req, body: ListPointRequest) -> LsRep:
    cid = body.cid
    return await list_storey_service(cid)


@summary('获取TSP监测点-扬尘')
async def post_list_tsp_point(req, body: ListPointRequest) -> LtpRep:
    cid = body.cid
    return await list_tsp_point_service(cid)


@summary('新版监测点-包含进线变压器层级')
async def post_list_point_level(req, body: CidReq) -> LplResp:
    cid = body.cid
    return await list_point_level_service(cid)


@summary('新版监测点-进线')
async def post_list_point_inline(req, body: CidReq) -> LpiResp:
    cid = body.cid
    return await list_point_inline_service(cid)