1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
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