common_utils.py 4.69 KB
import argparse
from dataclasses import dataclass
from chinese_calendar import is_workday, is_holiday
from datetime import datetime
import re
from pandas import Series
from pot_libs.logger import log


class NumListHelper(object):
    @classmethod
    def parse_num_list(cls, string):
        """ parse num list like: '1-6' or '2'.
        # NOTE: DO NOT use string like '1 -6' or '1 - 6', it will raise
        an exception.
        """
        m = re.match(r'(\d+)(?:-(\d+))?$', string)
        if not m:
            raise argparse.ArgumentTypeError("'" + string + (
                "' is not a range of number. Expected forms like '0-5'."))
        start = m.group(1)
        end = m.group(2) or start
        return list(range(int(start, 10), int(end, 10) + 1))

    @classmethod
    def parse_comma_delimited_num(cls, string):
        """ parse comma delimited num list like: '1,3-8,10-15'.
        # NOTE: the return list is NOT sorted.
        """
        num_str_l = [i.strip() for i in string.split(',')]
        num_str_l = filter(lambda x: x != '', num_str_l)
        n_list = []  # final format like: [1,2,3,4,9,10,11,12]
        for n_str in num_str_l:
            n_list.extend(cls.parse_num_list(n_str))
        n_list = list(set(n_list))  # revmove duplicated items
        return n_list


def choose_list(s, round_num=2):
    """求列表s=[]求 元素个数,最大值,最小值,元素和,平均值"""
    count = 0
    total = 0
    maxnum = max(s)
    minnum = min(s)
    for i in s:
        count = count + 1  # 元素个数
        total = total + i
    average = total / count
    # 最大最小值的索引
    max_index = s.index(maxnum)
    min_index = s.index(minnum)
    return round(count, round_num), round(maxnum, round_num), \
           round(minnum, round_num), round(average,
                                           round_num), max_index, min_index


def round_0(value):
    if value == 0:
        return 0
    elif not value:
        return ""
    else:
        return round(value)


def round_2(value):
    if value == 0:
        return 0
    elif not value:
        return ""
    else:
        return round(value, 2)


def round_2n(value):
    if value == 0:
        return 0
    elif not value:
        return None
    else:
        return round(value, 2)


def round_1(value):
    if value == 0:
        return 0
    elif not value:
        return ""
    else:
        return round(value, 1)


def round_4(value):
    if value == 0:
        return 0
    elif not value:
        return ""
    else:
        return round(value, 4)


def process_es_data(data, key='key'):
    res = {}
    for _item in data:
        res[_item[key]] = _item
    return res


@dataclass
class ChineseCalendar:
    date_str: str

    def is_workday(self) -> bool:
        return is_workday(datetime.strptime(self.date_str, "%Y-%m-%d"))

    def is_holiday(self) -> bool:
        return is_holiday(datetime.strptime(self.date_str, "%Y-%m-%d"))


def multiplication_two(a, b):
    """两个值的乘法"""
    try:
        return a * b
    except Exception:
        return ""


def division_two(a, b):
    """两个值的除法"""
    try:
        return a / b
    except Exception:
        return ""


def correlation(l1, l2):
    """求两个列表相关系数"""
    """
    当γ = -1时,完全相关;
    当γ≥0时,不相关;
    当 - 0.3≤γ<0时,微弱相关;
    当 - 0.5≤γ<-0.3时,低度相关;
    当 - 0.8≤γ<-0.5时,显著相关;
    当 - 0.1<γ<-0.8时,高度相关
    """
    s1 = Series(l1)
    s2 = Series(l2)
    r = round_2(s1.corr(s2))
    if r == -1:
        return r, "完全相关"
    elif r >= 0:
        return r, "不相关"
    elif 0 > r >= -0.3:
        return r, "微弱相关"
    elif -0.3 > r >= -0.5:
        return r, "低度相关"
    elif -0.5 > r >= -0.8:
        return r, "显著相关"
    elif -0.8 > r > -1:
        return r, "显著相关"
    else:
        log.info(f"求{l1}, {l2}相关系数{r}, 超出判断范围")
        return "", ""


def make_tdengine_data_as_list(tdengine_data):
    """
    将tdengine查询到的数据装换成 列表形式  列表中的数据为字典形式 {字段-字段值、}
    适合处理 多表last_row数据
    :param tdengine_data: {
        "column_meta":[["key1", type, len],["key2", type, len],["key3", type, len]],
        "data":[["val1","val2","id1"], ["val3","val4,"id2""]], ....}
    :return: {
        {"key1":"val1", "key2":"val2"}, {"key1":"val3", "key2":"val4"}
    ]
    """
    head = [re.findall(r'last_row\((.*)\)', meta[0])[0] if "(" in meta[0] else
            meta[0] for meta in tdengine_data["column_meta"]]
    result = []
    for res in tdengine_data["data"]:
        result.append(dict(zip(head, res)))
    return result