新建1.5架构分支

pot_libs @ 05cfad93

+Subproject commit 05cfad9387871294a0215dabeccad37adbc487e3

requirements.txt

+aiobotocore==1.0.7
+aioelasticsearch==0.7.0
+aiofiles==0.5.0
+aiohttp==3.6.2
+aioitertools==0.7.0
+aiomysql==0.0.20
+aioredis==1.3.1
+aiotask-context==0.6.1
+appdirs==1.4.4
+aredis==1.1.8
+async-timeout==3.0.1
+asyncio-redis==0.16.0
+attrs==19.3.0
+black==20.8b1
+botocore==1.15.32
+certifi==2020.6.20
+cffi==1.14.0
+chardet==3.0.4
+click==7.1.2
+cryptography==2.9.2
+cycler==0.10.0
+docutils==0.15.2
+elasticsearch==7.8.0
+h11==0.9.0
+h2==3.2.0
+hiredis==1.0.1
+hpack==3.0.0
+hstspreload==2020.6.16
+httptools==0.1.1
+httpx==0.11.1
+hyperframe==5.2.0
+idna==2.9
+jmespath==0.10.0
+jsonschema==3.2.0
+kiwisolver==1.3.1
+limits==1.5.1
+matplotlib==3.3.3
+multidict==4.7.6
+mypy-extensions==0.4.3
+numpy==1.19.1
+pandas==1.0.5
+pathspec==0.8.0
+pendulum==2.1.1
+Pillow==8.0.1
+pycparser==2.20
+pycryptodome==3.9.8
+PyJWT==1.7.1
+PyMySQL==0.9.2
+pyparsing==2.4.7
+pyrsistent==0.16.0
+python-dateutil==2.8.1
+pytz==2020.1
+pytzdata==2020.1
+PyYAML==5.3.1
+redis==3.5.3
+regex==2020.10.28
+requests==2.24.0
+rfc3986==1.4.0
+sanic==20.3.0
+sanic-jwt==1.4.1
+sanic-limiter==0.1.3
+sanic-session==0.7.3
+schedule==0.6.0
+scipy==1.5.4
+six==1.15.0
+sniffio==1.1.0
+toml==0.10.2
+typed-ast==1.4.1
+typing-extensions==3.7.4.2
+ujson==3.0.0
+urllib3==1.25.9
+uvloop==0.14.0
+websockets==8.1
+Werkzeug==1.0.1
+wrapt==1.12.1
+XlsxWriter==1.3.7
+yarl==1.4.2

unify_api/__init__.py

+import sys
+sys.path.append(f'/home/ubuntu/data/code/unify_api_1.5/pot_libs')
+sys.path.append(f'/home/ubuntu/data/code/unify_api_1.5')
\ No newline at end of file

unify_api/modules/adio/__init__.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/8

unify_api/modules/adio/components/__init__.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+

unify_api/modules/adio/components/adio.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+from dataclasses import dataclass
+
+from pot_libs.common.components.fields import Item, DateTime
+from pot_libs.sanic_api import Model
+from pot_libs.sanic_api.column import List, Str, Float, Opt
+from unify_api.utils.response_code import DbErr
+
+
+@dataclass
+class AdioHistory(Model):
+    item: Item = Item
+    value_slots: list = List('数值').items(Float())
+    threhold: float = Opt(Float().eg(30))
+
+
+@dataclass
+class AdioHistoryResponse(Model, DbErr):
+    temperature: list = List().items(AdioHistory)
+    residual_current: list = List().items(AdioHistory)
+    time_slots: list = List('时间轴').items(Str('时间'))
+
+
+@dataclass
+class AdioCurrent(Model):
+    type: str = Str('location.type[temperature|residual_current]').eg(
+        'temperature')
+    item: Item = Item
+    real_time: DateTime = DateTime
+    value: float = Float('adio实时数值')
+
+
+@dataclass
+class AdioCurrentResponse(Model, DbErr):
+    temperature: list = List('温度列表').items(AdioCurrent)
+    residual_current: list = List('漏电流列表').items(AdioCurrent)
+
+
+@dataclass
+class AdioIndex(Model):
+    type: str = Str('location.type[temperature|residual_current]').eg(
+        'temperature')
+    item: Item = Item
+    max: float = Float('最大值')
+    max_time: DateTime = DateTime
+    min: float = Float('最小值')
+    min_time: DateTime = DateTime
+    avg: float = Float('平均值')
+
+
+@dataclass
+class AdioIndexResponse(Model, DbErr):
+    adio_indexes: list = List('指标统计列表').items(AdioIndex)
+
+
+adio_history_example = {
+    "范例1": {
+        "page_size": 20,
+        "page_num": 1,
+        "filter": {
+            "equals": [
+                {
+                    "field": "point_id",
+                    "value": 127
+                }
+            ],
+            "ranges": [
+                {
+                    "field": "datetime",
+                    "start": "2020-07-22 00:00:00",
+                    "end": "2020-07-22 23:59:59"
+                }
+            ],
+            "in_groups": [
+                {
+                    "field": "location_id",
+                    "group": [
+                        133, 134, 135
+                    ]
+                }
+            ]
+        },
+        "sort": {
+            "field": "datetime",
+            "direction": "desc"
+        }
+    }
+}
+
+adio_current_example = {
+    "范例1": {
+        "filter": {
+            "equals": [],
+            "ranges": [],
+            "in_groups": [
+                {
+                    "field": "location_id",
+                    "group": [
+                        259, 260, 261, 262, 263
+                    ]
+                }
+            ]
+        }
+    }
+}
+
+adio_index_example = {
+    "范例1": {
+        "page_size": 20,
+        "page_num": 1,
+        "filter": {
+            "equals": [
+                {
+                    "field": "point_id",
+                    "value": 175
+                }
+            ],
+            "ranges": [
+                {
+                    "field": "datetime",
+                    "start": "2020-07-22 00:00:00",
+                    "end": "2020-07-22 23:59:59"
+                }
+            ],
+            "in_groups": [
+                {
+                    "field": "location_id",
+                    "group": [
+                        259, 260, 261, 262, 263
+                    ]
+                }
+            ]
+        },
+        "sort": {
+            "field": "datetime",
+            "direction": "desc"
+        }
+    }
+}

unify_api/modules/adio/components/adio_card_cps.py

+from dataclasses import dataclass
+from pot_libs.sanic_api import Model
+from pot_libs.sanic_api.column import List
+
+
+@dataclass
+class AcResp(Model):
+    inline: list = List("进线").eg([])
+    transformer: list = List("变压器").eg([])
+    feeder: list = List("馈线").eg([])
+    power_dist: list = List("配电").eg([])
+    device: list = List("设备级").eg([])

unify_api/modules/adio/dao/adio_card_dao.py

+import copy
+
+from pot_libs.mysql_util.mysql_util import MysqlUtil
+
+
+async def monitor_location_join_by_locations(locations):
+    """monitor和location关联"""
+    sql = "SELECT m.mtid, m.name, m.m_type, l.lid, l.item, l.ad_type " \
+          "FROM monitor m inner join location l on m.mtid = l.mtid " \
+          "WHERE l.lid in %s and m.demolished=0 order by field(l.lid,{})".\
+        format(str(locations).replace("[", "").replace("]", ""))
+    async with MysqlUtil() as conn:
+        monitor_location_list = await conn.fetchall(
+            sql, args=(tuple(locations),))
+    return monitor_location_list
+
+
+async def alarm_setting_by_locations(locations):
+    """根据locations查询alarm_setting"""
+    sql = "select lid, threshold from soe_config_record " \
+          "where lid in %s " \
+          "and `etype` in ('overTemp', 'overResidualCurrent')"
+    async with MysqlUtil() as conn:
+        alarm_setting_list = await conn.fetchall(sql, args=(tuple(locations),))
+    return alarm_setting_list

unify_api/modules/adio/dao/adio_dao.py

+from pot_libs.mysql_util.mysql_util import MysqlUtil
+
+
+async def get_location_dao(lids):
+    location_info = {}
+    sql = "SELECT lid, item, ad_type FROM location WHERE lid IN %s"
+    async with MysqlUtil() as conn:
+        result = await conn.fetchall(sql, args=(lids,))
+        if result:
+            for res in result:
+                id = res.get("lid")
+                item = res.get("item")
+                type = res.get("ad_type")
+                location_info[id] = {"item": item, "type": type}
+    return location_info
+
+
+async def get_location_15min_dao(lid, start, end):
+    sql = f"SELECT lid,value_max,value_max_time,value_avg,value_min," \
+          f"value_min_time  FROM `location_15min_aiao`WHERE lid = %s " \
+          f"and create_time BETWEEN '{start}' and '{end}'"
+    async with MysqlUtil() as conn:
+        result = await conn.fetchall(sql, args=(lid,))
+    return result

unify_api/modules/adio/service/adio_card.py

+import json
+import time
+import aioredis
+
+from pot_libs.settings import SETTING
+from unify_api import constants
+from unify_api.constants import POINT_LEVEL_MAP
+from unify_api.modules.adio.components.adio_card_cps import AcResp
+from unify_api.modules.adio.dao.adio_card_dao import \
+    monitor_location_join_by_locations, alarm_setting_by_locations
+from unify_api.modules.common.procedures.list_point_pds import \
+    monitor_map_point_location
+
+ADIO_CURRENT = "adio_current"
+
+
+async def post_adio_card_service(location_list, cid):
+    """安全监测-卡片信息-level"""
+    # 构造cid下每个monitor包含的point和location
+    monitor_p_dic, monitor_l_dic = await monitor_map_point_location(cid)
+    # 1. 获取每个location_id的详细信息, mysql的in查询,打乱了location_list顺序
+    monitor_location_list = await monitor_location_join_by_locations(
+        location_list)
+    # 2. 查询温度漏电流阈值
+    alarm_setting_list = await alarm_setting_by_locations(location_list)
+    alarm_dic = {alarm["lid"]: alarm for alarm in alarm_setting_list}
+    # 3. 根据location_list查询redis
+    # multi_exec查询redis
+    redis = await aioredis.create_redis_pool(SETTING.redis_single)
+    tr = redis.multi_exec()
+    for location_id in location_list:
+        tr.hget(ADIO_CURRENT, location_id)
+    res_redis = await tr.execute()
+    # 3. 构造返回
+    ret_data = {
+        "inline": {},
+        "transformer": {},
+        "feeder": {},
+        "power_dist": {},
+        "device": {}
+    }
+    i = 0
+    for info in monitor_location_list:
+        m_name = info.get("name")
+        mtid = info.get("mtid")
+        m_type = POINT_LEVEL_MAP.get(info.get("m_type"))
+        location_id = info.get("lid")
+        location_type = info.get("ad_type")
+        location_item = info.get("item")
+        # redis数据
+        adio_value = ""
+        redis_data = res_redis[i]
+        i += 1
+        if redis_data:
+            adio_info = json.loads(redis_data)
+            time_now = int(time.time())
+            real_tt = adio_info.get("timestamp", 0)
+            if (time_now - real_tt) <= constants.REAL_EXP_TIME:
+                adio_value = round(adio_info.get("value", 0), 2)
+
+        threshold = alarm_dic[location_id]["threshold"]
+        if m_name in ret_data[m_type]:
+            if location_type == "temperature":
+                ret_data[m_type][m_name].setdefault(location_type, []).append(
+                    {"item": location_item, "value": adio_value}
+                )
+                # 阈值取最小数值
+                if threshold < ret_data[m_type][m_name]["t_threshold"]:
+                    ret_data[m_type][m_name]["t_threshold"] = threshold
+            else:
+                ret_data[m_type][m_name].setdefault(location_type, []).append(
+                    {"item": "漏电流", "value": adio_value}
+                )
+                ret_data[m_type][m_name]["r_threshold"] = threshold
+            # 卡片增加location_id字段
+            ret_data[m_type][m_name]["location_id"].append(location_id)
+        else:
+            if location_type == "temperature":
+                ret_data[m_type][m_name] = {
+                    "name": m_name,
+                    "t_threshold": threshold,
+                    "temperature": [
+                        {"item": location_item, "value": adio_value}]
+                }
+            else:
+                ret_data[m_type][m_name] = {
+                    "name": m_name,
+                    "r_threshold": threshold,
+                    "residual_current": [
+                        {"item": "漏电流", "value": adio_value}]
+                }
+            # 卡片增加location_id字段
+            ret_data[m_type][m_name]["location_id"] = [location_id]
+            # 卡片增加point_id字段
+            ret_data[m_type][m_name]["point_id"] = monitor_p_dic[mtid]
+    # 返回内容,转换为list
+    for key, value in ret_data.items():
+        if value:
+            ret_data[key] = [j for i, j in value.items()]
+        else:
+            ret_data[key] = []
+    return AcResp(
+        inline=ret_data["inline"],
+        transformer=ret_data["transformer"],
+        feeder=ret_data["feeder"],
+        power_dist=ret_data["power_dist"],
+        device=ret_data["device"],
+    )

unify_api/modules/adio/views/__init__.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+

unify_api/modules/adio/views/adio.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+import json
+import time
+from datetime import datetime
+from pot_libs.aredis_util.aredis_utils import RedisUtils
+from pot_libs.es_util.es_query import EsQuery
+from pot_libs.es_util.es_utils import EsUtil
+from pot_libs.sanic_api import summary, description, examples
+from pot_libs.mysql_util.mysql_util import MysqlUtil
+from pot_libs.logger import log
+from unify_api.utils import time_format
+from unify_api import constants
+from unify_api.utils.common_utils import round_2
+from unify_api.modules.adio.components.adio import (
+    AdioHistoryResponse,
+    AdioCurrentResponse,
+    AdioIndexResponse,
+    AdioIndex,
+    adio_index_example,
+    adio_history_example,
+    AdioHistory,
+    AdioCurrent,
+    adio_current_example,
+)
+from pot_libs.common.components.query import PageRequest, Range, Equal, Filter
+from unify_api.modules.adio.dao.adio_dao import get_location_dao, \
+    get_location_15min_dao
+
+
+@summary("返回安全监测历史曲线")
+@description("包含温度曲线和漏电流曲线，取最大值")
+@examples(adio_history_example)
+async def post_adio_history(req, body: PageRequest) -> AdioHistoryResponse:
+    #  1.获取intervel和时间轴
+    try:
+        date_start = body.filter.ranges[0].start
+        date_end = body.filter.ranges[0].end
+    except:
+        log.error("para error, ranges is NULL")
+        return AdioHistoryResponse(temperature=[], residual_current=[], time_slots=[])
+
+    try:
+        # 形如 interval = 900 slots=['00:00', '00:15', '00:30'
+        intervel, slots = time_format.time_pick_transf(date_start, date_end)
+    except:
+        log.error("para error, date format error")
+        return AdioHistoryResponse(temperature=[], residual_current=[], time_slots=[])
+
+    try:
+        location_group = body.filter.in_groups[0].group
+    except:
+        log.warning("para exception, in_groups is NULL, no location_id")
+        return AdioHistoryResponse(temperature=[], residual_current=[], time_slots=slots)
+
+    if not location_group:
+        log.warning("para exception, in_groups is NULL, no location_id")
+        return AdioHistoryResponse(temperature=[], residual_current=[], time_slots=slots)
+
+    # 3.获取温度曲线和漏电流曲线数据
+    # 动态漏电流阈值
+    sql = "select threshold from soe_config_record where lid in %s " \
+          "and etype = %s limit 1"
+    async with MysqlUtil() as conn:
+        settings = await conn.fetchall(sql, args=(tuple(location_group),
+                                                  "overResidualCurrent"))
+    residual_current_threhold = settings[0]["threshold"] if settings else 30
+    if intervel == 24 * 3600:
+        table_name = "location_1day_aiao"
+        date_format = "%%m-%%d"
+    else:
+        table_name = "location_15min_aiao"
+        date_format = "%%H:%%i"
+    sql = f"SELECT lid,value_max,DATE_FORMAT(create_time, '{date_format}') " \
+          f"date_time,ad_field FROM {table_name} WHERE lid in %s and " \
+          f"create_time BETWEEN '{date_start}' and '{date_end}' " \
+          f"order by create_time"
+    async with MysqlUtil() as conn:
+        datas = await conn.fetchall(sql, args=(location_group, ))
+    type_name = {
+        "residual_current": "漏电流", "temp1": "A相", "temp2": "B相",
+        "temp3": "C相", "temp4": "N线",
+    }
+    location_info, location_type_name = {}, {}
+    temp_list, residual_currents = [], []
+    for lid in location_group:
+        location_info[lid] = ["" for _ in slots]
+    for data in datas:
+        lid = data.get("lid")
+        value = data.get("value_max")
+        mid_slot = data.get("date_time")
+        if lid not in location_type_name.keys():
+            ad_field = data.get("ad_field")
+            location_type_name[lid] = ad_field
+        slot_index = slots.index(mid_slot)
+        location_info[lid][slot_index] = value
+    for key, value in location_info.items():
+        item_name = location_type_name.get(key)
+        item = type_name.get(item_name)
+        adio_his = AdioHistory(item=item, value_slots=value)
+        if item_name == "residual_current":
+            adio_his.threhold = residual_current_threhold
+            residual_currents.append(adio_his)
+        else:
+            temp_list.append(adio_his)
+    return AdioHistoryResponse(
+        temperature=temp_list, residual_current=residual_currents,
+        time_slots=slots
+    )
+
+
+@summary("返回安全监测实时参数")
+@description("包含温度和漏电流")
+@examples(adio_current_example)
+async def post_adio_current(req, body: PageRequest) -> AdioCurrentResponse:
+    try:
+        in_group = body.filter.in_groups[0]
+    except:
+        log.warning("para exception, in_groups is NULL, no location_id")
+        return AdioCurrentResponse(temperature=[], residual_current=[])
+
+    # location_ids
+    location_group = in_group.group
+    if not location_group:
+        log.warning("para exception, in_groups is NULL, no location_id")
+        return AdioCurrentResponse(temperature=[], residual_current=[])
+
+    # 读取location表信息
+    location_info = await get_location_dao(location_group)
+    temperature = []
+    residual_current = []
+    for location_id, item_info in location_info.items():
+        try:
+            adio_info = await RedisUtils().hget("adio_current", location_id)
+            if adio_info:
+                adio_info = json.loads(adio_info)
+        except Exception:
+            log.error("redis error")
+            return AdioCurrentResponse().db_error()
+
+        time_now = int(time.time())
+        if adio_info:
+            real_tt = adio_info.get("timestamp", 0)
+            if (time_now - real_tt) <= constants.REAL_EXP_TIME:
+                adio_value = round(adio_info.get("value", 0), 2)
+            else:
+                adio_value = ""  # 超过4小时的值直接丢弃
+            time_str = time_format.get_datetime_str(real_tt)
+        else:
+            adio_value = ""
+            time_str = time_format.get_datetime_str(time_now)
+
+        if item_info.get("type") == "residual_current":
+            adio_current = AdioCurrent(
+                type="residual_current", item="漏电流", real_time=time_str, value=adio_value
+            )
+            residual_current.append(adio_current)
+        else:
+            adio_current = AdioCurrent(
+                type="temperature",
+                item=item_info.get("item", ""),
+                real_time=time_str,
+                value=adio_value,
+            )
+            temperature.append(adio_current)
+
+    return AdioCurrentResponse(temperature=temperature, residual_current=residual_current)
+
+
+@summary("返回安全监测指标统计")
+@description("温度和漏电流的最高、最低、平均值")
+@examples(adio_index_example)
+async def post_adio_index(req, body: PageRequest) -> AdioIndexResponse:
+    try:
+        location_group = body.filter.in_groups[0].group
+    except:
+        log.warning("para exception, in_groups is NULL, no location_id")
+        return AdioIndexResponse(adio_indexes=[])
+
+    if not location_group:
+        log.warning("para exception, in_groups is NULL, no location_id")
+        return AdioIndexResponse(adio_indexes=[])
+
+    # # load location表信息
+    location_info = await get_location_dao(location_group)
+
+    # 获取时间
+    try:
+        start = body.filter.ranges[0].start
+        end = body.filter.ranges[0].end
+    except:
+        log.error("para error, ranges is NULL")
+        return AdioIndexResponse(adio_indexes=[])
+    adio_indexes = []
+    for lid in location_group:
+        value_max, value_min, value_avg = [], [], []
+        value_max_time, value_min_time = [], []
+        datas = await get_location_15min_dao(lid, start, end)
+        for data in datas:
+            value_max.append(data.get("value_max"))
+            value_min.append(data.get("value_min"))
+            value_avg.append(data.get("value_avg"))
+            value_max_time.append(data.get("value_max_time"))
+            value_min_time.append(data.get("value_min_time"))
+        if value_max:
+            value_max_max = max([m for m in value_max])
+            value_max_max_index = value_max.index(value_max_max)
+            value_max_time_data = value_max_time[value_max_max_index]
+        else:
+            value_max_max, value_max_time_data = "", ""
+        if value_min:
+            value_min_min = min([m for m in value_min])
+            value_min_min_index = value_min.index(value_min_min)
+            value_min_time_data = value_min_time[value_min_min_index]
+        else:
+            value_min_min, value_min_time_data = "", ""
+        value_avg_list = [m for m in value_avg]
+        value_avg_data = sum(value_avg_list)/len(value_avg_list) \
+            if value_avg_list else 0
+        adio_index = AdioIndex(
+            type=location_info[lid]["type"],
+            item=location_info[lid]["item"],
+            max=round_2(value_max_max),
+            max_time=str(value_max_time_data) if value_max_time_data else "",
+            min=round_2(value_min_min),
+            min_time=str(value_min_time_data) if value_min_time_data else "",
+            avg=round_2(value_avg_data),
+        )
+        adio_indexes.append(adio_index)
+    return AdioIndexResponse(adio_indexes=adio_indexes)
+
+
+

unify_api/modules/adio/views/adio_card.py

+from pot_libs.sanic_api import summary
+from unify_api.modules.adio.components.adio_card_cps import AcResp
+from unify_api.modules.adio.service.adio_card import post_adio_card_service
+from unify_api.modules.common.components.common_cps import AcReq
+
+
+@summary("安全监测-卡片信息-level")
+async def post_adio_card(req, body: AcReq) -> AcResp:
+    location_list = body.location_list
+    cid = body.cid
+    return await post_adio_card_service(location_list, cid)

unify_api/modules/alarm_manager/__init__.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/8
+

unify_api/modules/alarm_manager/components/__init__.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/8

unify_api/modules/alarm_manager/components/alarm_setting.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+from dataclasses import dataclass
+
+from pot_libs.common.components.fields import Cid
+from pot_libs.sanic_api import Model
+from pot_libs.sanic_api.column import Int, List, Str, Float, Opt
+from unify_api.utils.response_code import DbErr
+
+
+@dataclass
+class AlarmSetting(Model):
+    id: int = Int('数据库表ID')
+    type: str = Str('设置类型：【electric，adio】').eg('electric')
+    name: str = Str(
+        '电气量直接取name, adio量取{item}{type}'
+    ).eg('A相温度告警')
+    level: int = Int('等级').eg(2)
+    threshold: float = Float('阈值').eg(80)
+    duration: int = Opt(Int('持续时间').eg(60))
+    enable: int = Int('是否启用：0|1').eg(1)
+
+
+@dataclass
+class AlarmSettingUpdate(Model):
+    id: int = Int('数据库表ID')
+    threshold: float = Float('阈值').eg(80)
+    duration: int = Opt(Int('持续时间').eg(60))
+    enable: int = Int('是否启用：0|1').eg(1)
+    cid: Cid = Cid
+
+
+@dataclass
+class AlarmSettingResponse(Model):
+    total: int = Int('总量')
+    rows: list = List("所有设置").items(AlarmSetting)
+
+
+@dataclass
+class PointAlarmSettingRequest(Model):
+    point_id: int = Int('point_id')
+    location_ids: list = List("location_ids").items(Int('location_id'))
+    enable: int = Opt(Int('不传【ALL】：0【禁用】|1【启用】').eg(1))
+
+
+@dataclass
+class PointAlarmSettingResponse(Model, DbErr):
+    residual_current_op: int = Opt(Int("1-漏电流可操作, 0-漏电流不可操作"))
+    point_alarm_settings: list = List("point_alarm_setting").items(
+        AlarmSetting)
+    location_alarm_settings: list = List("location_alarm_setting").items(
+        AlarmSetting)
+
+
+point_alarm_setting_example = {
+    "范例1": {
+        "point_id": 128,
+        "location_ids": [136, 137, 138, 139, 140],
+        "enable": 1
+    }
+}

unify_api/modules/alarm_manager/components/alarm_static_cps.py

+from dataclasses import dataclass
+
+from pot_libs.common.components.fields import Cid
+from pot_libs.sanic_api import Model
+from pot_libs.sanic_api.column import Str, Int, Opt, List, Float, Dict
+
+
+@dataclass
+class SduAlarmReq(Model):
+    cid: Cid
+    start: str = Str("开始时间").eg("2020-07-30 00:00:00")
+    end: str = Str("结束时间").eg("2020-07-30 23:59:59")
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+
+
+@dataclass
+class RiskCount(Model):
+    security_user: int = Opt(Int("安全用户").eg(10))
+    risk_user: int = Opt(Int("风险用户").eg(2))
+
+
+@dataclass
+class ContentName(Model):
+    ele_overload: int = Opt(Int("用电超载").eg(10))
+    high_power_app: int = Opt(Int("大功率电器").eg(2))
+    illegal_ele_app: int = Opt(Int("违规电器").eg(2))
+    power_quality: int = Opt(Int("电能质量").eg(2))
+    ele_car_battery: int = Opt(Int("电能质量").eg(2))
+
+
+@dataclass
+class SlotValue(Model):
+    slots: list = List("slots").eg(["00-01", "00-02", "00-03"])
+    value: list = List("时间段对应的值").eg([1, 2, 3])
+
+
+@dataclass
+class SduAlarmResp(Model):
+    total_alarm_cnt: int = Int("总报警数").eg(12)
+    alarm_points_cnt: int = Opt(Int("报警户数").eg(12))
+    aver_alarm: float = Opt(Float("平均报警").eg(2.1))
+    risk_distribution: RiskCount = Opt(RiskCount)
+    content_distribution: ContentName = Opt(ContentName)
+    alarm_ranking: list = Opt(List("报警排名").eg([{"name": "某房间", "value": 10}]))
+    high_power_app: SlotValue = Opt(SlotValue)
+    ele_overload: SlotValue = Opt(SlotValue)
+    illegal_ele_app: SlotValue = Opt(SlotValue)
+    power_quality: SlotValue = Opt(SlotValue)
+    ele_car_battery: SlotValue = Opt(SlotValue)
+    illegal_app_list: list = Opt(
+        List("违规电器统计").eg([{"name": "热得快", "value": 4}])
+    )
+
+
+@dataclass
+class NlaReq(Model):
+    cid: Cid
+    storeys: list = List("楼层列表").eg([1, 2, 3])
+    importance: list = List("报警等级,默认:[1, 2, 3]").eg([1, 2, 3])
+
+    page_size: int = Int("每页记录数").eg(20)
+    page_num: int = Int("当前页码").eg(1)
+    start: str = Opt(Str("开始时间").eg("2020-05-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-03-10 23:59:59"))
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+    point_ids: list = Opt(List("房间检测点").eg([1, 3, 4]))
+
+
+@dataclass
+class WlaReq(Model):
+    cids: list = List("cid列表").eg([78, 83])
+    start: str = Opt(Str("开始时间").eg("2020-05-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-03-10 23:59:59"))
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+
+
+@dataclass
+class SasReq(Model):
+    cids: list = List("cid列表").eg([78, 113])
+    start: str = Str("开始时间").eg("2021-02-01 00:00:00")
+    end: str = Str("结束时间").eg("2021-02-28 23:59:59")
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+
+
+@dataclass
+class SassReq(Model):
+    cid: Cid
+    page_size: int = Int("每页记录数").eg(10)
+    page_num: int = Int("当前页码").eg(1)
+    sort: str = Str("desc: 降序, asc: 升序").eg("desc")
+    start: str = Opt(Str("开始时间").eg("2021-02-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-02-28 23:59:59"))
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+
+
+@dataclass
+class SassResp(Model):
+    alarm_ranking_total: int = Int('报警记录总量')
+    ele_app_ranking: list = List("电器识别").eg([{"name": "热得快", "value": 10}])
+    alarm_ranking: list = List("报警记录排名").eg([{"name": "某房间", "value": 10}])
+
+
+@dataclass
+class AppReq(Model):
+    cid: Cid
+    start: str = Opt(Str("开始时间").eg("2021-02-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-02-28 23:59:59"))
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+
+
+@dataclass
+class AppResp(Model):
+    ele_app_ranking: list = List("电器识别").eg([{"name": "热得快", "value": 10}])
+
+
+@dataclass
+class SisReq(Model):
+    cid: Cid
+    start: str = Opt(Str("开始时间").eg("2021-02-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-02-28 23:59:59"))
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+
+
+@dataclass
+class SisResp(Model):
+    ele_overload: dict = Dict("线路过载").eg(
+        {"slots": ["00-01", "00-02", "00-03"], "value": [1, 2, 3]})
+    illegal_ele_app: dict = Dict("违规电器").eg(
+        {"slots": ["00-01", "00-02", "00-03"], "value": [1, 2, 3]})
+    power_quality: dict = Dict("电能质量").eg(
+        {"slots": ["00-01", "00-02", "00-03"], "value": [1, 2, 3]})
+
+
+@dataclass
+class SiasReq(Model):
+    cid: Cid
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+
+
+@dataclass
+class SiasResp(Model):
+    risk_distribution: RiskCount = Opt(RiskCount)
+    content_distribution: ContentName = Opt(ContentName)
+    electric_use_score: float = Opt(Float("用电安全指数").eg(90.3))
+
+
+@dataclass
+class SebReq(Model):
+    cid: Cid
+    storeys: list = List("楼层列表").eg([1, 2, 3])
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(4)
+
+
+@dataclass
+class SebResp(Model):
+    return_data: list = List("用电行为卡片数据").eg(
+        [{"name": "89栋", "room_data": [
+            {
+                "room_name": "房间名",
+                "illegal_ele_app": "违规",
+                "high_power_app": "大功率",
+                "normal_app": "正常",
+                "alarm_score": "报警分"
+            },
+            {
+                "room_name": "房间名",
+                "illegal_ele_app": "违规",
+                "high_power_app": "大功率",
+                "normal_app": "正常",
+                "alarm_score": "报警分"
+            }
+        ]}])
+
+
+@dataclass
+class SiarResp(Model):
+    illegal_app: list = List("违规电器").eg(
+        [{"name": "电动车电池", "value": 2}, {"name": "热得快", "value": 1}])
+    illegal_behavior: list = List("违规行为").eg(
+        [{"name": "104室", "value": 6}, {"name": "105室", "value": 1}])
+    alarm_ranking: list = List("报警排名").eg(
+        [{"name": "104室", "value": 2}, {"name": "105室", "value": 1}])
+
+
+@dataclass
+class LazReq(Model):
+    cid: Cid
+    point_list: list = List().eg([439, 440])
+    importance: list = List("报警等级,默认:[1, 2, 3]").eg([1, 2, 3])
+    page_size: int = Int("每页记录数").eg(20)
+    page_num: int = Int("当前页码").eg(1)
+    start: str = Opt(Str("开始时间").eg("2020-05-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-03-10 23:59:59"))
+    event_type: list = List("内容筛选").eg(["overTempRange1min", "overTHDU"])
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(6)
+
+
+@dataclass
+class TimeCount(Model):
+    daytime_cnt: int = Opt(Int("日间").eg(10))
+    night_cnt: int = Opt(Int("夜间").eg(10))
+    morning_cnt: int = Opt(Int("凌晨").eg(10))
+
+
+@dataclass
+class ZsResp(Model):
+    total_alarm_cnt: int = Int("报警总数").eg(12)
+    alarm_points_cnt: int = Opt(Int("报警监测点").eg(10))
+    aver_alarm: float = Opt(Float("平均报警").eg(2.1))
+    safe_run: float = Opt(Float("安全运行").eg(15))
+    time_interval_distribution: TimeCount = Opt(TimeCount)
+
+
+@dataclass
+class ZasReq(Model):
+    cid: Cid
+    page_size: int = Int("每页记录数").eg(10)
+    page_num: int = Int("当前页码").eg(1)
+    start: str = Opt(Str("开始时间").eg("2021-02-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-02-28 23:59:59"))
+    product: int = Int("product 1-知电 2-安电 3-安电管理 4-识电U").eg(6)
+
+
+@dataclass
+class ZasResp(Model):
+    total: int = Int('报警记录总量')
+    alarm_ranking: list = List("报警排名").eg([{"name": "某房间", "value": 10}])

unify_api/modules/alarm_manager/components/list_alarm.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+from dataclasses import dataclass
+
+from pot_libs.common.components.fields import DateTime, Group, Item, Cid
+from pot_libs.sanic_api import Model
+from pot_libs.sanic_api.column import Int, List, Str, Opt, Float
+from unify_api.utils.response_code import DbErr
+
+
+@dataclass
+class Alarm(Model):
+    date_time: DateTime
+    importance: int = Int('告警级别').eg(2)  # 数据库对应importance
+    name: str = Str('监测点名称').eg('母排A相')  # point直接取，location拼item
+    type: str = Str('事件类型').eg('underU')
+    type_name: str = Str('事件类型').eg('欠压')
+    description: str = Str('描述').eg('负载率超过80%， 达到89%')
+    redirect_type: str = Opt(Str('跳转类型').eg('scope | temp_trend'))
+    es_id: str = Str('es_id').eg('180_over_gap_u_a_1594274674')
+    url: str = Opt(Str(
+        '查看详情,url 可选：【scope_details， temp_trend】'
+    ).eg('/scope_details?doc_id={doc_id}'))
+    event_duration: int = Int('持续时间').eg('7')
+    company_name: str = Str('公司名').eg('广州ABC集团')
+    point_id: int = Opt(Int('point_id').eg(245))
+    storey_name: str = Opt(Str('storey名字').eg('264栋'))
+    room_name: str = Opt(Str('room名字').eg('101'))
+    storey_room_name: str = Opt(Str('storey+room名字').eg('264栋101'))
+    content: str = Opt(Str('内容').eg('负载率超过80%， 达到89%'))
+    sid: str = Opt(Str('设备id').eg('A2004000815'))
+    mtid: int = Opt(Int('持续时间').eg('7'))
+
+
+@dataclass
+class ListAlarmResponse(Model, DbErr):
+    total: int = Int('总量')
+    point_ids: list = Opt(List('监测点id').eg([1, 2]))
+    rows: list = List('报警信息').items(Alarm)
+
+
+@dataclass
+class ListAlarmReq(Model):
+    cid: int = Int('cid')
+    point_id: int = Opt(Int('监测点id 全部不传').eg(1))
+    start: str = Opt(Str("开始时间").eg("2020-05-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-03-10 23:59:59"))
+    importance: list = Opt(List('告警等级').eg([1, 2, 3]))
+    page_size: int = Int('每页行数').eg(20)
+    page_num: int = Int("当前请求页数").eg(1)
+    alarm_type: list = Opt(List('告警类型').eg(["overI", "unbalanceU"]))
+
+
+@dataclass
+class QueryDetails(Model):
+    doc_id: str = Str('es doc id')
+    wave_range:str = Opt(Str('波形范围').eg('100ms | all'))
+
+
+@dataclass
+class ScopeContent(Model):
+    ia: list = List().items(Int())
+    ib: list = Opt(List().items(Int()))
+    ic: list = List().items(Int())
+    ua: list = Opt(List().items(Int()))
+    ub: list = Opt(List().items(Int()))
+    uc: list = Opt(List().items(Int()))
+    uab: list = Opt(List().items(Int()))
+    ucb: list = Opt(List().items(Int()))
+
+
+@dataclass
+class ScopeDetailsResponse(Model, DbErr):
+    ctnum: int = Int('二表法 | 三表法').eg(3)
+    group: Group = Group
+    item: Item = Item
+    type: str = Str().eg('电流波动')
+    date_time: DateTime = DateTime
+    location: int = Int('故障点')
+    contents: ScopeContent = ScopeContent
+
+
+@dataclass
+class TempTrendResponse(Model, DbErr):
+    group: Group = Group
+    item: Item = Item
+    date_time: str = Str('时间')
+    description: str = Str('事件描述')
+    time_slots: list = List('时间轴').items(Str('时间'))
+    realtime: list = List('实时温度').items(Float())
+    daily: list = List('日前预测').items(Float())
+    quarterly: list = List('实时预测').items(Float())
+
+
+list_alarm_example = {
+    "范例1": {
+        "page_size": 20,
+        "page_num": 1,
+        "filter": {
+            "equals": [
+                {
+                    "field": "point_id",
+                    "value": 128
+                }
+            ],
+            "ranges": [
+                {
+                    "field": "datetime",
+                    "start": "2020-05-02 17:32:31",
+                    "end": "2020-07-03 17:32:31"
+                }
+            ],
+            "in_groups": [
+                {
+                    "field": "location_id",
+                    "group": [
+                        136, 137, 138, 139, 140
+                    ]
+                },
+                {
+                    "field": "type",
+                    "group": [
+                        "over_gap_i",
+                        "overResidualCurrent",
+                    ]
+                }
+            ]
+        },
+        "sort": {
+            "field": "datetime",
+            "direction": "desc"
+        }
+    }
+}
+
+temp_trend_example = {
+    "范例1": {
+        "doc_id": "199_overTemp__1595237202"
+    }
+}
+
+scope_details_example = {
+    "范例1": {
+        "doc_id": "180_over_gap_u_a_1594274674"
+    }
+}
+
+
+@dataclass
+class WlazReq(Model):
+    cid: Cid
+    point_list: list = List().eg([439, 440])
+    start: str = Opt(Str("开始时间").eg("2020-05-01 00:00:00"))
+    end: str = Opt(Str("结束时间").eg("2021-03-10 23:59:59"))

unify_api/modules/alarm_manager/dao/alarm_setting_dao.py

+from pot_libs.mysql_util.mysql_util import MysqlUtil
+
+
+async def post_update_alarm_emq_dao(set_id):
+    """根据id查询alarm_setting表"""
+    sql = "select * from soe_config_record where id=%s"
+    async with MysqlUtil() as conn:
+        alarm_dic = await conn.fetchone(sql, args=(set_id,))
+    return alarm_dic
+
+
+async def company_extend_dao(cid):
+    """根据cid查询company_extend表"""
+    sql = "select * from company_extend where cid=%s"
+    async with MysqlUtil() as conn:
+        event_dic = await conn.fetchall(sql, args=(cid,))
+    return event_dic
+
+
+async def list_alarm_data_dao(cid, point_list, page_num, page_size, start,
+                              end, importance, event_type):
+    sql = """select * from point_1min_event where pid in %s and 
+    event_datetime>='%s' and  event_datetime <"%s" and importance in %s 
+    and event_type in %s limit %s, %s"""
+    async with MysqlUtil() as conn:
+        data = await conn.fetchall(sql, args=(point_list, start, end,
+                                              importance, event_type,
+                                              (page_num-1)*page_size,
+                                               page_size))
+    return data
+
+
+async def get_list_alarm_dao(mid_sql, page_size, page_num):
+    start = (page_num-1)*page_size
+    sql = f"SELECT point_1min_event.*, company.fullname " \
+          f"FROM `point_1min_event` LEFT JOIN company " \
+          f"on company.cid=point_1min_event.cid where {mid_sql} " \
+          f"limit {start}, {page_size}"
+    async with MysqlUtil() as conn:
+        data = await conn.fetchall(sql)
+    return data
+
+
+async def get_total_list_alarm_dao(mid_sql):
+    sql = f"SELECT count(id) total FROM `point_1min_event` where {mid_sql}"
+    async with MysqlUtil() as conn:
+        data = await conn.fetchone(sql)
+    return data["total"]

unify_api/modules/alarm_manager/dao/list_alarm_dao.py

+import pendulum
+from pot_libs.es_util.es_utils import EsUtil
+from pot_libs.mysql_util.mysql_util import MysqlUtil
+from pot_libs.logger import log
+from unify_api.constants import POINT_1MIN_EVENT, SDU_ALARM_LIST, Product, \
+    EVENT_TYPE_MAP
+from unify_api.utils.time_format import convert_es_str, end_now_str
+
+index = POINT_1MIN_EVENT
+
+
+async def new_list_alarm_dao(cid, points, page_num, page_size, start, end,
+                             product, importance):
+    """es报警信息分页列表, 可根据报警等级筛选"""
+    # alarm_list = list(EVENT_TYPE_MAP.keys())  # 安电u
+    #
+    # if product == Product.RecognitionElectric.value:  # 识电u
+    alarm_list = SDU_ALARM_LIST
+    
+    query_body = {
+        "from": (page_num - 1) * page_size,
+        "size": page_size,
+        "query": {
+            "bool": {
+                "must": [
+                    {
+                        "term": {
+                            "cid": cid
+                        }
+                    },
+                    {
+                        "terms": {
+                            "type.keyword": alarm_list
+                        }
+                    },
+                    {
+                        "terms": {
+                            "point_id": points
+                        }
+                    },
+                    {
+                        "terms": {
+                            "importance": importance
+                        }
+                    }
+                ]
+            }
+        },
+        "sort": [
+            {
+                "datetime": {
+                    "order": "desc"
+                }
+            }
+        ]
+    }
+    if start and end:
+        start_es = convert_es_str(start)
+        end_es = end_now_str(end)
+        query_body["query"]["bool"]["must"].append(
+            {
+                "range": {
+                    "datetime": {
+                        "gte": start_es,
+                        "lte": end_es
+                    }
+                }
+            }
+        )
+    else:
+        # 有些脏数据超过现在时间的不显示
+        now_date = pendulum.now()
+        time_format = "%Y-%m-%dT%H:%M:%S+08:00"
+        end_date = now_date.strftime(time_format)
+        end_es = str(end_date)
+        query_body["query"]["bool"]["must"].append(
+            {
+                "range": {
+                    "datetime": {
+                        "lte": end_es
+                    }
+                }
+            }
+        )
+    log.info(f"index:{index}--query_body:{query_body}")
+    async with EsUtil() as es:
+        es_re = await es.search_origin(body=query_body, index=index)
+    return es_re
+
+
+async def new_list_alarm_dao_new15(cid, points, start, end, importance, offset,
+                                   page_size):
+    mid_li = [f"cid={cid}", f"event_type in {tuple(SDU_ALARM_LIST)}"]
+    if len(points) == 1:
+        mid_li.append(f"pid={points[0]}")
+    else:
+        mid_li.append(f"pid in {tuple(points)}")
+    if len(importance) > 0:
+        importance = str(tuple(importance)).replace(',)', ')')
+        mid_li.append(f"importance in {importance}")
+    if start and end:
+        mid_li.append(f"event_datetime BETWEEN '{start}' and '{end}'")
+    else:
+        now_date = pendulum.now()
+        time_format = "%Y-%m-%d %H:%M:%S"
+        end_date = str(now_date.strftime(time_format))
+        mid_li.append(f"event_datetime < '{end_date}'")
+    # 查询总数
+    total_sql = f"""
+        select count(*) total_count from point_1min_event where
+{' and '.join(mid_li)}
+    """
+    async with MysqlUtil() as conn:
+        total = await conn.fetchone(total_sql)
+    if total.get('total_count', 0) <= 0:
+        return 0, []
+    # 查询详情
+    sql = f"""
+        select * from point_1min_event where {' and '.join(mid_li)}
+        order by event_datetime desc limit %s,%s
+    """
+    async with MysqlUtil() as conn:
+        data = await conn.fetchall(sql, args=(offset, page_size))
+    return total.get('total_count', 0), data
+
+
+async def wx_list_alarm_dao(cids, product, start, end):
+    """小程序消息列表, 取当前范围最新40条"""
+    
+    # alarm_list = list(EVENT_TYPE_MAP.keys())  # 安电u
+    #
+    # if product == Product.RecognitionElectric.value:  # 识电u
+    alarm_list = SDU_ALARM_LIST
+    
+    query_body = {
+        "size": 40,
+        "query": {
+            "bool": {
+                "must": [
+                    {
+                        "terms": {
+                            "cid": cids
+                        }
+                    },
+                    {
+                        "terms": {
+                            "type.keyword": alarm_list
+                        }
+                    }
+                ]
+            }
+        },
+        "sort": [
+            {
+                "datetime": {
+                    "order": "desc"
+                }
+            }
+        ]
+    }
+    if start and end:
+        start_es = convert_es_str(start)
+        end_es = convert_es_str(end)
+        query_body["query"]["bool"]["must"].append(
+            {
+                "range": {
+                    "datetime": {
+                        "gte": start_es,
+                        "lte": end_es
+                    }
+                }
+            }
+        )
+    
+    log.info(f"index:{index}--query_body:{query_body}")
+    async with EsUtil() as es:
+        es_re = await es.search_origin(body=query_body, index=index)
+    return es_re
+
+
+async def list_alarm_zdu_dao(cid, points, page_num, page_size, start, end,
+                             importance, event_type):
+    """智电u, es报警信息分页列表, 可根据报警等级筛选"""
+    
+    query_body = {
+        "from": (page_num - 1) * page_size,
+        "size": page_size,
+        "query": {
+            "bool": {
+                "must": [
+                    {
+                        "term": {
+                            "cid": cid
+                        }
+                    },
+                    {
+                        "terms": {
+                            "point_id": points
+                        }
+                    },
+                    {
+                        "terms": {
+                            "importance": importance
+                        }
+                    }
+                ]
+            }
+        },
+        "sort": [
+            {
+                "datetime": {
+                    "order": "desc"
+                }
+            }
+        ]
+    }
+    if start and end:
+        start_es = convert_es_str(start)
+        end_es = convert_es_str(end)
+        query_body["query"]["bool"]["must"].append(
+            {
+                "range": {
+                    "datetime": {
+                        "gte": start_es,
+                        "lte": end_es
+                    }
+                }
+            }
+        )
+    if event_type:
+        query_body["query"]["bool"]["must"].append(
+            {
+                "terms": {
+                    "type.keyword": event_type
+                }
+            }
+        )
+    log.info(f"index:{index}--query_body:{query_body}")
+    async with EsUtil() as es:
+        es_re = await es.search_origin(body=query_body, index=index)
+    return es_re
+
+
+async def list_alarm_zdu_dao_new15(cid, points, start,
+                                   end, importance, event_type):
+    mid_li = [f"cid={cid}"]
+    if len(points) == 1:
+        mid_li.append(f"pid={points[0]}")
+    else:
+        mid_li.append(f"pid in {tuple(points)}")
+    if len(importance) == 1:
+        mid_li.append(f"importance={importance[0]}")
+    else:
+        mid_li.append(f"importance in {tuple(importance)}")
+    if event_type:
+        if len(event_type) == 1:
+            mid_li.append(f"event_type={event_type[0]}")
+        else:
+            mid_li.append(f"event_type in {tuple(event_type)}")
+    if start and end:
+        mid_li.append(f"event_datetime BETWEEN '{start}' and '{end}'")
+    sql = f"select * from point_1min_event where {' and '.join(mid_li)} " \
+          f"order by event_datetime desc "
+    async with MysqlUtil() as conn:
+        data = await conn.fetchall(sql)
+    return data
+
+
+async def wx_list_alarm_zdu_dao(cid, points, start, end):
+    """智电u, wx小程序, 取当前范围最新100条"""
+    
+    query_body = {
+        "size": 100,
+        "query": {
+            "bool": {
+                "must": [
+                    {
+                        "term": {
+                            "cid": cid
+                        }
+                    },
+                    {
+                        "terms": {
+                            "point_id": points
+                        }
+                    }
+                ]
+            }
+        },
+        "sort": [
+            {
+                "datetime": {
+                    "order": "desc"
+                }
+            }
+        ]
+    }
+    if start and end:
+        start_es = convert_es_str(start)
+        end_es = convert_es_str(end)
+        query_body["query"]["bool"]["must"].append(
+            {
+                "range": {
+                    "datetime": {
+                        "gte": start_es,
+                        "lte": end_es
+                    }
+                }
+            }
+        )
+    log.info(f"index:{index}--query_body:{query_body}")
+    async with EsUtil() as es:
+        es_re = await es.search_origin(body=query_body, index=index)
+    return es_re

unify_api/modules/alarm_manager/service/alarm_setting_service.py

+from hbmqtt.mqtt.constants import QOS_1
+from pot_libs.aiomqtt_util.hbmqtt_utils import MqttUtil
+from pot_libs.common.components.responses import Success
+from pot_libs.logger import log
+from pot_libs.mysql_util.mysql_util import MysqlUtil
+from pot_libs.utils.exc_util import BusinessException
+from unify_api.constants import EVENT_TYPE_SYNC_DEVICE
+from unify_api.modules.alarm_manager.dao.alarm_setting_dao import \
+    post_update_alarm_emq_dao
+from unify_api.modules.device_cloud.procedures.mqtt_helper import \
+    change_param_to_config
+
+
+async def post_update_alarm_emq_service(set_id, threshold, duration, enable,
+                                        cid, user_id, alarm_cid_dic):
+    """更新告警设置-直接与装置通信"""
+    # 1. 判断黑名单
+    # 逻辑: 工厂为99且user_id在[88, 16]中
+    if cid in alarm_cid_dic:
+        user_lis = alarm_cid_dic.get(cid)
+        if user_id in user_lis:
+            return Success(success=0, message="用户工厂在黑名单")
+    # 2. 根据id查询alarm_setting
+    alarm_dic = await post_update_alarm_emq_dao(set_id)
+    if not alarm_dic:
+        raise BusinessException(message=f"正在设置的报警设置id={set_id}不存在")
+
+    if not EVENT_TYPE_SYNC_DEVICE[alarm_dic["etype"]]:
+        raise BusinessException(
+            message=f"{alarm_dic} event_type表sync_device值不为1")
+    point_id = alarm_dic["pid"]
+    req_dic = {
+        "point_id": point_id,
+        "location_id": alarm_dic["lid"],
+        "event_type": alarm_dic["etype"],
+        "threshold": threshold,
+        "duration": duration,
+        "enable": enable,
+    }
+    # location_id和point_id只存在1个, 过滤掉另外一个参数
+    req_json = {k: v for k, v in req_dic.items() if v is not None}
+    # 3. 转换为与装置通信的报文
+    pub_json = await change_param_to_config(req_json, method="config")
+    if not pub_json:
+        raise BusinessException(
+            message=f"{point_id} 找不到meter信息")
+    # 4. 调用emq
+    try:
+        # client_name = "sync_web_client_1"
+        async with MqttUtil() as emq:
+            res_data = await emq.device_response(
+                sid=pub_json.get("sid"),
+                request_id=pub_json.get("request_id"),
+                data=pub_json
+            )
+    except TimeoutError:
+        return Success(message="time out 15s with device", success=0)
+
+    log.info(f'res_data: {res_data}')
+    if res_data["status_code"] != 200:
+        return Success(message="操作失败status=error, status_code=500", success=0)
+    # 5. update alarm_setting数据
+    if duration:
+        sql = "UPDATE soe_config_record SET threshold=%s, duration=%s, " \
+              "`enable`=%s WHERE id=%s"
+        sql_args = (threshold, duration, enable, set_id)
+    else:
+        sql = "UPDATE soe_config_record SET threshold=%s,enable=%s WHERE id=%s"
+        sql_args = (threshold, enable, set_id)
+    async with MysqlUtil() as conn:
+        modified_cnt = await conn.execute(sql, args=sql_args)
+        log.info(f"UPDATE alarm_setting modified_cnt={modified_cnt}")
+    return Success(success=1, message="操作成功")

unify_api/modules/alarm_manager/views/__init__.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+

unify_api/modules/alarm_manager/views/alarm_setting.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+import json
+
+from pot_libs.aiohttp_util.aiohttp_utils import AioHttpUtils
+from pot_libs.mysql_util.mysql_util import MysqlUtil
+from pot_libs.common.components.responses import Success
+from pot_libs.sanic_api import summary, examples
+from pot_libs.logger import log
+from pot_libs.settings import SETTING
+from pot_libs.utils.exc_util import BusinessException
+from unify_api.constants import EVENT_TYPE_SYNC_DEVICE, RESIDUAL_CURRENT_OP
+from unify_api.modules.alarm_manager.components.alarm_setting import (
+    AlarmSettingUpdate,
+    PointAlarmSettingRequest,
+    PointAlarmSettingResponse,
+    AlarmSetting,
+    point_alarm_setting_example
+)
+from unify_api.modules.alarm_manager.service.alarm_setting_service import \
+    post_update_alarm_emq_service
+
+
+@summary('获取某监测点告警设置列表')
+@examples(point_alarm_setting_example)
+async def post_point_alarm_setting(req, body: PointAlarmSettingRequest) \
+        -> PointAlarmSettingResponse:
+    point_id = body.point_id
+    location_ids = body.location_ids
+    enable = body.enable
+    point_alarm_settings = []
+    location_alarm_settings = []
+    if enable in (0, 1):
+        sql_point = "SELECT s.id,s.etype,s.threshold,s.duration,s.enable," \
+                    "e.name,e.importance FROM soe_config_record s " \
+                    "INNER JOIN event_type e ON s.etype=e.e_type  " \
+                    "WHERE s.pid=%s AND s.enable=%s"
+
+        sql_location = "SELECT l.item,s.id,s.etype,e.name,e.importance, " \
+                       "s.threshold, s.duration, s.enable " \
+                       "FROM soe_config_record s INNER JOIN event_type e " \
+                       "ON s.`etype`=e.e_type INNER JOIN location l " \
+                       "on l.lid=s.lid " \
+                       "WHERE s.lid IN %s AND s.enable=%s"
+    else:
+        sql_point = "SELECT s.id, s.etype, e.name, e.importance, " \
+                    "s.threshold, s.duration, s.enable " \
+                    "FROM soe_config_record s INNER JOIN event_type e " \
+                    "ON s.`etype`=e.e_type " \
+                    "WHERE s.pid=%s"
+        sql_location = "SELECT l.item,l.ad_type as location_type,s.id, " \
+                       "s.etype,e.name,e.importance,s.threshold,s.duration, " \
+                       "s.enable FROM soe_config_record s INNER JOIN " \
+                       "event_type e ON s.`etype`=e.e_type INNER JOIN " \
+                       "location l on l.lid=s.lid WHERE s.lid IN %s"
+    try:
+        async with MysqlUtil() as conn:
+            if enable in (0, 1):
+                res_point = await conn.fetchall(sql_point, args=(
+                    point_id, enable,)) if point_id else {}
+                res_location = await conn.fetchall(sql_location, args=(
+                    tuple(location_ids), enable,)) if location_ids else {}
+            else:
+                res_point = await conn.fetchall(sql_point, args=(
+                    point_id,)) if point_id else {}
+                res_location = await conn.fetchall(sql_location, args=(
+                    tuple(location_ids),)) if location_ids else {}
+            for res in res_point:
+                alarm_setting = AlarmSetting(
+                    id=res.get("id"),
+                    type=res.get("etype"),
+                    name=res.get("name"),
+                    level=res.get("importance"),
+                    threshold=res.get("threshold"),
+                    duration=res.get("duration", None),
+                    enable=res.get("enable"),
+                )
+                point_alarm_settings.append(alarm_setting)
+
+            for res in res_location:
+                alarm_setting = AlarmSetting(
+                    id=res.get("id"),
+                    type=res.get("etype"),
+                    name=(res.get("item") + res.get("name")) if res.get(
+                        "item") != "default" else res.get("name"),
+                    level=res.get("importance"),
+                    threshold=res.get("threshold"),
+                    duration=res.get("duration", None),
+                    enable=res.get("enable"))
+                location_alarm_settings.append(alarm_setting)
+    except Exception as e:
+        log.exception(e)
+        return PointAlarmSettingResponse().db_error()
+    residual_current_op = 0
+    user_id = req.ctx.user_id
+    if user_id in RESIDUAL_CURRENT_OP:
+        residual_current_op = 1
+    return PointAlarmSettingResponse(
+        residual_current_op=residual_current_op,
+        point_alarm_settings=point_alarm_settings,
+        location_alarm_settings=location_alarm_settings
+    )
+
+
+# @summary('更新告警设置')
+# async def post_update_alarm(req, body: AlarmSettingUpdate) -> Success:
+#     id = body.id
+#     threshold = body.threshold
+#     duration = body.duration
+#     enable = body.enable
+#     cid = body.cid
+#     # user_id = 88
+#     user_id = req.ctx.user_id
+#     alarm_cid_dic = SETTING.alarm_cid
+#     # 1. 判断黑名单
+#     # 逻辑: 工厂为99且user_id在[88, 16]中
+#     if cid in alarm_cid_dic:
+#         user_lis = alarm_cid_dic.get(cid)
+#         if user_id in user_lis:
+#             return Success(success=0, message="用户工厂在黑名单")
+#     # 2. 调用后台接口，传数给装置
+#     async with MysqlUtil() as conn:
+#         query_sql = "select `point_id`, `location_id`, `type` " \
+#                     "from `alarm_setting` where `id`=%s"
+#         alarm_map = await conn.fetchone(query_sql, args=(id,))
+#     if not alarm_map:
+#         raise BusinessException(message=f"正在设置的报警设置id={id}不存在")
+#
+#     if alarm_map and EVENT_TYPE_SYNC_DEVICE[alarm_map["type"]]:
+#         req_json = {
+#             "events": [
+#                 {
+#                     k: v for k, v in {
+#                         "point_id": alarm_map["point_id"],
+#                         "location_id": alarm_map["location_id"],
+#                         "event_type": alarm_map["type"],
+#                         "threshold": threshold,
+#                         "duration": duration,
+#                         "enable": enable,
+#                     }.items() if v is not None}]}
+#         try:
+#             log.info(
+#                 f"post_update_alarm url={SETTING.event_post_url} "
+#                 f"req_json = {req_json}")
+#             resp_str, status = await AioHttpUtils().post(
+#                 SETTING.event_post_url,
+#                 req_json,
+#                 timeout=50,
+#             )
+#             log.info(f"post_update_alarm resp_str={resp_str} status={status}")
+#             resp = json.loads(resp_str)
+#         except Exception as e:
+#             log.exception(e)
+#             return Success(success=0, message=f"操作失败{str(e)}")
+#
+#         if resp[0].get("timeout"):
+#             return Success(
+#                 message=f"操作失败status=error, error_data={resp[0]['timeout']}",
+#                 success=0)
+#         if resp[0].get("status_code") == 500:
+#             return Success(message="操作失败status=error, status_code=500",
+#                            success=0)
+#
+#     if duration:
+#         sql = "UPDATE alarm_setting " \
+#               "SET threshold=%s, duration=%s, `enable`=%s " \
+#               "WHERE id=%s"
+#         sql_args = (threshold, duration, enable, id,)
+#     else:
+#         sql = "UPDATE alarm_setting SET threshold=%s, `enable`=%s WHERE id=%s"
+#         sql_args = (threshold, enable, id,)
+#     try:
+#         async with MysqlUtil() as conn:
+#             modified_cnt = await conn.execute(sql, args=sql_args)
+#             log.info(f"UPDATE alarm_setting modified_cnt={modified_cnt}")
+#             # 这里执行一个update,result 返回modified_cnt， 当result=0时也是成功的，因为没引发异常
+#             return Success(success=1, message="操作成功")
+#     except Exception as e:
+#         log.exception(e)
+#         return Success(success=0, message=f"操作失败{str(e)}")
+
+
+@summary('更新告警设置-直接与装置通信')
+async def post_update_alarm(req, body: AlarmSettingUpdate) -> Success:
+    set_id = body.id  # alarm_setting表id
+    threshold = body.threshold
+    duration = body.duration
+    enable = body.enable
+    cid = body.cid
+    user_id = req.ctx.user_id
+    # user_id = 88
+    alarm_cid_dic = SETTING.alarm_cid
+    return await post_update_alarm_emq_service(set_id, threshold, duration,
+                                               enable, cid, user_id,
+                                               alarm_cid_dic)

unify_api/modules/alarm_manager/views/alarm_static.py

+from pot_libs.mysql_util.mysql_util import MysqlUtil
+from pot_libs.sanic_api import summary
+from unify_api.modules.alarm_manager.components.alarm_static_cps import \
+    SduAlarmReq, SduAlarmResp, ContentName, RiskCount, SasReq, SassReq, \
+    SassResp, AppReq, AppResp, SisReq, SisResp, SiasReq, SiasResp, SebReq, \
+    SebResp, SiarResp, ZsResp, ZasReq, ZasResp
+from unify_api.modules.alarm_manager.procedures.alarm_static_pds import \
+    alarm_content_info, risk_distribution
+from unify_api.modules.alarm_manager.service.alarm_static_service import \
+    sdu_alarm_statistics_service, sdu_alarm_statistics_sort_service, \
+    sdu_app_statistics_sort_service, sdu_electric_behave_service, \
+    sdu_index_alarm_ranking_service, zdu_level_distribution_service, \
+    zdu_content_distribution_service, zdu_summary_service, \
+    zdu_alarm_sort_service, sdu_alarm_statistics_sort_service_new15, \
+    sdu_app_statistics_sort_service_new15, \
+    sdu_index_alarm_ranking_service_new15
+from unify_api.modules.home_page.components.security_info_cps import \
+    SecurityCountResp, AlarmContentDistributionResp
+from unify_api.modules.home_page.procedures.count_info_pds import \
+    electric_use_info_sdu, electric_use_info_sdu_new15
+from unify_api.utils.time_format import last30_day_range
+
+
+@summary("报警统计")
+async def post_alarm_statistics(req, body: SduAlarmReq) -> SduAlarmResp:
+    """目前用于 识电u->报警统计"""
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    # 获取point信息
+    sql_point = "SELECT pid, name FROM point WHERE cid=%s"
+    async with MysqlUtil() as conn:
+        points = await conn.fetchall(sql_point, args=(cid,))
+    if not points:
+        return SduAlarmResp()
+    point_id_list = [i["pid"] for i in points]
+    # 1.调用函数获取报警统计信息
+    alarm_info_map = await alarm_content_info(cid, start, end, points)
+    if not alarm_info_map:
+        return SduAlarmResp()
+    high_power_app, ele_overload, illegal_ele_app, power_quality, \
+    ele_car_battery, point_dic, total_alarm_cnt = (
+        alarm_info_map["high_power_app"],
+        alarm_info_map["ele_overload"],
+        alarm_info_map["illegal_ele_app"],
+        alarm_info_map["power_quality"],
+        alarm_info_map["ele_car_battery"],
+        alarm_info_map["point_dic"],
+        alarm_info_map["total_alarm_cnt"]
+    )
+    # point_dic排序
+    point_list = []
+    for key, val in point_dic.items():
+        point_list.append({"name": key, "value": val})
+    point_list.sort(key=lambda x: x["value"], reverse=True)
+    # 报警类型总数,用于求比例
+    cn = ContentName(ele_overload=sum(ele_overload["value"]),
+                     high_power_app=sum(high_power_app["value"]),
+                     illegal_ele_app=sum(illegal_ele_app["value"]),
+                     power_quality=sum(power_quality["value"]),
+                     ele_car_battery=sum(ele_car_battery["value"])
+                     )
+    # 2.计算风险分布
+    # 其中当前时段为发生过I级报警为风险用户，其余为安全用户
+    security_user, risk_user = await risk_distribution(start, end,
+                                                       point_id_list)
+    rc = RiskCount(security_user=security_user, risk_user=risk_user)
+    return SduAlarmResp(
+        total_alarm_cnt=total_alarm_cnt,
+        high_power_app=high_power_app,
+        ele_overload=ele_overload,
+        illegal_ele_app=illegal_ele_app,
+        power_quality=power_quality,
+        ele_car_battery=ele_car_battery,
+        alarm_ranking=point_list,
+        content_distribution=cn,
+        risk_distribution=rc
+    )
+
+
+@summary("报警统计-识电u")
+async def post_sdu_alarm_statistics(req, body: SduAlarmReq) -> SduAlarmResp:
+    """目前用于 识电u->报警统计"""
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    return await sdu_alarm_statistics_service([cid], start, end, product)
+
+
+@summary("报警统计-wx-识电u")
+async def post_sdu_alarm_statistics_wx(req, body: SasReq) -> SduAlarmResp:
+    """目前用于 识电u->报警统计"""
+    product = body.product
+    cids = body.cids
+    start = body.start
+    end = body.end
+    return await sdu_alarm_statistics_service(cids, start, end, product)
+
+
+@summary("报警统计-报警记录-排名-识电u")
+async def post_sdu_alarm_statistics_sort(req,
+                                         body: SassReq) -> SassResp:
+    """目前用于 识电u->报警统计"""
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    page_size = body.page_size
+    page_num = body.page_num
+    sort = body.sort
+    # return await sdu_alarm_statistics_sort_service(cid, start, end, page_size,
+    #                                                page_num, sort)
+    return await sdu_alarm_statistics_sort_service_new15(cid, start, end,
+                                                         page_size, page_num,
+                                                         sort)
+
+
+@summary("报警统计-电器识别-排名-识电u")
+async def post_sdu_app_statistics_sort(req, body: AppReq) -> AppResp:
+    """目前用于 识电u->报警统计"""
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    return await sdu_app_statistics_sort_service_new15(cid, start, end)
+
+
+@summary("首页-运行趋势-识电u")
+async def post_sdu_index_statistics(req, body: SisReq) -> SisResp:
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    sr = await sdu_alarm_statistics_service([cid], start, end, product)
+    return SisResp(
+        ele_overload=sr.ele_overload,
+        illegal_ele_app=sr.illegal_ele_app,
+        power_quality=sr.power_quality
+    )
+
+
+@summary("首页-报警统计-新版识电u")
+async def post_sdu_index_alarm_statistics(req, body: SiasReq) -> SiasResp:
+    # 最近30天, 不包含今天
+    start, end = last30_day_range()
+    product = body.product
+    cid = body.cid
+    # 安全和报警统计
+    res = await sdu_alarm_statistics_service([cid], start, end, product)
+    # 安全指数
+    alarm_res = await electric_use_info_sdu_new15(cid)
+    electric_use_score = round(alarm_res.electric_use_score)
+    return SiasResp(
+        risk_distribution=res.risk_distribution,
+        content_distribution=res.content_distribution,
+        electric_use_score=electric_use_score
+    )
+
+
+@summary("用电行为-卡片数据-新版识电u")
+async def post_sdu_electric_behave(req, body: SebReq) -> SebResp:
+    """近30天用电行为"""
+    cid = body.cid
+    storeys = body.storeys
+    product = body.product
+    # 最近30天, 不包含今天
+    start, end = last30_day_range()
+    return await sdu_electric_behave_service(cid, start, end, storeys, product)
+
+
+@summary("首页-报警违规排名-新版识电u")
+async def post_sdu_index_alarm_ranking(req, body: SiasReq) -> SiarResp:
+    """近30天用电行为"""
+    cid = body.cid
+    product = body.product
+    # 最近30天, 不包含今天
+    start, end = last30_day_range()
+    return await sdu_index_alarm_ranking_service_new15(cid, start, end, product)
+
+
+@summary("报警统计-报警等级-智电u")
+async def post_zdu_level_distribution(req,
+                                      body: SduAlarmReq) -> SecurityCountResp:
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    return await zdu_level_distribution_service(cid, start, end, product)
+
+
+@summary("报警统计-报警内容-智电u")
+async def post_zdu_content_distribution(req,
+                                        body: SduAlarmReq) \
+        -> AlarmContentDistributionResp:
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    return await zdu_content_distribution_service(cid, start, end, product)
+
+
+@summary("报警统计-统计概况信息-智电u")
+async def post_zdu_summary(req, body: SduAlarmReq) -> ZsResp:
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    return await zdu_summary_service(cid, start, end, product)
+
+
+@summary("报警统计-报警排名-智电u")
+async def post_zdu_alarm_sort(req, body: ZasReq) -> ZasResp:
+    product = body.product
+    cid = body.cid
+    start = body.start
+    end = body.end
+    page_size = body.page_size
+    page_num = body.page_num
+    return await zdu_alarm_sort_service(cid, start, end, page_size, page_num)

unify_api/modules/alarm_manager/views/list_alarm.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/9
+import random
+from datetime import datetime, timedelta
+
+from pot_libs.sanic_api import summary, description, examples
+from pot_libs.es_util.es_utils import EsUtil
+from pot_libs.logger import log
+from pot_libs.es_util.es_query import EsQuery
+from pot_libs.utils.exc_util import DBException, BusinessException
+from unify_api.constants import Product
+from unify_api.modules.alarm_manager.components.alarm_static_cps import NlaReq, \
+    WlaReq, LazReq
+from unify_api.modules.alarm_manager.service.list_alarm_service import \
+    new_list_alarm_service, wx_list_alarm_service, list_alarm_zdu_service, \
+    wx_list_alarm_zdu_service, list_alarm_service_new15
+from unify_api.modules.common.procedures.cids import get_cid_info, get_cids, \
+    get_proxy_cids
+from unify_api.utils import time_format
+from unify_api import constants
+from pot_libs.common.components.query import PageRequest, Equal, Range, Filter, \
+    InGroup
+from unify_api.modules.alarm_manager.components.list_alarm import (
+    ListAlarmResponse, ListAlarmReq,
+    Alarm,
+    list_alarm_example, WlazReq,
+)
+
+
+@summary("返回报警信息列表")
+@description("筛选字段：监测点")
+# @examples(list_alarm_example)
+async def post_list_alarm(req, body: ListAlarmReq) -> ListAlarmResponse:
+    cid = body.cid
+    point_id = body.point_id
+    start = body.start
+    end = body.end
+    importance = body.importance
+    page_size = body.page_size
+    page_num = body.page_num
+    alarm_type = body.alarm_type
+    return await list_alarm_service_new15(cid, point_id, start, end,
+                                          importance, page_size, page_num,
+                                          alarm_type)
+
+
+@summary("上面的备份")
+@description("筛选字段：监测点")
+@examples(list_alarm_example)
+async def post_list_alarm_bak(req, body: PageRequest) -> ListAlarmResponse:
+    # 将"2020-05-02 17:32:31”格式的时间转化为"2020-05-02T17:32:31+08:00"
+    # 小程序首页展示报警信息列表最新40条，不需要传时间
+    if body.filter.ranges:
+        _range = body.filter.ranges[0]
+        field, start, end = _range.field, _range.start, _range.end
+        start_dt = time_format.convert_to_dt(start)
+        start_str = time_format.convert_dt_to_str(start_dt, date_type="tz")
+        end_dt = time_format.convert_to_dt(end)
+        end_str = time_format.convert_dt_to_str(end_dt, date_type="tz")
+        _range = Range(field=field, start=start_str, end=end_str)
+        ranges = [_range]
+    else:
+        ranges = []
+
+    # TODO:当没有选择监测点的时候，需要从req取cid，作为筛选条件
+    if req.json.get("cid"):
+        cid = req.json["cid"]
+        equal = Equal(field="cid", value=cid)
+        body.filter.equals.append(equal)
+
+    if req.json.get("product") == Product.AndianUManage.value:
+        proxy_id = req.json.get("proxy_id")
+        product = req.json.get("product")
+        user_id = req.ctx.user_id
+        req_cids = req.json.get("cids")
+
+        # cids = await get_cids(user_id, product)
+        cids = await get_proxy_cids(user_id, product, proxy_id)
+        if req_cids:
+            if any(i for i in req_cids if i not in cids):
+                raise BusinessException(
+                    message=f"你没有工厂{set(req_cids) - set(cids)}没有权限")
+            in_group = InGroup(field="cid", group=req_cids)
+        else:
+            in_group = InGroup(field="cid", group=cids)
+        body.filter.in_groups.append(in_group)
+
+    if req.json.get("product") in [Product.RecognitionElectric.value,
+                                   Product.IntelligentU.value]:
+        req_cid = req.json.get("cid")
+        if not req_cid:
+            product = req.json.get("product")
+            user_id = req.ctx.user_id
+            cids = await get_cids(user_id, product)
+        else:
+            cids = [req_cid]
+            in_group = InGroup(field="cid", group=cids)
+            body.filter.in_groups.append(in_group)
+
+    filter = Filter(
+        equals=body.filter.equals,
+        ranges=ranges,
+        in_groups=body.filter.in_groups,
+        keywords=body.filter.keywords,
+    )
+
+    # 重新封装PageRequest
+    page_request = PageRequest(
+        page_size=body.page_size, page_num=body.page_num, sort=body.sort,
+        filter=filter
+    )
+    query_body = EsQuery().query(page_request)
+    if not query_body.get("query"):
+        query = {"bool": {"must_not": [{"terms": {"mode.keyword": ["scope"]}}]}}
+        query_body["query"] = query
+    else:
+        must_not = [{"terms": {"mode.keyword": ["scope"]}}]
+        query_body["query"]["bool"]["must_not"] = must_not
+    index = "poweriot_point_1min_event"
+    log.info(f"index:{index}--query_body:{query_body}")
+    async with EsUtil() as es:
+        es_results = await es.search_origin(body=query_body, index=index)
+        if not es_results:
+            log.warning(
+                "Can not find data on es(index: %s): %s" % (index, query_body))
+            raise DBException
+
+        cid_info_map = await get_cid_info(all=True)
+        rows = []
+        for info in es_results["hits"]["hits"]:
+            es_id = info["_id"]
+            source = info["_source"]
+            type = source.get("type")
+            mode = source.get("mode")
+            type_str = constants.EVENT_TYPE_MAP.get(type, type)
+            point_id = source.get("point_id")
+            location_id = source.get("location_id")
+            date_time = source.get("datetime")
+            dt = time_format.convert_to_dt(date_time)
+            date_time = time_format.convert_dt_to_timestr(dt)
+            event_duration = source.get("event_duration")
+            if point_id and mode == "scope":
+                url = "/scope_details?doc_id=%s" % es_id
+                redirect_type = "scope"
+            elif location_id and type in constants.TEMP_SCOPE_URL_TYPE:
+                url = "/temp_trend?doc_id=%s" % es_id
+                redirect_type = "temp_trend"
+            else:
+                url = None
+                redirect_type = ""
+            cid = int(source.get("cid")) if source.get("cid") else source.get(
+                "cid")
+            alarm = Alarm(
+                name=source.get("name"),
+                importance=source.get("importance"),
+                date_time=date_time,
+                type=type,
+                type_name=type_str,
+                description=source.get("message"),
+                redirect_type=redirect_type,
+                es_id=es_id,
+                url=url,
+                event_duration=round(event_duration)
+                if isinstance(event_duration, float) else event_duration,
+                company_name=cid_info_map.get(cid, {}).get("fullname", ""),
+            )
+            rows.append(alarm)
+
+        real_total = es_results["hits"]["total"]
+        total = real_total if real_total < constants.ES_TOTAL_LIMIT else constants.ES_TOTAL_LIMIT
+        return ListAlarmResponse(total=total, rows=rows)
+
+
+@summary("报警记录-列表分页")
+async def post_new_list_alarm(req, body: NlaReq) -> ListAlarmResponse:
+    # 1. 获取参数
+    cid = body.cid
+    storeys = body.storeys
+    page_num = body.page_num
+    page_size = body.page_size
+    start = body.start
+    end = body.end
+    product = body.product
+    importance = body.importance
+    point_ids = body.point_ids
+    # 2. 调用service
+    return await new_list_alarm_service(cid, storeys,
+                                        (page_num - 1) * page_size, page_size,
+                                        start, end, importance, point_ids,
+                                        product)
+
+
+
+@summary("小程序消息列表")
+async def post_wx_list_alarm(req, body: WlaReq) -> ListAlarmResponse:
+    # 1. 获取参数
+    cids = body.cids
+    product = body.product
+    # 如果没有传start和end, 取最新40条, 传了start/end取范围内最新40条
+    start = body.start
+    end = body.end
+    # 2. 调用service
+    return await wx_list_alarm_service(cids, product, start, end)
+
+
+@summary("报警记录-列表分页-智电u")
+async def post_list_alarm_zdu(req, body: LazReq) -> ListAlarmResponse:
+    # 1. 获取参数
+    cid = body.cid
+    point_list = body.point_list
+    page_num = body.page_num
+    page_size = body.page_size
+    start = body.start
+    end = body.end
+    importance = body.importance
+    event_type = body.event_type
+    # 2. 调用service
+    return await list_alarm_zdu_service(cid, point_list, page_num, page_size,
+                                        start, end, importance, event_type)
+
+
+@summary("小程序消息列表-智电u")
+async def post_wx_list_alarm_zdu(req, body: WlazReq) -> ListAlarmResponse:
+    # 1. 获取参数
+    cid = body.cid
+    point_list = body.point_list
+    # 如果没有传start和end, 取最新100条, 传了start/end取范围内最新100条
+    start = body.start
+    end = body.end
+    # 2. 调用service
+    return await wx_list_alarm_zdu_service(cid, point_list, start, end)

unify_api/modules/carbon_neutral/components/carbon_reduce_cps.py

+from dataclasses import dataclass
+from pot_libs.sanic_api import Model
+from pot_libs.sanic_api.column import Str, Int, Float, Dict
+
+
+@dataclass
+class CiResp(Model):
+    total_area: int = Int("总面积").eg(120000)
+    total_kwh: float = Float("总能耗").eg(100000.23)
+    energy_density: float = Float("能耗密度").eg(5.12)
+    conversion_year: float = Float("折算年").eg(60.23)
+    energy_standard: float = Float("能耗标准").eg(75.12)
+    env_benefit: str = Str("环境效益").eg("符合能耗标准，请继续保持")
+
+
+@dataclass
+class TaResp(Model):
+    working_day: dict = Dict("工作日").eg(
+        {"slots": [], "values": [], "avg_power": 110.12}
+    )
+    non_working_day: dict = Dict("非工作日").eg(
+        {"slots": [], "values": [], "avg_power": 110.12}
+    )
+
+
+@dataclass
+class CeiResp(Model):
+    carbon_index: int = Int("碳排指数").eg(62)
+    carbon_status: str = Str("达标情况").eg("达标")
+    conversion_year: float = Float("折算年").eg(60.23)
+    energy_standard: float = Float("能耗标准").eg(75.12)

unify_api/modules/carbon_neutral/procedures/carbon_reduce_pds.py

+from pot_libs.logger import log
+from unify_api.constants import REDUCE_CO2
+from unify_api.utils.common_utils import round_2
+
+
+def env_benefit_ca(conversion_year, energy_standard, total_area):
+    """环境效益逻辑"""
+    # 1. 若折算年 < 能耗标准，环境效益的文字显示为“相比能耗标准，
+    # 节约用电XX万kWh，减少排放XX万kg二氧化碳，相当于多种XX棵树”；
+    # 2. 若折算年 > 能耗标准，环境效益的文字显示为“超出能耗标准，请注意控制节能降耗”；
+    # 3. 若折算年 = 能耗标准，环境效益的文字显示为“符合能耗标准，请继续保持”。
+    if not all([conversion_year, energy_standard, total_area]):
+        log.warning("能耗标准,折算年, 存在空值")
+        return ""
+    if conversion_year == energy_standard:
+        return "符合能耗标准，请继续保持"
+    elif conversion_year > energy_standard:
+        return "超出能耗标准，请注意控制节能降耗"
+    else:
+        # 1. 节约用电, (能耗标准 - 折算年) * 总面积
+        diff_kwh = (energy_standard - conversion_year)*total_area  # 节约用电int
+        if diff_kwh > 10000:
+            diff_kwh_str = str(round_2(diff_kwh / 10000)) + "万"
+        else:
+            diff_kwh_str = round_2(diff_kwh)
+        # 2. 减少二氧化碳排放
+        # 当用户节约1kWh电量时，减少碳排放0.2047kg；减少二氧化碳排放0.751kg
+        reduce_c02 = round(diff_kwh * REDUCE_CO2)  # 减少二氧化碳排放
+        if reduce_c02 > 10000:
+            reduce_c02_str = str(round_2(reduce_c02 / 10000)) + "万"
+        else:
+            reduce_c02_str = round_2(reduce_c02)
+        # 3. 植树折算
+        """
+        折算效益时，时间粒度为年则取18.3kg；
+        时间粒度为月则取1.525kg；
+        时间粒度为天则取0.05kg。
+        测算时段内的二氧化碳排放量除以上述系数得到相当于种多少棵树
+        """
+        add_tree = round(reduce_c02 / 1.525)  # 多种XX棵树int
+        # 多种数
+        return f"低于能耗标准用电{diff_kwh_str}kWh，" \
+               f"减少排放{reduce_c02_str}kg二氧化碳，" \
+               f"相当于多种{add_tree}棵树"

unify_api/modules/carbon_neutral/service/carbon_reduce_service.py

+from numpy import mean
+
+from unify_api.modules.alarm_manager.dao.alarm_setting_dao import \
+    company_extend_dao
+from unify_api.modules.carbon_neutral.components.carbon_reduce_cps import \
+    CiResp, TaResp, CeiResp
+from unify_api.modules.carbon_neutral.procedures.carbon_reduce_pds import \
+    env_benefit_ca
+from unify_api.modules.common.procedures.common_utils import carbon_index
+from unify_api.modules.elec_charge.dao.elec_charge_dao import \
+    point_day_power_dao, get_total_kwh_dao
+from unify_api.modules.home_page.procedures.count_info_pds import \
+    carbon_status_res
+from unify_api.utils.common_utils import round_2, division_two, \
+    multiplication_two, ChineseCalendar
+from unify_api.utils.time_format import day_of_month, proxy_power_slots
+
+
+async def carbon_summary_service(cid, start, end):
+    """碳中和-统计概况信息"""
+    # 1. 总面积, 能耗标准
+    extend_dic = await company_extend_dao(cid)
+    total_area = ""
+    energy_standard = ""
+    for info in extend_dic:
+        key = info["key"]
+        if key == "total_area":
+            total_area = int(info["value"])
+        if key == "energy_standard":
+            energy_standard = int(info["value"])
+    # 2.总能耗, 即电量
+    res_kwh = await get_total_kwh_dao(cid, start, end)
+    # res_kwh = await query_charge_aggs(start, end, [cid])
+    total_kwh = round_2(res_kwh["total_kwh"]) if res_kwh else ""
+    # 3. 能耗密度, 为总能耗除以总面积结果
+    energy_density = round_2(division_two(total_kwh, total_area))
+    # 4. 折算年, 为能耗密度除以当月天数再乘以365
+    m_number = day_of_month(start)
+    conversion_year = round_2(
+        multiplication_two(division_two(energy_density, m_number), 365))
+    # 5. 环境效益
+    env_benefit = env_benefit_ca(conversion_year, energy_standard, total_area)
+    return CiResp(
+        total_area=total_area,
+        total_kwh=total_kwh,
+        energy_density=energy_density,
+        conversion_year=conversion_year,
+        energy_standard=energy_standard,
+        env_benefit=env_benefit
+    )
+
+
+async def trend_analysis_service(cid, start, end):
+    """碳中和-趋势分析"""
+    slots = proxy_power_slots(start, end, "YYYY-MM-DD")
+    kwh_info = await point_day_power_dao(cid, start, end)
+    kwh_create_time = [kwh["create_time"] for kwh in kwh_info]
+    kwh_info = [kwh["kwh"] for kwh in kwh_info]
+    working_day = {"slots": [], "values": [], "avg_power": ""}
+    non_working_day = {"slots": [], "values": [], "avg_power": ""}
+    for slot in slots:
+        if ChineseCalendar(slot).is_workday():
+            working_day["slots"].append(slot[5:])
+            if slot in kwh_create_time:
+                index = kwh_create_time.index(slot)
+                working_day["values"].append(round_2(kwh_info[index]))
+            else:
+                working_day["values"].append("")
+        else:
+            non_working_day["slots"].append(slot[5:])
+            if slot in kwh_create_time:
+                index = kwh_create_time.index(slot)
+                non_working_day["values"].append(round_2(kwh_info[index]))
+            else:
+                non_working_day["values"].append("")
+    # 2. 补充平均值
+    working_value = working_day["values"]
+    working_value_after = [i for i in working_value if i]
+    if working_value_after:
+        working_avg_power = round_2(mean(working_value_after))
+        working_day["avg_power"] = working_avg_power
+
+    non_working_value = non_working_day["values"]
+    non_working_value_after = [i for i in non_working_value if i]
+    if non_working_value_after:
+        non_working_avg_power = round_2(
+            mean(non_working_value_after))
+        non_working_day["avg_power"] = non_working_avg_power
+
+    return TaResp(
+        working_day=working_day,
+        non_working_day=non_working_day
+    )
+
+
+async def carbon_emission_index_service(cid, start, end):
+    """首页-碳排指数等信息"""
+    carbon_info = await carbon_summary_service(cid, start, end)
+    # 1. 能耗标准
+    energy_standard = carbon_info.energy_standard
+    # 2. 折算年
+    conversion_year = carbon_info.conversion_year
+    # 3. 碳排指数
+    # 3.1 能耗偏差
+    if energy_standard:
+        deviation = (energy_standard - conversion_year) / energy_standard
+        # 3.2 碳排指数
+        c_index = carbon_index(deviation)
+        # 4. 达标情况
+        carbon_status = carbon_status_res(c_index)
+    else:
+        deviation, c_index, carbon_status = "", "", ""
+    return CeiResp(
+        carbon_index=c_index,
+        carbon_status=carbon_status,
+        conversion_year=conversion_year,
+        energy_standard=energy_standard
+    )

unify_api/modules/common/__init__.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/8

unify_api/modules/common/components/__init__.py

+# -*- coding:utf-8 -*-
+#
+# Author:jing
+# Date: 2020/7/8