-
Notifications
You must be signed in to change notification settings - Fork 0
/
output_calculator.py
executable file
·186 lines (156 loc) · 8.38 KB
/
output_calculator.py
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
183
184
185
186
import copy
import json
import logging
import datetime
import os
import sys
import argparse
import urllib3
import warnings
import numpy as np
import pandas as pd
from influxdb import DataFrameClient
from classes.alerts import SlackClient
warnings.filterwarnings("ignore")
urllib3.disable_warnings()
def calc_output(measurement, output_cfg, year, start_day, end_day):
str_locations = '('
for location in output_cfg['sourceLocations']:
str_locations = '%slocation=\'%s\' OR ' % (str_locations, location)
str_locations = '%s)' % str_locations[0:-4]
if output_cfg['aggregations']['hourly'] is not False:
# Query handling the hourly aggregation
query = "select %s(value) from %s where signal='%s' and %s AND " \
"time>='%s-%sT00:00:00Z' AND time<='%s-%sT23:59:59Z' " \
"GROUP BY time(1h), location" % (output_cfg['aggregations']['hourly'], measurement,
output_cfg['sourceSignal'],
str_locations, year, start_day, year, end_day)
else:
# Query getting data without hourly aggregation
query = "select value from %s where signal='%s' and %s AND time>='%s-%sT00:00:00Z' AND " \
"time<='%s-%sT23:59:59Z' GROUP BY location" % (measurement, output_cfg['sourceSignal'], str_locations,
year, start_day, year, end_day)
logger.info('Query: %s' % query)
res = influx_client.query(query)
df_daily_locs = list()
for location in output_cfg['sourceLocations']:
df_tmp = res[(measurement, (('location', location),))]
# Check if the the dataframe contains data
if len(df_tmp) > 0:
df_tmp.columns = [location]
df_tmp['dt'] = df_tmp.index
# Grouping handling the daily aggregation
if output_cfg['aggregations']['daily'] == 'max':
df_tmp_agg = df_tmp.groupby(pd.Grouper(key='dt', axis=0, freq='D')).max()
elif output_cfg['aggregations']['daily'] == 'mean':
df_tmp_agg = df_tmp.groupby(pd.Grouper(key='dt', axis=0, freq='D')).mean()
elif output_cfg['aggregations']['daily'] == 'min':
df_tmp_agg = df_tmp.groupby(pd.Grouper(key='dt', axis=0, freq='D')).min()
df_daily_locs.append(df_tmp_agg)
df_ret = pd.concat(df_daily_locs, axis=1)
if output_cfg['outlierFiltering']['enabled'] is True:
df_ret = filter_outlier(copy.deepcopy(df_ret), output_cfg)
# Final handling related to the location aggregation level
if output_cfg['aggregations']['locations'] == 'max':
return df_ret.max(axis=1)
elif output_cfg['aggregations']['locations'] == 'mean':
return df_ret.mean(axis=1)
elif output_cfg['aggregations']['locations'] == 'min':
return df_ret.min(axis=1)
return None
def filter_outlier(df, output_cfg):
for index, row in df.iterrows():
outliers = []
cnt_outliers = 0
row_values = row.values
for i in range(len(row_values)):
other_values = np.delete(row_values, i)
centroid = np.mean(other_values)
distance = row_values[i] - centroid
if distance > output_cfg['outlierFiltering']['threshold']:
cnt_outliers += 1
outliers.append(df.columns[i])
if 0 < cnt_outliers <= output_cfg['outlierFiltering']['numMaxOutliers']:
logger.warning(f"Found outliers [{index.strftime('%Y-%m-%d')}] -> {outliers}")
if output_cfg['outlierFiltering']['setNan']:
df.at[index, df.columns[i]] = np.nan
return df
if __name__ == "__main__":
# --------------------------------------------------------------------------- #
# Configuration file
# --------------------------------------------------------------------------- #
arg_parser = argparse.ArgumentParser()
arg_parser.add_argument("-c", help="configuration file")
arg_parser.add_argument("-p", help="period (today | yesterday | custom)")
arg_parser.add_argument("-l", help="log file (optional, if empty log redirected on stdout)")
args = arg_parser.parse_args()
# Load the main parameters
config_file = args.c
if os.path.isfile(config_file) is False:
print('\nATTENTION! Unable to open configuration file %s\n' % config_file)
sys.exit(1)
cfg = json.loads(open(args.c).read())
period = args.p
# Load the connections parameters and update the config dict with the related values
cfg_conns = json.loads(open(cfg['connectionsFile']).read())
cfg.update(cfg_conns)
# --------------------------------------------------------------------------- #
# Set logging object
# --------------------------------------------------------------------------- #
if not args.l:
log_file = None
else:
log_file = args.l
logger = logging.getLogger()
logging.basicConfig(format='%(asctime)-15s::%(levelname)s::%(funcName)s::%(message)s', level=logging.INFO,
filename=log_file)
logger.info('Starting program')
logger.info('Connection to InfluxDb server on socket [%s:%s]' % (cfg['influxDB']['host'], cfg['influxDB']['port']))
try:
influx_client = DataFrameClient(host=cfg['influxDB']['host'], port=cfg['influxDB']['port'],
password=cfg['influxDB']['password'], username=cfg['influxDB']['user'],
database=cfg['influxDB']['database'], ssl=cfg['influxDB']['ssl'])
except Exception as e:
logger.error('EXCEPTION: %s' % str(e))
sys.exit(3)
logger.info('Connection successful')
measurement = cfg['influxDB']['measurementInputsMeasurements']
str_res = ''
for output_cfg in cfg['output']:
tags = {'signal': output_cfg['targetSignal'], 'location': output_cfg['region']}
if period == 'today':
dt = datetime.datetime.now()
ret_dataset = calc_output(measurement, output_cfg, dt.year, '%02d-%02d' % (dt.month, dt.day),
'%02d-%02d' % (dt.month, dt.day))
influx_client.write_points(pd.DataFrame({'value': ret_dataset}), measurement, tags=tags)
logger.info('%s[%s][%s] = %.1f' % (tags['signal'], tags['location'], ret_dataset.index[0].date(),
ret_dataset.values[0]))
str_res = '%s%s: %s[%s] = %.1f\n' % (str_res, tags['location'], tags['signal'], ret_dataset.index[0].date(),
ret_dataset.values[0])
elif period == 'yesterday':
dt = datetime.datetime.now() - datetime.timedelta(days=1)
ret_dataset = calc_output(measurement, output_cfg, dt.year, '%02d-%02d' % (dt.month, dt.day),
'%02d-%02d' % (dt.month, dt.day))
influx_client.write_points(pd.DataFrame({'value': ret_dataset}), measurement, tags=tags)
logger.info('%s[%s][%s] = %.1f' % (tags['signal'], tags['location'], ret_dataset.index[0].date(),
ret_dataset.values[0]))
str_res = '%s%s: %s[%s] = %.1f\n' % (str_res, tags['location'], tags['signal'], ret_dataset.index[0].date(),
ret_dataset.values[0])
else:
# Configuration variables
start_day = cfg['period']['customSettings']['startDay']
end_day = cfg['period']['customSettings']['endDay']
for year in cfg['period']['customSettings']['years']:
ret_dataset = calc_output(measurement, output_cfg, year, start_day, end_day)
influx_client.write_points(pd.DataFrame({'value': ret_dataset}), measurement, tags=tags)
logger.info('Finished year %s for output %s, region %s' % (year, output_cfg['targetSignal'],
output_cfg['region']))
# Send results to Slack
if cfg['alerts']['slack']['enabled'] is True:
slack_client = SlackClient(logger, cfg)
slack_client.send_alert_message('%s RESULTS:' % period.upper(), '#000000')
if len(str_res) > 0:
slack_client.send_alert_message(str_res, '#00ff00')
else:
slack_client.send_alert_message('DATA NOT AVAILABLE', '#ff0000')
logger.info('Ending program')