import argparse from dataclasses import dataclass import pendulum 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"]: res[0] = pendulum.parse(res[0]).format("HH:mm") result.append(dict(zip(head, res))) return result