import json
import time
from datetime import datetime, timedelta
from math import sqrt
import pendulum

from pot_libs.settings import SETTING
from unify_api.modules.electric.dao.electric_dao import \
    get_elec_mtid_sid_by_cid
from unify_api.utils.common_utils import round_2
from pot_libs.aredis_util.aredis_utils import RedisUtils
from pot_libs.common.components.query import Range, Equal, Filter, PageRequest
from pot_libs.es_util.es_query import EsQuery
from pot_libs.es_util.es_utils import EsUtil
from pot_libs.logger import log
from pot_libs.mysql_util.mysql_util import MysqlUtil
from unify_api import constants
from unify_api.constants import Importance, EVENT_TYPE_MAP, SDU_ALARM_LIST
from unify_api.modules.alarm_manager.dao.list_static_dao import \
    sdu_alarm_importance_dao, alarm_aggs_importance, \
    sdu_alarm_importance_dao_new15
from unify_api.modules.common.dao.common_dao import monitor_point_join, \
    monitor_by_cid
from unify_api.modules.common.procedures.common_utils import get_electric_index
from unify_api.modules.common.procedures.points import get_points, \
    proxy_points, get_points_num
from unify_api.modules.common.procedures.pttl_max import pttl_max, \
    pttl_max_new15
from unify_api.modules.electric.views.electric import METERDATA_CURRENT_KEY
from unify_api.modules.home_page.components.count_info_cps import (
    MaxResidualCurrent,
    ElectricInfo,
)
from unify_api.utils.time_format import last30_day_range, convert_es_str
from unify_api.modules.home_page.dao.count_info_dao import (
    get_inline_by_cid, get_power_factor_kpi, get_pcvf_kpi, get_economic_kpi,
    get_md_space, get_tc_runtime
)
from unify_api.modules.electric_optimization.dao.power_index import (
    price_policy_by_cid
)
from unify_api.utils.taos_new import parse_td_columns, get_td_table_name, \
    td3_tbl_compate, get_td_engine_data


async def other_info(company_id):
    """
    今日报警数和累计安全运行天数,报警数
    """
    alarm_sql = f"""
        SELECT
            DATE(pevent.event_datetime) event_date,
            COUNT(*) event_count
        FROM
            point_1min_event pevent
        WHERE
            cid = %s
        GROUP BY
            DATE(pevent.event_datetime)
    """
    
    now_time = datetime.now()
    # 获取到工厂安装时间create_time
    async with MysqlUtil() as conn:
        company_sql = "select create_time from company where cid = %s"
        alarm_data = await conn.fetchone(alarm_sql, (company_id,))
        company = await conn.fetchone(company_sql, (company_id,))
    create_time_timestamp = company["create_time"]
    create_time = datetime.fromtimestamp(create_time_timestamp)
    
    today_alarm_count = 0
    alarm_count = 0
    if not alarm_data:
        log.warn(
            "No alarm data %s" % (company_id)
        )
        # 1. 增加逻辑，新增工厂如果还没有事件产生
        # 系统安全运行天数： 当前时间 - 工厂安装时间 + 1
        safe_run_days = (now_time - create_time).days + 1
        return today_alarm_count, safe_run_days, alarm_count
    
    # 5. 构造返回
    # 如果每天都有报警, 防止安全运行天数-1天, 所以total_days +2
    total_days = (now_time - create_time).days + 2
    has_alarm_days = 0
    for data in alarm_data:
        create_time.strftime("%Y-%m-%d")
        if data["event_date"].strftime("%Y-%m-%d") == str(now_time)[:10]:
            today_alarm_count += data["event_count"]
        if data["event_count"] != 0:
            # 没有报警，看做是安全运行了,统计累计安全运行的天数
            has_alarm_days += 1
        alarm_count += data["event_count"]
    safe_run_days = total_days - has_alarm_days
    log.info(
        f"today_alarm_count={today_alarm_count} safe_run_days={safe_run_days}")
    return today_alarm_count, safe_run_days, alarm_count


async def other_info_old(company_id):
    """
    今日报警数和累计安全运行天数,报警数
    :param company_id:
    :return:
    """
    filters = [
        {"term": {"cid": company_id}},
        {"term": {"mode": "alarm"}},
        # 报警包含scope类型
        # {"terms": {"mode.keyword": ["alarm", "scope"]}},
    ]
    query_body = {
        "query": {"bool": {"filter": filters}},
        "size": 0,
        "aggs": {
            "date_alarms": {
                "date_histogram": {
                    "field": "datetime",
                    "order": {"_key": "desc"},
                    "min_doc_count": 0,
                    "interval": "day",
                    "format": "yyyy-MM-dd",
                    "time_zone": "+08:00",
                }
            }
        },
    }
    
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    
    now_time = datetime.now()
    # 获取到工厂安装时间create_time
    async with MysqlUtil() as conn:
        company_sql = "select create_time from company where cid = %s"
        company = await conn.fetchone(company_sql, (company_id,))
    
    create_time_timestamp = company["create_time"]
    create_time = datetime.fromtimestamp(create_time_timestamp)
    
    today_alarm_count = 0
    alarm_count = 0
    date_buckets = es_result.get("aggregations", {}).get("date_alarms",
                                                         {}).get("buckets", [])
    if not date_buckets:
        log.warn(
            "Can not find data on es(index: %s): %s" % (
                constants.POINT_1MIN_EVENT, query_body)
        )
        # 1. 增加逻辑，新增工厂如果还没有事件产生
        # 系统安全运行天数： 当前时间 - 工厂安装时间 + 1
        safe_run_days = (now_time - create_time).days + 1
        return today_alarm_count, safe_run_days, alarm_count
    
    # 5. 构造返回
    # 如果每天都有报警, 防止安全运行天数-1天, 所以total_days +2
    total_days = (now_time - create_time).days + 2
    has_alarm_days = 0
    for i in date_buckets:
        if i["key_as_string"] == str(now_time)[:10]:
            today_alarm_count += i["doc_count"]
        if i["doc_count"] != 0:
            # 没有报警，看做是安全运行了,统计累计安全运行的天数
            has_alarm_days += 1
        alarm_count += i["doc_count"]
    safe_run_days = total_days - has_alarm_days
    log.info(
        f"today_alarm_count={today_alarm_count} safe_run_days={safe_run_days}")
    return today_alarm_count, safe_run_days, alarm_count


async def other_info_new15(cid):
    today_alarm_count, alarm_count, has_alarm_days = 0, 0, 0
    sql = "SELECT DATE_FORMAT(event_datetime,'%%Y-%%m-%%d') event_date, " \
          "count(id) doc_count FROM `point_1min_event` where cid=%s " \
          "and event_mode='alarm' GROUP BY event_date ORDER BY event_date desc"
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql, args=(cid,))
    now_time = datetime.now()
    # 获取到工厂安装时间create_time
    async with MysqlUtil() as conn:
        company_sql = "select create_time from company where cid = %s"
    company = await conn.fetchone(company_sql, (cid,))
    create_time_timestamp = company["create_time"]
    create_time = datetime.fromtimestamp(create_time_timestamp)
    if not datas:
        # 1. 增加逻辑，新增工厂如果还没有事件产生
        # 系统安全运行天数： 当前时间 - 工厂安装时间 + 1
        safe_run_days = (now_time - create_time).days + 1
        return today_alarm_count, safe_run_days, alarm_count
    # 5. 构造返回
    # 如果每天都有报警, 防止安全运行天数-1天, 所以total_days +2
    total_days = (now_time - create_time).days + 2
    for data in datas:
        if data["event_date"] == str(now_time)[:10]:
            today_alarm_count = data["doc_count"]
        if data["doc_count"]:
            # 不安全运行的天数
            has_alarm_days += 1
        alarm_count += data["doc_count"]
    safe_run_days = total_days - has_alarm_days
    log.info(
        f"today_alarm_count={today_alarm_count} safe_run_days={safe_run_days}")
    return today_alarm_count, safe_run_days, alarm_count


def datetime_to_timestamp(dt):
    ans_time = time.mktime(dt.timetuple())
    return ans_time


async def electric_use_info(company_id):
    """
    用电安全指数
    :param company_id:
    :return:
    """
    now = datetime.now()
    end_timestamp = datetime_to_timestamp(now)
    start_timestamp = datetime_to_timestamp(
        datetime(now.year, now.month, now.day) - timedelta(30))
    
    score_events = [
        i
        for i in EVENT_TYPE_MAP.keys()
        if i
           not in [
               "overTemp",
               "overTempRange1min",
               "overTempRange15min",
               "overTempTrendDaily",
               "overTempTrendQuarterly",
               "over_gap_u",
               "over_rms_u",
               "over_gap_i",
               "over_rms_i",
               "under_rms_u",
               "over_res_cur"
           ]
    ]
    query_body = {
        "query": {
            "bool": {
                "filter": [
                    {"term": {"cid": company_id}},
                    {"range": {"time": {"gte": start_timestamp,
                                        "lte": end_timestamp, }}},
                ],
            }
        },
        "size": 0,
        "aggs": {
            # 这里之所以分score_aggs和alarm_aggs是因为有些事件不计入报警分，但是报警统计又必须要有
            "score_aggs": {
                "filter": {"terms": {"type.keyword": score_events, }},
                "aggs": {"types": {"terms": {"field": "importance"}}},
            },
            "alarm_aggs": {
                "filter": {"term": {"cid": company_id}},
                "aggs": {"types": {"terms": {"field": "importance"}}},
            },
        },
    }
    
    log.info("cal_score_safe_electric query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    
    score_buckets = (
        es_result.get("aggregations", {}).get("score_aggs", {}).get("types",
                                                                    {}).get(
            "buckets", [])
    )
    first_alarm_cnt = 0
    second_alarm_cnt = 0
    third_alarm_cnt = 0
    for bucket in score_buckets:
        if bucket["key"] == Importance.First.value:
            first_alarm_cnt += bucket["doc_count"]
        elif bucket["key"] == Importance.Second.value:
            second_alarm_cnt += bucket["doc_count"]
        elif bucket["key"] == Importance.Third.value:
            third_alarm_cnt += bucket["doc_count"]
    
    company_point_map = await get_points([company_id])
    point_len = len(company_point_map.get(company_id) or {})
    alarm_score = (
        (
                first_alarm_cnt * 2 + second_alarm_cnt * 1 + third_alarm_cnt * 0.5) / point_len
        if point_len
        else 0
    )
    log.info(f"company_point_map:{company_point_map}, point_len:{point_len}, "
             f"alarm_score:{alarm_score}")
    if alarm_score >= 15:
        alarm_score = 15
    
    electric_use_score = get_electric_index(alarm_score)
    
    log.info(
        "point_len={} alarm_score={} electric_use_score={}".format(
            point_len, alarm_score, electric_use_score
        )
    )
    
    alarm_buckets = (
        es_result.get("aggregations", {}).get("alarm_aggs", {}).get("types",
                                                                    {}).get(
            "buckets", [])
    )
    first_alarm_cnt, second_alarm_cnt, third_alarm_cnt = 0, 0, 0
    for bucket in alarm_buckets:
        if bucket["key"] == Importance.First.value:
            first_alarm_cnt += bucket["doc_count"]
        elif bucket["key"] == Importance.Second.value:
            second_alarm_cnt += bucket["doc_count"]
        elif bucket["key"] == Importance.Third.value:
            third_alarm_cnt += bucket["doc_count"]
    return ElectricInfo(
        first_alarm_cnt=first_alarm_cnt,
        second_alarm_cnt=second_alarm_cnt,
        third_alarm_cnt=third_alarm_cnt,
        alarm_score=alarm_score,
        electric_use_score=electric_use_score,
    )


async def electric_use_info_new15(cid):
    """1.5用电安全指数"""
    now = str(datetime.now())
    start = str(datetime.now() - timedelta(30))
    score_events = ['overU', 'underTotalPF', 'underPhasePF', 'overTHDI',
                    'overTHDU', 'underU', 'unbalanceI', 'overPR',
                    'overResidualCurrent', 'unbalanceU', 'overI']
    score_sql = f"select importance, count(importance) doc_count from " \
                f"point_1min_event where cid=%s and event_datetime BETWEEN" \
                f"'{start}' and '{now}' GROUP BY importance"
    alarm_sql = f"select importance, count(importance) doc_count from " \
                f"point_1min_event where cid=%s and event_datetime BETWEEN" \
                f"'{start}' and '{now}' and event_type in %s " \
                f"GROUP BY importance"
    first_score, second_score, third_score = 0, 0, 0
    async with MysqlUtil() as conn:
        score_datas = await conn.fetchall(score_sql, args=(cid,))
        alarm_datas = await conn.fetchall(alarm_sql, args=(cid, score_events))
    for data in score_datas:
        if data["importance"] == Importance.First.value:
            first_score += data["doc_count"]
        elif data["importance"] == Importance.Second.value:
            second_score += data["doc_count"]
        elif data["importance"] == Importance.Third.value:
            third_score += data["doc_count"]
    point_len = await proxy_points([cid])
    alarm_score = (
        (first_score * 2 + second_score * 1 + third_score * 0.5) / point_len
        if point_len else 0
    )
    alarm_score = 15 if alarm_score >= 15 else alarm_score
    electric_use_score = get_electric_index(alarm_score)
    first_alarm, second_alarm, third_alarm = 0, 0, 0
    for data in alarm_datas:
        if data["importance"] == Importance.First.value:
            first_alarm += data["doc_count"]
        elif data["importance"] == Importance.Second.value:
            second_alarm += data["doc_count"]
        elif data["importance"] == Importance.Third.value:
            third_alarm += data["doc_count"]
    return ElectricInfo(
        first_alarm_cnt=first_alarm,
        second_alarm_cnt=second_alarm,
        third_alarm_cnt=third_alarm,
        alarm_score=alarm_score,
        electric_use_score=electric_use_score,
    )


async def normal_rate_of_location(company_id):
    """
    获取温度和漏电流达标率
    :param company_id:
    :return:
    """
    location_map = {}
    async with MysqlUtil() as conn:
        location_sql = "select `location`.`id`, `group`, `item`, `location`.`type`, `threshold` from location left join alarm_setting on location.id=alarm_setting.location_id where location.cid = %s and  `enable`=1 and location.type in %s and `alarm_setting`.type in %s"
        locations = await conn.fetchall(
            sql=location_sql,
            args=(
                company_id,
                ("residual_current", "temperature"),
                ("overTemp", "overResidualCurrent"),
            ),
        )
        for location in locations:
            location_map[location["id"]] = location
    
    # todo批量hmget
    count_info_map = {
        "residual_current": {"total": 0, "normal": 0},
        "temperature": {"total": 0, "normal": 0},
    }
    print(f"len(location_map)={len(location_map)}")
    
    location_ids = list(location_map.keys())
    
    adio_currents = []
    if location_ids:
        adio_currents = await RedisUtils().hmget("adio_current",
                                                 *location_ids)
    
    adio_info_map = {}
    for index, item_byte in enumerate(adio_currents):
        if item_byte:
            item = json.loads(item_byte.decode())
            adio_info_map[location_ids[index]] = item
    
    for location_id, location_info in location_map.items():
        audio_info = adio_info_map.get(location_id)
        count_info_map[location_info["type"]]["total"] += 1
        if audio_info and audio_info.get("value"):
            time_now = int(time.time())
            real_tt = audio_info.get("timestamp", 0)
            if (time_now - real_tt) > constants.REAL_EXP_TIME:
                # 超过4小时的值不统计在normal里
                log.warn(f"adio_current location_id={location_id} has expire!")
                continue
            
            print(
                "threshold={} location_info['type'] = {} audio_info['value']={}".format(
                    location_info["threshold"], location_info["type"],
                    audio_info["value"],
                )
            )
            if (
                    isinstance(location_info["threshold"], float)
                    and audio_info["value"] < location_info["threshold"]
            ):
                count_info_map[location_info["type"]]["normal"] += 1
    if count_info_map["temperature"]["total"] == 0:
        temperature_qr = "100%"
    else:
        temperature_qr = (
                str(
                    round(
                        (
                                count_info_map["temperature"]["normal"]
                                / count_info_map["temperature"]["total"]
                        )
                        * 100,
                    )
                )
                + "%"
        )
    
    if count_info_map["residual_current"]["total"] == 0:
        residual_current_qr = "100%"
    else:
        residual_current_qr = (
                str(
                    round(
                        (
                                count_info_map["residual_current"]["normal"]
                                / count_info_map["residual_current"]["total"]
                        )
                        * 100
                    )
                )
                + "%"
        )
    
    return temperature_qr, residual_current_qr


async def normal_rate_of_location_new15(cid):
    """获取温度和漏电流达标率"""
    location_map = {}
    location_sql = "select l.lid, l.item, l.ad_type type, s.threshold from " \
                   "location l left join soe_config_record s on l.lid=s.lid " \
                   "where l.cid = %s and  s.enable=1 and l.ad_type in %s " \
                   "and s.etype in %s"
    async with MysqlUtil() as conn:
        locations = await conn.fetchall(
            location_sql, args=(cid, ("residual_current", "temperature"),
                                ("overTemp", "overResidualCurrent")))
    for location in locations:
        location_map[location["lid"]] = location
    count_info_map = {
        "residual_current": {"total": 0, "normal": 0},
        "temperature": {"total": 0, "normal": 0},
    }
    location_ids = list(location_map.keys())
    adio_currents = []
    if location_ids:
        adio_currents = await RedisUtils().hmget("adio_current",
                                                 *location_ids)
    adio_info_map = {}
    for index, item_byte in enumerate(adio_currents):
        if item_byte:
            item = json.loads(item_byte.decode())
            adio_info_map[location_ids[index]] = item
    
    for location_id, location_info in location_map.items():
        audio_info = adio_info_map.get(location_id)
        count_info_map[location_info["type"]]["total"] += 1
        if audio_info and audio_info.get("value"):
            time_now = int(time.time())
            real_tt = audio_info.get("timestamp", 0)
            if (time_now - real_tt) > constants.REAL_EXP_TIME:
                # 超过4小时的值不统计在normal里
                log.warn(f"adio_current location_id={location_id} has expire!")
                continue
            if (
                    isinstance(location_info["threshold"], float)
                    and audio_info["value"] < location_info["threshold"]
            ):
                count_info_map[location_info["type"]]["normal"] += 1
    if count_info_map["temperature"]["total"] == 0:
        temperature_qr = "100%"
    else:
        temperature_qr = (
                str(
                    round(
                        (
                                count_info_map["temperature"]["normal"]
                                / count_info_map["temperature"]["total"]
                        )
                        * 100,
                    )
                )
                + "%"
        )
    
    if count_info_map["residual_current"]["total"] == 0:
        residual_current_qr = "100%"
    else:
        residual_current_qr = (
                str(
                    round(
                        (
                                count_info_map["residual_current"]["normal"]
                                / count_info_map["residual_current"]["total"]
                        )
                        * 100
                    )
                )
                + "%"
        )
    
    return temperature_qr, residual_current_qr


async def current_load(company_id):
    """
    实时负荷
    :param company_id:
    :return:
    """
    async with MysqlUtil() as conn:
        point_sql = "select pid from point where cid= %s " \
                    "and add_to_company = 1"
        points = await conn.fetchall(point_sql, args=(company_id,))
        point_ids = [p["pid"] for p in points]
    
    if not point_ids:
        return ""
    
    async with MysqlUtil() as conn:
        meter_sql = (
            "SELECT pid, mid FROM change_meter_record WHERE pid in %s ORDER BY pid, start_time"
        )
        change_meters = await conn.fetchall(meter_sql,
                                            args=(tuple(point_ids),))
        # 正序排序，最后这个map存储的是按照start_time是最近的mid
        change_meter_map = {m["pid"]: m["mid"] for m in change_meters if
                            m["mid"] is not None}
    
    newest_mids = list(change_meter_map.values())
    
    meterdata_currents = []
    if newest_mids:
        meterdata_currents = await RedisUtils().hmget(METERDATA_CURRENT_KEY,
                                                      *newest_mids)
    now_tt = int(time.time())
    
    if meterdata_currents:
        total = 0
        for item in meterdata_currents:
            # 这里是有可能item为None的
            if item:
                item = json.loads(item.decode())
                mdptime_tt = None
                if "mdptime" in item:
                    mdptime = datetime.strptime(item["mdptime"],
                                                "%Y-%m-%d %H:%M:%S")
                    mdptime_tt = time.mktime(mdptime.timetuple())
                item_tt = item.get("timestamp") or mdptime_tt
                if item_tt:
                    # 小于2分钟内的数据相加为实时负荷
                    if now_tt - item_tt <= 2 * 60:
                        total += item["pttl"]
        return total
    return ""


async def current_load_new15(cid, end_dt=None):
    """实时负荷"""
    datas = await get_elec_mtid_sid_by_cid(cid)
    td_mt_tables = tuple(
        (get_td_table_name("electric", data["mtid"]) for data in datas if
         data["mtid"]))
    td_mt_tables = td3_tbl_compate(td_mt_tables)
    if not end_dt:
        end_dt = pendulum.now(tz="Asia/Shanghai")
    start_dt = end_dt.subtract(minutes=2)
    
    sql = f"select last_row(mdptime, pttl) from electric_stb " \
          f"where TBNAME IN {td_mt_tables} and ts>='{str(start_dt)}' and ts " \
          f"<='{str(end_dt)}' group by tbname"
    
    url = f"{SETTING.stb_url}db_electric?tz=Asia/Shanghai"
    is_succ, results = await get_td_engine_data(url, sql)
    if not is_succ:
        return ""
    
    if not results["data"]:  # 兼容：mt表（2.0架构）里面拿不到数据再从sid表（1.0架构）里面拿
        td_s_tables = tuple(
            (f"s{data['sid'].lower()}_e" for data in datas if data["sid"]))
        td_s_tables = td3_tbl_compate(td_s_tables)
        sql = f"select last_row(mdptime, pttl) from electric_stb " \
              f"where TBNAME IN {td_s_tables} group by tbname"
        is_succ, results = await get_td_engine_data(url, sql)
        if not is_succ:
            return ""
    
    head = parse_td_columns(results)
    datas = []
    for res in results["data"]:
        datas.append(dict(zip(head, res)))
    total = 0
    for item in datas:
        if not item:
            continue
        total += item["pttl"]
    return total


async def power_count_info(company_id):
    """
    近30天負荷最大值
    :param company_id:
    :param start_time:
    :param end_time:
    :return:
    """
    now = datetime.now()
    start_time = (now - timedelta(30)).strftime("%Y-%m-%d %H:%M:%S")
    end_time = now.strftime("%Y-%m-%d %H:%M:%S")
    
    max_30d_load, _time = await pttl_max(company_id, start_time, end_time, -1)
    cur_load = await current_load(company_id)
    return cur_load, max_30d_load


async def power_count_info_new15(cid):
    """近30天負荷最大值"""
    now = datetime.now()
    start_time = (now - timedelta(30)).strftime("%Y-%m-%d 00:00:00")
    end_time = now.strftime("%Y-%m-%d %H:%M:%S")
    
    max_30d_load, _time = await pttl_max_new15(cid, start_time, end_time, -1)
    cur_load = await current_load_new15(cid)
    return round_2(cur_load), round_2(max_30d_load)


async def get_max_aiao_of_filed(company_id, start_time, end_time,
                                filed="temperature"):
    """
    近30日最高温度， 近30日最大漏电流
    :param company_id:
    :param start_time:
    :param end_time:
    :param filed:
    :return:
    """
    range = Range(field="time", start=start_time, end=end_time)
    equal = Equal(field="cid", value=company_id)
    # 这里内部调用，不做任何keyerror的假设
    support_field_map = {
        "residual_current": Equal(field="type", value="residual_current"),
        "temperature": Equal(field="type", value="temperature"),
    }
    filter = Filter(
        equals=[equal, support_field_map[filed]], ranges=[range], in_groups=[],
        keywords=[],
    )
    page_request = PageRequest(page_size=1, page_num=1, sort=None,
                               filter=filter)
    # 获取近30天漏电流最大值
    query_body = EsQuery.aggr_index(page_request, stats_items=["value_max"])
    async with EsUtil() as es:
        es_results = await es.search_origin(body=query_body,
                                            index=constants.LOCATION_15MIN_AIAO)
    value_max = es_results.get("aggregations", {}).get("value_max_max", {})
    rc_max_hits = value_max.get("hits", {}).get("hits")
    
    max_info, location_map = {}, {}
    if rc_max_hits:
        max_info = rc_max_hits[0]["_source"]
        location_id = max_info["location_id"]
        async with MysqlUtil() as conn:
            location_sql = "select `group`, `item`, `type` from location where id = %s"
            location_map = await conn.fetchone(sql=location_sql,
                                               args=(location_id,))
    max_ts = max_info.get("value_max_time", 0)
    max_occur_time = (
        str(datetime.strptime(max_ts, "%Y-%m-%dT%H:%M:%S+08:00"))[:-3]
        if max_info
        else None
    )
    
    return MaxResidualCurrent(
        max=round(max_info["value_max"], 2) if max_info else None,
        location_name=f"{location_map['group']}_{'漏电流' if location_map['item'] == 'default' else location_map['item']}"
        if location_map
        else None,
        occur_time=max_occur_time,
    )


async def get_max_aiao_of_filed_new15(cid, start, end, filed="temperature"):
    value_max, location_name, occur_time = None, None, None
    sql = f"SELECT a.value_max,a.value_max_time,p.name,b.item FROM " \
          f"`location_15min_aiao` a LEFT JOIN location b on a.lid=b.lid " \
          f"LEFT JOIN point p on p.mtid=a.mtid where " \
          f"a.create_time BETWEEN '{start}' and '{end}' and a.cid=%s " \
          f" and a.ad_type=%s order by a.value_max desc limit 1"
    async with MysqlUtil() as conn:
        datas = await conn.fetchone(sql, args=(cid, filed))
    if datas:
        value_max = round(datas["value_max"], 2)
        item_name = '漏电流' if datas['item'] == 'default' else datas['item']
        location_name = f"{datas['name']}_{item_name}"
        occur_time = datas.get("value_max_time")
        occur_time = str(occur_time) if occur_time else None
    return MaxResidualCurrent(
        max=value_max,
        location_name=location_name,
        occur_time=occur_time,
    )


async def company_power_use_info(company_id, es_time_start, es_time_end):
    async with EsUtil() as es:
        filters = [
            {"term": {"cid": company_id}},
            {"range": {
                "quarter_time": {"gte": es_time_start, "lt": es_time_end, }}},
        ]
        query_body = {
            "query": {"bool": {"filter": filters}},
            "size": 0,
            "aggs": {"kwh": {"sum": {"field": "kwh"}},
                     "charge": {"sum": {"field": "charge"}}, },
        }
        es_result = await es.search_origin(body=query_body,
                                           index=constants.COMPANY_15MIN_POWER)
        return {
            "charge": es_result["aggregations"]["charge"]["value"],
            "kwh": es_result["aggregations"]["kwh"]["value"],
        }


async def company_power_use_info_new15(company_id, start, end):
    async with MysqlUtil() as conn:
        sql = f"SELECT sum(kwh) kwh, sum(charge) charge FROM " \
              f"`company_15min_power` where create_time BETWEEN " \
              f"'{start}' and '{end}' AND cid=%s"
        datas = await conn.fetchone(sql, args=(company_id,))
    return datas


async def get_company_charge_price(company_id, es_time_start, es_time_end):
    power_use_info = await company_power_use_info_new15(company_id,
                                                        es_time_start,
                                                        es_time_end)
    if power_use_info["kwh"]:
        unit_price = power_use_info["charge"] / power_use_info["kwh"]
    else:
        unit_price = ""
    return unit_price


async def power_charge_price(company_id):
    """
     首页获取昨日平均电价， 上月平均电价
    :param company_id:
    :return:
    """
    # 昨日平均电价
    now = datetime.now()
    yestoday = now - timedelta(1)
    yestoday_start = datetime(yestoday.year, yestoday.month, yestoday.day, 0,
                              0, 0)
    yestoday_end = yestoday_start + timedelta(1)
    
    es_yestoday_start = datetime.strftime(yestoday_start,
                                          "%Y-%m-%dT%H:%M:%S+08:00")
    es_yestoday_end = datetime.strftime(yestoday_end,
                                        "%Y-%m-%dT%H:%M:%S+08:00")
    yestoday_price = await get_company_charge_price(company_id,
                                                    es_yestoday_start,
                                                    es_yestoday_end)
    
    if now.month == 1:
        last_month = 12
        year = now.year - 1
        last_month_start = datetime(year=year, month=last_month, day=1)
    else:
        last_month_start = datetime(year=now.year, month=now.month - 1, day=1)
    last_month_end = datetime(year=now.year, month=now.month, day=1)
    es_last_month_start = datetime.strftime(last_month_start,
                                            "%Y-%m-%dT%H:%M:%S+08:00")
    es_last_month_end = datetime.strftime(last_month_end,
                                          "%Y-%m-%dT%H:%M:%S+08:00")
    last_month_price = await get_company_charge_price(
        company_id, es_last_month_start, es_last_month_end
    )
    
    return yestoday_price, last_month_price


async def power_charge_price_new15(cid):
    """ 首页获取昨日平均电价， 上月平均电价"""
    # 昨日平均电价
    now = datetime.now()
    yestday = now - timedelta(1)
    yestday_start = datetime(yestday.year, yestday.month, yestday.day, 0, 0, 0)
    yestday_end = yestday_start + timedelta(1)
    yestday_datas = await company_power_use_info_new15(cid, str(yestday_start),
                                                       str(yestday_end))
    if yestday_datas["kwh"]:
        yestoday_price = yestday_datas["charge"] / yestday_datas["kwh"]
    else:
        yestoday_price = ""
    if now.month == 1:
        last_month = 12
        year = now.year - 1
        last_month_start = datetime(year=year, month=last_month, day=1)
    else:
        last_month_start = datetime(year=now.year, month=now.month - 1, day=1)
    last_month_end = datetime(year=now.year, month=now.month, day=1)
    last_month_datas = await company_power_use_info_new15(
        cid, str(last_month_start), str(last_month_end)
    )
    if last_month_datas["kwh"]:
        last_month_price = last_month_datas["charge"] / last_month_datas["kwh"]
    else:
        last_month_price = ""
    return round_2(yestoday_price), round_2(last_month_price)


async def power_factor(company_id):
    """
     首页获取实时功率因数， 上月功率因数
    :param company_id:
    :return:
    """
    async with MysqlUtil() as conn:
        point_sql = (
            "select pid, inlid_belongedto from point where cid= %s and add_to_company=%s"
        )
        points = await conn.fetchall(point_sql, args=(company_id, 1))
        point_ids = [i["pid"] for i in points]
    
    now = datetime.now()
    if now.month == 1:
        last_month_dt = datetime(year=now.year - 1, month=12, day=1)
    else:
        last_month_dt = datetime(year=now.year, month=now.month - 1, day=1)
    
    # 首页功率因数取所有进线中最小的
    async with MysqlUtil() as conn:
        sql = "SELECT inlid, `name` FROM inline WHERE cid=%s"
        inlines = await conn.fetchall(sql, args=(company_id,))
        inline_ids = [inline["inlid"] for inline in inlines]
        
        power_factor_results = []
        sql = "SELECT inlid, save_charge pf_cost, `kpi_x`, `save_charge` " \
              "FROM algo_power_factor_result WHERE inlid in %s and month=%s"
        if inline_ids:
            power_factor_results = await conn.fetchall(sql, args=(
                inline_ids, last_month_dt))
        pf_kpi_x_list = [
            i["kpi_x"] for i in power_factor_results if
            type(i["kpi_x"]) in [int, float]
        ]
        last_month_cos = min(pf_kpi_x_list) if len(pf_kpi_x_list) else ""
    
    async with EsUtil() as es:
        dt = pendulum.now(tz="Asia/Shanghai")
        tstamp = dt.int_timestamp // (15 * 60) * (15 * 60)
        dt = pendulum.from_timestamp(tstamp, tz="Asia/Shanghai")
        filters = [
            {"terms": {"pid": point_ids}},
            {"terms": {
                "quarter_time": [str(dt), str(dt - timedelta(minutes=15))]}},
        ]
        query_body = {
            "_source": ["pid", "quarter_time", "pttl_mean", "qttl_mean"],
            "query": {"bool": {"filter": filters}},
            "size": 10000,
            "sort": [{"pid": {"order": "asc"}},
                     {"quarter_time": {"order": "desc"}}],
        }
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_15MIN_INDEX)
        point_infos = es_result["hits"]["hits"]
        point_map = {}
        for i in point_infos:
            item = i["_source"]
            point_map.setdefault(item["pid"], []).append(
                {
                    "quarter_time": item["quarter_time"],
                    "pttl_mean": item["pttl_mean"],
                    "qttl_mean": item["qttl_mean"],
                }
            )
        
        total_pttl, total_qttl = 0, 0
        for point_id, records in point_map.items():
            total_pttl += records[0]["pttl_mean"]
            total_qttl += records[0]["qttl_mean"]
        
        # 计算实时功率的公式
        cos_ttl = ""
        l = sqrt(total_pttl * total_pttl + total_qttl * total_qttl)
        if l:
            cos_ttl = round(total_pttl / l, 2)
        
        if type(last_month_cos) in [int, float]:
            last_month_cos = round(last_month_cos, 2)
    return cos_ttl, last_month_cos


async def power_factor_new15(cid):
    """首页获取实时功率因数， 上月功率因数"""
    point_sql = "select pid,inlid from point where cid=%s and add_to_company=1"
    async with MysqlUtil() as conn:
        points = await conn.fetchall(point_sql, args=(cid,))
    point_ids = [i["pid"] for i in points]
    now = datetime.now()
    if now.month == 1:
        last_month_dt = datetime(year=now.year - 1, month=12, day=1)
    else:
        last_month_dt = datetime(year=now.year, month=now.month - 1, day=1)
    # 首页功率因数取所有进线中最小的
    inline_sql = "SELECT inlid, `name` FROM inline WHERE cid=%s"
    async with MysqlUtil() as conn:
        inlines = await conn.fetchall(inline_sql, args=(cid,))
        inline_ids = [inline["inlid"] for inline in inlines]
        power_factor_results = []
        sql = "SELECT inlid, save_charge pf_cost, `kpi_x`, `save_charge` " \
              "FROM algo_power_factor_result WHERE inlid in %s and month=%s"
        if inline_ids:
            power_factor_results = await conn.fetchall(sql, args=(
                inline_ids, last_month_dt))
        pf_kpi_x_list = [
            i["kpi_x"] for i in power_factor_results if
            type(i["kpi_x"]) in [int, float]
        ]
        last_month_cos = min(pf_kpi_x_list) if len(pf_kpi_x_list) else ""
        
        dt = pendulum.now(tz="Asia/Shanghai")
        tstamp = dt.int_timestamp // (15 * 60) * (15 * 60)
        dt = pendulum.from_timestamp(tstamp, tz="Asia/Shanghai")
        end_dt = (dt - timedelta(minutes=15)).strftime("%Y-%m-%d %H:%M:%S")
        str_dt = dt.strftime("%Y-%m-%d %H:%M:%S")
        electric_sql = f"SELECT pid,create_time,pttl_mean,qttl_mean FROM " \
                       f"`point_15min_electric` where create_time in " \
                       f"('{str_dt}', '{end_dt}') and pid in %s"
        results = await conn.fetchall(electric_sql, args=(point_ids,))
        point_map = {}
        for res in results:
            point_map.setdefault(res["pid"], []).append(
                {
                    "quarter_time": str(res["create_time"]),
                    "pttl_mean": res["pttl_mean"],
                    "qttl_mean": res["qttl_mean"],
                }
            )
        total_pttl, total_qttl = 0, 0
        for point_id, records in point_map.items():
            total_pttl += records[0]["pttl_mean"]
            total_qttl += records[0]["qttl_mean"]
        # 计算实时功率的公式
        cos_ttl = ""
        l = sqrt(total_pttl * total_pttl + total_qttl * total_qttl)
        if l:
            cos_ttl = round(total_pttl / l, 2)
        
        if type(last_month_cos) in [int, float]:
            last_month_cos = round(last_month_cos, 2)
    return cos_ttl, last_month_cos


async def optimization_count_info(company_id: int):
    """
    首页用电经济指数和用电优化模块统计数据
    :param company_id:
    :return:
    """
    
    async with MysqlUtil() as conn:
        sql = "SELECT inlid, `name` FROM inline WHERE cid=%s"
        inlines = await conn.fetchall(sql, args=(company_id,))
        inline_ids = [inline["inlid"] for inline in inlines]
    
    # 获取公司上月用电
    # now = datetime.now()
    # es_start_time = (
    #     pendulum.datetime(now.year, now.month, 1)
    #         .subtract(months=1)
    #         .strftime("%Y-%m-%dT%H:%M:%S+08:00")
    # )
    # es_end_time = pendulum.datetime(now.year, now.month, 1).strftime(
    #     "%Y-%m-%dT%H:%M:%S+08:00")
    # power_use_info = await company_power_use_info(company_id, es_start_time,
    #                                               es_end_time)
    
    now = datetime.now()
    start_time = (
        pendulum.datetime(now.year, now.month, 1)
        .subtract(months=1)
        .strftime("%Y-%m-%d %H:%M:%S")
    )
    end_time = pendulum.datetime(now.year, now.month, 1).strftime(
        "%Y-%m-%d %H:%M:%S")
    power_use_info = await company_power_use_info_new15(company_id, start_time,
                                                        end_time)
    month_charge = power_use_info.get("charge")
    month_kwh = power_use_info.get("kwh")
    count_info_map = {
        "avg_price": month_charge / month_kwh if month_kwh else ""
    }
    if not inline_ids:
        count_info_map.update(
            {
                "power_factor": {"save_charge": "", "kpi_x": "", "desc": "", },
                "pcvf": {"save_charge": "", "kpi_x": "", "desc": "", },
                "power_save": {"save_charge": "", "kpi_x": "", "desc": "", },
                "md_space": {"save_charge": "", "kpi_x": "", "desc": "", },
                "save_percent": 0,
                "md_space_p": "",
                "mean_load_factor": "",
            }
        )
        return count_info_map
    
    now = datetime.now()
    if now.month == 1:
        last_month_dt = datetime(year=now.year - 1, month=12, day=1)
    else:
        last_month_dt = datetime(year=now.year, month=now.month - 1, day=1)
    
    # 功率因数
    async with MysqlUtil() as conn:
        sql = "SELECT inlid, `cos`, save_charge pf_cost, kpi_x, save_charge " \
              "FROM algo_power_factor_result WHERE inlid in %s " \
              "and month=%s"
        power_factor_results = await conn.fetchall(sql, args=(
            inline_ids, last_month_dt))
        total_pf_save = round(
            sum([i["pf_cost"] for i in power_factor_results if
                 i["pf_cost"] and i["pf_cost"] >= 0]),
            2,
        )
        total_pf_save = 0 if total_pf_save <= 0 else total_pf_save
        
        pf_kpi_x_list = [
            i["kpi_x"] for i in power_factor_results if
            type(i["kpi_x"]) in [int, float]
        ]
        pf_kpi_x = min(pf_kpi_x_list) if len(pf_kpi_x_list) else ""
        if pf_kpi_x == "":
            pf_desc = ""
        elif pf_kpi_x >= 0.9:
            pf_desc = "无空间"
        elif 0.85 < pf_kpi_x < 0.9:
            pf_desc = "空间较小"
        elif 0.8 < pf_kpi_x <= 0.85:
            pf_desc = "空间适中"
        else:
            pf_desc = "空间较大"
        
        count_info_map["power_factor"] = {
            "save_charge": total_pf_save if pf_kpi_x != "" else "",
            "kpi_x": pf_kpi_x,
            "desc": pf_desc,
        }
    
    # 移峰填谷指数
    async with MysqlUtil() as conn:
        sql = "select `score`, `cost_save` from `algo_plsi_result` " \
              "where `inlid` in %s and `month` = %s"
        last_month_str = datetime.strftime(last_month_dt, "%Y-%m")
        pcvfs = await conn.fetchall(sql, args=(inline_ids, last_month_str))
        
        pcvf_kpi_x_list = [i["score"] for i in pcvfs if
                           type(i["score"]) in [int, float]]
        pcvf_kpi_x = min(pcvf_kpi_x_list) if len(pcvf_kpi_x_list) else ""
        total_pcvf_save = round(
            sum([i["cost_save"] for i in pcvfs if
                 i["cost_save"] and i["cost_save"] >= 0]), 2
        )
        
        if pcvf_kpi_x == "":
            pcvf_desc = ""
        elif pcvf_kpi_x >= 90:
            pcvf_desc = "无空间"
        elif 80 < pcvf_kpi_x < 90:
            pcvf_desc = "空间较小"
        elif 70 < pcvf_kpi_x <= 80:
            pcvf_desc = "空间适中"
        else:
            pcvf_desc = "空间较大"
        
        total_pcvf_save = 0 if total_pcvf_save <= 0 else total_pcvf_save
        count_info_map["pcvf"] = {
            "save_charge": total_pcvf_save if pcvf_kpi_x != "" else "",
            "kpi_x": pcvf_kpi_x,
            "desc": pcvf_desc,
        }
    
    # 经济运行
    async with MysqlUtil() as conn:
        sql = "select `kpi_x`, `save_charge`, `mean_load_factor` " \
              "from `algo_economic_operation_result` where " \
              "`inlid` in %s and `month` = %s"
        last_month_str = datetime.strftime(last_month_dt, "%Y-%m")
        economic_operations = await conn.fetchall(sql, args=(
            inline_ids, last_month_str))
        economic_kpi_x_list = [
            i["kpi_x"] for i in economic_operations if
            type(i["kpi_x"]) in [int, float]
        ]
        economic_kpi_x = max(
            economic_kpi_x_list) if economic_kpi_x_list else ""
        total_economic_save = round(
            sum(
                [
                    i["save_charge"]
                    for i in economic_operations
                    if i["save_charge"] and i["save_charge"] >= 0
                ]
            ),
            2,
        )
        total_economic_save = 0 if total_economic_save <= 0 else \
            total_economic_save
        if economic_kpi_x == "":
            economic_desc = ""
        elif economic_kpi_x <= 0.6:
            economic_desc = "无空间"
        elif 0.6 < economic_kpi_x <= 0.7:
            economic_desc = "空间较小"
        elif 0.7 < economic_kpi_x <= 0.8:
            economic_desc = "空间适中"
        else:
            economic_desc = "空间较大"
        
        count_info_map["power_save"] = {
            "save_charge": total_economic_save if economic_kpi_x != "" else "",
            "kpi_x": economic_kpi_x,
            "desc": economic_desc,
        }
    
    # 最大需量
    async with MysqlUtil() as conn:
        sql = (
            "select a.inline_md_charge, a.kpi_x, a.save_charge "
            "from `algo_md_space_analysis_result` a "
            "inner join`algo_md_space_analysis_unit` b "
            " on a.space_analysis_id=b.id "
            "where b.inlid in %s and a.month = %s and b.valid=1;"
        )
        last_month_str = datetime.strftime(last_month_dt, "%Y-%m")
        md_spaces = await conn.fetchall(sql, args=(inline_ids, last_month_str))
        
        md_space_kpi_x_list = [i["kpi_x"] for i in md_spaces if
                               type(i["kpi_x"]) in [int, float]]
        md_space_kpi_x = max(md_space_kpi_x_list) if len(
            md_space_kpi_x_list) else ""
        total_md_space_save = round(
            sum(
                [i["save_charge"] for i in md_spaces if
                 i["save_charge"] and i["save_charge"] >= 0]
            ),
            2,
        )
        total_md_space_save = 0 if total_md_space_save <= 0 else \
            total_md_space_save
        if md_space_kpi_x == "":
            md_space_desc = ""
        elif md_space_kpi_x <= 0:
            md_space_desc = "无空间"
        elif 0 < md_space_kpi_x <= 0.1:
            md_space_desc = "空间较小"
        elif 0.1 < md_space_kpi_x <= 0.2:
            md_space_desc = "空间适中"
        else:
            md_space_desc = "空间较大"
        count_info_map["md_space"] = {
            "save_charge": total_md_space_save if md_space_kpi_x != "" else "",
            "kpi_x": md_space_kpi_x,
            "desc": md_space_desc,
        }
    
    total_save_cost = 0
    for _, item in count_info_map.items():
        total_save_cost += (
            item["save_charge"] if isinstance(item, dict) and item[
                "save_charge"] else 0
        )
    save_percent = total_save_cost / month_charge if month_charge else ""
    count_info_map["save_percent"] = save_percent
    
    # 计算最大需量
    async with MysqlUtil() as conn:
        sql = "select `price_md`,`price_tc` from `price_policy` where `cid`=%s"
        price_policy = await conn.fetchone(sql, args=(company_id,))
    
    total_md_space_charge = sum(
        [i["inline_md_charge"] for i in md_spaces if i["inline_md_charge"]])
    total_md_space_p = (
        total_md_space_charge / price_policy["price_md"]
        if price_policy and price_policy["price_md"]
        else ""
    )
    count_info_map["md_space_p"] = total_md_space_p
    
    # 经济运行最低负载率
    mean_load_factors = [
        i["mean_load_factor"] for i in economic_operations if
        i["mean_load_factor"]
    ]
    count_info_map["mean_load_factor"] = ""
    if mean_load_factors:
        count_info_map["mean_load_factor"] = min(mean_load_factors)
    return count_info_map


async def electric_use_info_sdu(cid):
    """用电安全指数, 识电u"""
    start, end = last30_day_range()
    start_es = convert_es_str(start)
    end_es = convert_es_str(end)
    query_body = {
        "query": {
            "bool": {
                "filter": [
                    {"term": {"cid": cid}},
                    {"range": {
                        "datetime": {"gte": start_es, "lte": end_es, }}},
                    {"terms": {"type.keyword": SDU_ALARM_LIST}}
                ],
            }
        },
        "size": 0,
        "aggs": {
            "alarm_aggs": {
                "terms": {"field": "importance"}
            }
        }
    }
    
    log.info("cal_score_safe_electric query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    
    score_buckets = (
        es_result.get("aggregations", {}).get("alarm_aggs", {}).get("buckets",
                                                                    [])
    )
    first_alarm_cnt = 0
    second_alarm_cnt = 0
    third_alarm_cnt = 0
    for bucket in score_buckets:
        if bucket["key"] == Importance.First.value:
            first_alarm_cnt += bucket["doc_count"]
        elif bucket["key"] == Importance.Second.value:
            second_alarm_cnt += bucket["doc_count"]
        elif bucket["key"] == Importance.Third.value:
            third_alarm_cnt += bucket["doc_count"]
    
    company_point_map = await get_points([cid])
    point_len = len(company_point_map.get(cid) or {})
    alarm_score = (
        (
                first_alarm_cnt * 2 + second_alarm_cnt * 1 + third_alarm_cnt * 0.5) / point_len
        if point_len
        else 0
    )
    if alarm_score >= 15:
        alarm_score = 15
    
    electric_use_score = get_electric_index(alarm_score)
    
    log.info(
        "point_len={} alarm_score={} electric_use_score={}".format(
            point_len, alarm_score, electric_use_score
        )
    )
    return ElectricInfo(
        first_alarm_cnt=first_alarm_cnt,
        second_alarm_cnt=second_alarm_cnt,
        third_alarm_cnt=third_alarm_cnt,
        alarm_score=alarm_score,
        electric_use_score=electric_use_score,
    )


async def electric_use_info_sdu_new15(cid):
    start, end = last30_day_range()
    first_alarm_cnt = 0
    second_alarm_cnt = 0
    third_alarm_cnt = 0
    sql = f"select importance,count(1) doc_count from point_1min_event " \
          f"where cid={cid} and event_datetime BETWEEN '{start}' and '{end}' " \
          f"and event_type in {tuple(SDU_ALARM_LIST)} GROUP BY importance"
    log.info(f"sql:{sql}")
    async with MysqlUtil() as conn:
        datas = await conn.fetchall(sql)
    for data in datas:
        if data["importance"] == Importance.First.value:
            first_alarm_cnt += data["doc_count"]
        elif data["importance"] == Importance.Second.value:
            second_alarm_cnt += data["doc_count"]
        elif data["importance"] == Importance.Third.value:
            third_alarm_cnt += data["doc_count"]
    point_len = await get_points_num(cid)
    alarm_score = (
        (
                first_alarm_cnt * 2 + second_alarm_cnt * 1 + third_alarm_cnt * 0.5) / point_len
        if point_len
        else 0
    )
    alarm_score = 15 if alarm_score >= 15 else alarm_score
    electric_use_score = get_electric_index(alarm_score)
    return ElectricInfo(
        first_alarm_cnt=first_alarm_cnt,
        second_alarm_cnt=second_alarm_cnt,
        third_alarm_cnt=third_alarm_cnt,
        alarm_score=alarm_score,
        electric_use_score=electric_use_score,
    )


async def electric_use_info_points_sdu(start, end, points):
    """用电安全指数, 识电u, 根据points来计算"""
    start_es = convert_es_str(start)
    end_es = convert_es_str(end)
    query_body = {
        "query": {
            "bool": {
                "filter": [
                    {"terms": {"point_id": points}},
                    {"range": {
                        "datetime": {"gte": start_es, "lte": end_es, }}},
                    {"terms": {"type.keyword": SDU_ALARM_LIST}}
                ],
            }
        },
        "size": 0,
        "aggs": {
            "alarm_aggs": {
                "terms": {"field": "importance"}
            }
        }
    }
    
    log.info("electric_use_info_points query_body={}".format(query_body))
    async with EsUtil() as es:
        es_result = await es.search_origin(body=query_body,
                                           index=constants.POINT_1MIN_EVENT)
    
    score_buckets = (
        es_result.get("aggregations", {}).get("alarm_aggs", {}).get("buckets",
                                                                    [])
    )
    first_alarm_cnt = 0
    second_alarm_cnt = 0
    third_alarm_cnt = 0
    for bucket in score_buckets:
        if bucket["key"] == Importance.First.value:
            first_alarm_cnt += bucket["doc_count"]
        elif bucket["key"] == Importance.Second.value:
            second_alarm_cnt += bucket["doc_count"]
        elif bucket["key"] == Importance.Third.value:
            third_alarm_cnt += bucket["doc_count"]
    
    alarm_score = (first_alarm_cnt * 2 + second_alarm_cnt * 1 +
                   third_alarm_cnt * 0.5) / len(points)
    
    if alarm_score >= 15:
        alarm_score = 15
    
    electric_use_score = get_electric_index(alarm_score)
    
    log.info(
        "point_len={} alarm_score={} electric_use_score={}".format(
            len(points), alarm_score, electric_use_score
        )
    )
    return ElectricInfo(
        first_alarm_cnt=first_alarm_cnt,
        second_alarm_cnt=second_alarm_cnt,
        third_alarm_cnt=third_alarm_cnt,
        alarm_score=alarm_score,
        electric_use_score=electric_use_score,
    )


async def electric_use_info_points_sdu_new15(start, end, points):
    """用电安全指数, 识电u, 根据points来计算"""
    sql = f"select importance,count(*) as doc_count from  point_1min_event " \
          f"where  pid in %s and  event_datetime BETWEEN %s and %s " \
          f"and event_type in %s group by importance"
    async with MysqlUtil() as conn:
        results = await conn.fetchall(sql, args=(points, start, end,
                                                 SDU_ALARM_LIST))
    
    first_alarm_cnt = 0
    second_alarm_cnt = 0
    third_alarm_cnt = 0
    for result in results:
        if result["importance"] == Importance.First.value:
            first_alarm_cnt += result["doc_count"]
        elif result["importance"] == Importance.Second.value:
            second_alarm_cnt += result["doc_count"]
        elif result["importance"] == Importance.Third.value:
            third_alarm_cnt += result["doc_count"]
    
    alarm_score = (first_alarm_cnt * 2 + second_alarm_cnt * 1 +
                   third_alarm_cnt * 0.5) / len(points)
    
    if alarm_score >= 15:
        alarm_score = 15
    
    electric_use_score = get_electric_index(alarm_score)
    
    log.info(
        "point_len={} alarm_score={} electric_use_score={}".format(
            len(points), alarm_score, electric_use_score
        )
    )
    return ElectricInfo(
        first_alarm_cnt=first_alarm_cnt,
        second_alarm_cnt=second_alarm_cnt,
        third_alarm_cnt=third_alarm_cnt,
        alarm_score=alarm_score,
        electric_use_score=electric_use_score,
    )


async def optimization_count_info_new(company_id: int):
    """
    首页用电经济指数和用电优化模块统计数据
    :param company_id:
    :return:
    """
    
    inlines = await get_inline_by_cid(company_id)
    inline_ids = [inline["inlid"] for inline in inlines]
    
    # 获取公司上月用电
    now = datetime.now()
    es_start_time = (
        pendulum.datetime(now.year, now.month, 1)
        .subtract(months=1)
        .strftime("%Y-%m-%dT%H:%M:%S+08:00")
    )
    es_end_time = pendulum.datetime(now.year, now.month, 1).strftime(
        "%Y-%m-%dT%H:%M:%S+08:00")
    power_use_info = await company_power_use_info_new15(company_id,
                                                        es_start_time,
                                                        es_end_time)
    month_charge = power_use_info["charge"]
    count_info_map = {
        "avg_price": power_use_info["charge"] / power_use_info["kwh"]
        if power_use_info["kwh"]
        else ""
    }
    if not inline_ids:
        count_info_map.update(
            {
                "power_factor": {"save_charge": "", "kpi_x": "", "desc": "", },
                "pcvf": {"save_charge": "", "kpi_x": "", "desc": "", },
                "power_save": {"save_charge": "", "kpi_x": "", "desc": "", },
                "md_space": {"save_charge": "", "kpi_x": "", "desc": "", },
                "save_percent": 0,
                "md_space_p": "",
                "mean_load_factor": "",
            }
        )
        return count_info_map
    if now.month == 1:
        last_month_dt = datetime(year=now.year - 1, month=12, day=1)
    else:
        last_month_dt = datetime(year=now.year, month=now.month - 1, day=1)
    last_month_str = datetime.strftime(last_month_dt, "%Y-%m")
    # 功率因数
    
    power_factor_results = await get_power_factor_kpi(inline_ids,
                                                      last_month_dt)
    total_pf_save = round(
        sum([i["pf_cost"] for i in power_factor_results if
             i["pf_cost"] and i["pf_cost"] >= 0]),
        2,
    )
    total_pf_save = 0 if total_pf_save <= 0 else total_pf_save
    pf_kpi_x_list = [
        (i["name"], i["kpi_x"]) for i in power_factor_results if
        type(i["kpi_x"]) in [int, float]
    ]
    
    if len(pf_kpi_x_list):
        pf_kpi_x_num = [pf_kpi[1] for pf_kpi in pf_kpi_x_list]
        pf_kpi_x = min(pf_kpi_x_num)
        if pf_kpi_x >= 0.9:
            pf_desc = "上月功率因数比较合理，请继续保持"
        else:
            pf_kpi_x_name = []
            for index, kpi_num in enumerate(pf_kpi_x_num):
                if kpi_num < 0.9:
                    pf_kpi_x_name.append(pf_kpi_x_list[index][0])
            pf_desc = f"{'、'.join(pf_kpi_x_name)}可以进行无功补偿"
    else:
        pf_kpi_x = ""
        pf_desc = ""
    if pf_kpi_x == "":
        pf_space = ""
    elif pf_kpi_x >= 0.9:
        pf_space = "无空间"
    elif 0.85 < pf_kpi_x < 0.9:
        pf_space = "空间较小"
    elif 0.8 < pf_kpi_x <= 0.85:
        pf_space = "空间适中"
    else:
        pf_space = "空间较大"
    count_info_map["power_factor"] = {
        "save_charge": total_pf_save if pf_kpi_x != "" else "",
        "kpi_x": pf_kpi_x,
        "desc": pf_desc,
        "space": pf_space
    }
    # 移峰填谷指数
    pcvfs = await get_pcvf_kpi(inline_ids, last_month_str)
    pcvf_kpi_x_list = [(i["name"], i["score"]) for i in pcvfs if
                       type(i["score"]) in [int, float]]
    
    if len(pcvf_kpi_x_list):
        pcvf_kpi_x_num = [pcvf_kpi[1] for pcvf_kpi in pcvf_kpi_x_list]
        pcvf_kpi_x = min(pcvf_kpi_x_num)
        pcvf_kpi_x_name = []
        
        if pcvf_kpi_x < 70:
            for index, kpi_num in enumerate(pcvf_kpi_x_num):
                if kpi_num < 70:
                    pcvf_kpi_x_name.append(pcvf_kpi_x_list[index][0])
            pcvf_desc = f"{'、'.join(pcvf_kpi_x_name)}可以进行无功补偿"
        elif pcvf_kpi_x < 90:
            for index, kpi_num in enumerate(pcvf_kpi_x_num):
                if kpi_num < 90:
                    pcvf_kpi_x_name.append(pcvf_kpi_x_list[index][0])
            pcvf_desc = f"请合理调整{'、'.join(pcvf_kpi_x_name)}用电时间或" \
                        f"引入新能源，转移高峰电量至低谷"
        else:
            pcvf_desc = "平均电价处于较低水平，请继续保持"
    
    else:
        pcvf_kpi_x = ""
        pcvf_desc = ""
    if pcvf_kpi_x == "":
        pcvf_space = ""
    elif pcvf_kpi_x >= 90:
        pcvf_space = "无空间"
    elif 80 < pcvf_kpi_x < 90:
        pcvf_space = "空间较小"
    elif 70 < pcvf_kpi_x <= 80:
        pcvf_space = "空间适中"
    else:
        pcvf_space = "空间较大"
    total_pcvf_save = round(
        sum([i["cost_save"] for i in pcvfs if
             i["cost_save"] and i["cost_save"] >= 0]), 2
    )
    
    total_pcvf_save = 0 if total_pcvf_save <= 0 else total_pcvf_save
    count_info_map["pcvf"] = {
        "save_charge": total_pcvf_save if pcvf_kpi_x != "" else "",
        "kpi_x": pcvf_kpi_x,
        "desc": pcvf_desc,
        "space": pcvf_space
    }
    
    # 经济运行
    economic_operations = await get_economic_kpi(inline_ids, last_month_str)
    economic_kpi_x_list = [
        (i["name"], i["kpi_x"]) for i in economic_operations if
        type(i["kpi_x"]) in [int, float]
    ]
    total_economic_save = round(
        sum(
            [
                i["save_charge"]
                for i in economic_operations
                if i["save_charge"] and i["save_charge"] >= 0
            ]
        ),
        2,
    )
    if len(economic_kpi_x_list):
        econ_kpi_x_num = [econ_kpi[1] for econ_kpi in economic_kpi_x_list]
        econ_kpi_x = max(econ_kpi_x_num)
        econ_kpi_x_min = min(econ_kpi_x_num)
        econ_kpi_x_name = []
        if econ_kpi_x >= 0.9:
            for index, kpi_max in enumerate(econ_kpi_x_num):
                if kpi_max >= 0.9:
                    econ_kpi_x_name.append(economic_kpi_x_list[index][0])
            economic_desc = f"{'、'.join(econ_kpi_x_name)}负载率超过" \
                            f"安全阈值，应考虑扩容"
        elif econ_kpi_x >= 0.7:
            for index, kpi_max in enumerate(econ_kpi_x_num):
                if kpi_max >= 0.7:
                    econ_kpi_x_name.append(economic_kpi_x_list[index][0])
            economic_desc = f"{'、'.join(econ_kpi_x_name)}负载率较高，" \
                            f"可考虑调整负荷或扩容"
        elif econ_kpi_x_min < 0.6 and econ_kpi_x < 0.9:
            for index, kpi_max in enumerate(econ_kpi_x_num):
                if kpi_max < 0.6:
                    econ_kpi_x_name.append(economic_kpi_x_list[index][0])
            economic_desc = f"{'、'.join(econ_kpi_x_name)}负载率较低，" \
                            f"请考虑容改需或更换小容量变压器"
        else:
            economic_desc = "进线负载率比较经济，请继续保持"
    else:
        econ_kpi_x = ""
        economic_desc = ""
    if econ_kpi_x == "":
        econ_space = ""
    elif econ_kpi_x <= 0.6:
        econ_space = "无空间"
    elif 0.6 < econ_kpi_x <= 0.7:
        econ_space = "空间较小"
    elif 0.7 < econ_kpi_x <= 0.8:
        econ_space = "空间适中"
    else:
        econ_space = "空间较大"
    count_info_map["power_save"] = {
        "save_charge": total_economic_save if econ_kpi_x != "" else "",
        "kpi_x": econ_kpi_x,
        "desc": economic_desc,
        "space": econ_space
    }
    
    #  容量、需量价格
    price_policy = await price_policy_by_cid(company_id)
    price_md = price_policy["price_md"] if price_policy["price_md"] else 0
    price_tc = price_policy["price_tc"] if price_policy["price_tc"] else 0
    # 最大需量
    md_spaces = await get_md_space(inline_ids, last_month_dt)
    
    md_space_kpi_x_list = [i["kpi_x"] for i in md_spaces if
                           type(i["kpi_x"]) in [int, float]]
    md_space_kpi_x = max(md_space_kpi_x_list) if len(
        md_space_kpi_x_list) else ""
    total_md_space_save = round(
        sum(
            [i["save_charge"] for i in md_spaces if
             i["save_charge"] and i["save_charge"] >= 0]
        ),
        2,
    )
    total_md_space_save = 0 if total_md_space_save <= 0 else total_md_space_save
    if md_space_kpi_x == "":
        md_space_space = ""
    elif md_space_kpi_x <= 0:
        md_space_space = "无空间"
    elif 0 < md_space_kpi_x <= 0.1:
        md_space_space = "空间较小"
    elif 0.1 < md_space_kpi_x <= 0.2:
        md_space_space = "空间适中"
    else:
        md_space_space = "空间较大"
    md_space_tc_runtimes = await get_tc_runtime(inline_ids)
    md_space_name = []
    for index, item in enumerate(md_spaces):
        if type(item["inline_md_predict"]) in [int, float] and \
                md_space_tc_runtimes[index]["tc_runtime"] * price_tc >= \
                price_md * item["inline_md_predict"]:
            md_space_name.append(md_space_tc_runtimes[index]["name"])
    
    if len(md_space_name):
        md_space_desc = f"若次月负荷无较大变动，建议{'、'.join(md_space_name)}" \
                        f"选择按最大需量计费"
    else:
        md_space_desc = "不存在容改需空间"
    
    count_info_map["md_space"] = {
        "save_charge": total_md_space_save if md_space_kpi_x != "" else "",
        "kpi_x": md_space_kpi_x,
        "desc": md_space_desc,
        "space": md_space_space
    }
    # 节费率
    total_save_cost = 0
    for _, item in count_info_map.items():
        total_save_cost += (
            item["save_charge"] if isinstance(item, dict) and item[
                "save_charge"] else 0
        )
    save_percent = total_save_cost / month_charge if month_charge else ""
    count_info_map["save_percent"] = save_percent
    # 计算最大需量
    total_md_space_charge = sum(
        [i["inline_md_charge"] for i in md_spaces if i["inline_md_charge"]])
    total_md_space_p = (
        total_md_space_charge / price_policy["price_md"]
        if price_policy and price_policy["price_md"]
        else ""
    )
    count_info_map["md_space_p"] = total_md_space_p
    # 经济运行最低负载率
    mean_load_factors = [
        i["mean_load_factor"] for i in economic_operations if
        i["mean_load_factor"]
    ]
    count_info_map["mean_load_factor"] = ""
    if mean_load_factors:
        count_info_map["mean_load_factor"] = min(mean_load_factors)
    return count_info_map


async def cid_alarm_importance_count(cid, start, end):
    """计算工厂报警数, 按报警等级"""
    monitor_point_list = await monitor_point_join(cid)
    point_list = [i["pid"] for i in monitor_point_list]
    es_res = await sdu_alarm_importance_dao_new15(start, end, point_list)
    es_res_key = {i["key"]: i for i in es_res}
    
    res_list = []
    for info in monitor_point_list:
        name = info.get("name")
        point_id = info.get("pid")
        tmp_dic = {"name": name, "alarm_count": 0,
                   "first": 0, "second": 0, "third": 0}
        if point_id in es_res_key:
            inner_bucket = es_res_key[point_id]["importance"]["buckets"]
            for b in inner_bucket:
                if b["key"] == Importance.First.value:
                    tmp_dic["first"] += b["doc_count"]
                elif b["key"] == Importance.Second.value:
                    tmp_dic["second"] += b["doc_count"]
                elif b["key"] == Importance.Third.value:
                    tmp_dic["third"] += b["doc_count"]
        
        tmp_dic["alarm_count"] = tmp_dic["first"] + tmp_dic["second"] + \
                                 tmp_dic[
                                     "third"]
        res_list.append(tmp_dic)
    # 按照报警数, 排top5
    res_list_sort = sorted(res_list, key=lambda i: i["alarm_count"],
                           reverse=True)[:5]
    return res_list_sort


async def alarm_importance_count_total(cid, start, end):
    """计算工厂报警数, 按报警等级"""
    es_res = await alarm_aggs_importance(cid, start, end)
    first_cnt, second_cnt, third_cnt = 0, 0, 0
    for buckets in es_res:
        if buckets["importance"] == Importance.First.value:
            first_cnt += buckets["alarm_count"]
        elif buckets["importance"] == Importance.Second.value:
            second_cnt += buckets["alarm_count"]
        elif buckets["importance"] == Importance.Third.value:
            third_cnt += buckets["alarm_count"]
    return {
        "first_cnt": first_cnt,
        "second_cnt": second_cnt,
        "third_cnt": third_cnt,
        "total_cnt": first_cnt + second_cnt + third_cnt
    }


def safety_ratio_res(score, client_name):
    """根据安全指数, 求安全风险系数"""
    # wechat和web共用
    if 90 <= score <= 100:
        return "风险较低" if client_name == "wechat" else "当前安全状况良好，请继续保持！"
    elif 80 <= score < 90:
        return "低风险" if client_name == "wechat" else "当前安全状况较好，请关注报警信息！"
    elif 60 <= score < 80:
        return "中风险" if client_name == "wechat" else "当前安全状况一般，请密切关注报警信息！"
    elif 40 <= score < 60:
        return "风险较高" if client_name == "wechat" else "当前安全状况较差，请重视报警信息！"
    elif 0 <= score < 40:
        return "高风险" if client_name == "wechat" else "当前安全状况很差，请重视报警信息并处理！"
    else:
        log.error(f"score:{score}, 安全指数不在0-100范围内")
        return ""


def health_status_res(score, client_name):
    """根据健康指数, 求健康状态"""
    if 90 <= score <= 100:
        return "状态优" if client_name == "wechat" else "当前健康状况良好，请继续保持！"
    elif 75 <= score < 90:
        return "状态较好" if client_name == "wechat" else "当前健康状况较好，请关注电能质量信息！"
    elif 60 <= score < 75:
        return "状况一般" if client_name == "wechat" else "当前健康状况一般，请密切关注电能质量信息！"
    elif 0 <= score < 60:
        return "状态差" if client_name == "wechat" else "当前健康状况差，请重视电能质量信息并改善！"
    else:
        log.error(f"score:{score}, 健康指数不在0-100范围内")
        return ""


def carbon_status_res(score):
    """根据碳排指数, 求达标情况, wechat使用"""
    if 80 <= score <= 100:
        return "达标"
    elif 0 <= score < 80:
        return "不达标"
    else:
        log.error(f"score:{score}, 碳排指数不在0-100范围内")
        return ""


def carbon_status_res_web(score):
    """根据碳排指数, 求达标情况, web使用"""
    if not isinstance(score, int):
        return ""
    if 90 <= score <= 100:
        return "当前能耗达标且水平很低，请继续保持！"
    elif 80 <= score < 90:
        return "当前能耗达标且水平较低，请继续保持！"
    elif 60 <= score < 80:
        return "当前能耗不达标，请关注能耗控制！"
    elif 0 <= score < 60:
        return "当前能耗不达标且水平较高，请重视能耗控制！"
    else:
        log.error(f"score:{score}, 碳排指数不在0-100范围内")
        return ""


def economic_index_desc(score, client_name):
    """经济指数, 返回文字说明"""
    if 90 <= score <= 100:
        return "状态良好" if client_name == "wechat" else "当前经济性良好，请继续保持"
    elif 75 <= score < 90:
        return "存在空间" if client_name == "wechat" else "存在空间，建议关注优化措施"
    elif 60 <= score < 75:
        return "空间较大" if client_name == "wechat" else "空间较大，建议尝试优化措施！"
    elif 0 <= score < 60:
        return "空间很大" if client_name == "wechat" else "空间很大，建议采取优化措施！"
    else:
        log.error(f"score:{score}, 经济指数不在0-100范围内")
        return ""
