-
Notifications
You must be signed in to change notification settings - Fork 0
/
alerter.py
345 lines (333 loc) · 14.5 KB
/
alerter.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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
#!/usr/bin/python
import sys
reload(sys)
sys.setdefaultencoding('utf8')
import time
import json
import datetime
import re
import copy
import utils
import db
import logger
import config
log = logger.get_logger(__name__)
conf = config.get_config()
import scheduler
schedule = scheduler.get_scheduler()
import output
import sensors
import image_utils
# for an image apply the configured object detection techniques
def analyze_image(sensor,data):
alert_text = ""
if len(data) == 0: return [alert_text]
# detect movements if there are at least two images
if "motion_detection" in sensor and len(data) >= 2:
result = image_utils.detect_movement(sensor,data,is_base64=True)
if result is not None:
difference = result[0]
image = result[1]
if difference > 0: log.info("["+sensor["module_id"]+"]["+sensor["group_id"]+"]["+sensor["sensor_id"]+"] motion detected: "+str(difference)+"%")
if difference > sensor["motion_detection"]["threshold"]:
# save the image on disk
image_utils.save_tmp_image("motion_detection",image)
alert_text = utils.lang(sensor["motion_detection"]["display_name"])+" ("+str(difference)+"%)"
# detect objects in the last image
if "object_detection" in sensor:
result = image_utils.detect_objects(sensor,data[(len(data)-1)],is_base64=True)
if result is not None:
# detect something
text = result[0]
image = result[1]
# save the image on disk
image_utils.save_tmp_image("objects_detection",image)
# save a new image with the object highlighted into the sensor
measures = []
measure = {}
measure["key"] = sensor["sensor_id"]
measure["value"] = image
measures.append(measure)
sensors.store(sensor,measures)
# return the alert text
log.info("["+sensor["module_id"]+"]["+sensor["group_id"]+"]["+sensor["sensor_id"]+"] object detected: "+text)
alert_text = text
return [alert_text]
# retrieve for the database the requested data
def get_data(sensor,request):
split = request.split(',')
key = split[0]
start = split[1]
end = split[2]
transform = split[3] if len(split) > 3 else None
key_split = key.split(":")
# adjust start and end based on the request
query = None
if utils.is_number(start) and utils.is_number(end):
# request range with start and end the relative positions
query = db.range
else:
# request a timerange with start and end relative times from now
query = db.rangebyscore
start = utils.string2timestamp(start)
end = utils.string2timestamp(end)
# remove the module from the key
key = key.replace(key_split[0]+":","",1)
key = conf["constants"]["db_schema"]["root"]+":"+key_split[0]+":"+key
# handle special requests
if transform is not None and transform == "elapsed":
# retrieve the timestamp and calculate the time difference
data = query(key,start=start,end=end,withscores=True)
if len(data) == 0: return []
time_diff = (utils.now() - data[0][0])/60
return [time_diff]
elif transform is not None and transform == "timestamp":
# retrieve the timestamp
data = query(key,start=start,end=end,withscores=True)
if len(data) == 0: return []
return [data[0][0]]
elif transform is not None and transform == "distance":
# calculate the distance between the point and our location
data = query(key,start=start,end=end,withscores=False,formatter=conf["constants"]["formats"][sensor["format"]]["formatter"])
if len(data) == 0: return []
data = json.loads(data[0])
distance = utils.distance([data["latitude"],data["longitude"]],[conf["general"]["latitude"],conf["general"]["longitude"]])
return [int(distance)]
elif transform is not None and transform == "count":
# count the number of items selected
data = query(key,start=start,end=end,withscores=False)
return [len(data)]
else:
# just retrieve the data
data = query(key,start=start,end=end,withscores=False,formatter=conf["constants"]["formats"][sensor["format"]]["formatter"])
if sensor["format"] == "calendar": data = utils.parse_calendar(data)
if sensor["format"] == "position":
# define the key to return
key = "label"
if transform is not None and transform == "text": key = "text"
data = utils.parse_position(data,key)
if sensor["format"] == "image": data = analyze_image(sensor,data)
return data
# evaluate if a condition is met
def is_true(a,operator,b):
evaluation = True
# get a's value
if not isinstance(a,list): a = [a]
a = a[0]
# prepare b's value
if not isinstance(b,list): b = [b]
# b can be have multiple values, cycle through all of them
for value in b:
if value is None or a is None: evaluation = False
elif operator == "==":
if value != a: evaluation = False
elif operator == "!=":
if value == a: evaluation = False
elif operator == ">":
if not utils.is_number(value) or not utils.is_number(a): return False
if float(value) >= float(a): evaluation = False
elif operator == "<":
if not utils.is_number(value) or not utils.is_number(a): return False
if float(value) <= float(a): evaluation = False
else: evaluation = False
# return the evaluation
return evaluation
# calculate a sub expression
def sub_expression(a,operator,b):
# prepare the values (can be an array)
if isinstance(a,list): a = a[0]
if isinstance(b,list): b = b[0]
# perform integrity checks
if a is None or b is None: return None
if not utils.is_number(a) or not utils.is_number(b): return None
# calculate the expression
if operator == "+": return float(a)+float(b)
elif operator == "-": return float(a)-float(b)
elif operator == "*": return float(a)*float(b)
elif operator == "/": return float(a)/float(b)
return None
# evaluate if the given alert has to trigger
def run(module_id,rule_id,notify=True):
alert_text = ""
try:
module = utils.get_module(module_id)
for rule_template in module["rules"]:
if not rule_template["enabled"]: continue
# retrive the rule for the given rule_id
if rule_template["rule_id"] != rule_id: continue
# for each variable (if provided) run a different evaluation
variables = [""]
variable_sensor = None
if "for" in rule_template: variables = rule_template["for"]
for variable in variables:
# ensure the variable is a valid sensor
if variable != '' and utils.is_sensor(variable):
variable_sensor = utils.get_sensor_string(variable)
if variable_sensor is None:
log.error("invalid variable sensor "+variable)
continue
# restore the template
rule = copy.deepcopy(rule_template)
# for each definition retrieve the data
definitions = {}
suffix = {}
valid_data = True
for definition in rule["definitions"]:
if utils.is_sensor(rule["definitions"][definition]):
rule["definitions"][definition] = rule["definitions"][definition].replace("%i%",variable)
# check if the sensor exists
split = rule["definitions"][definition].split(',')
key = split[0]
start = split[1]
end = split[2]
sensor = utils.get_sensor_string(key)
if sensor is None:
log.error("invalid sensor "+key)
valid_data = False
break
sensors.init_plugins()
sensor = sensors.init_sensor(sensor)
# retrieve and store the data
definitions[definition] = get_data(sensor,rule["definitions"][definition])
if len(definitions[definition]) == 0:
log.debug("invalid data from sensor "+key)
valid_data = False
break
# store the suffix
suffix[definition] = conf["constants"]["formats"][sensor["format"]]["suffix"].encode('utf-8')
else:
definitions[definition] = rule["definitions"][definition]
# if not all the data is valid, return
if not valid_data: continue
# for each condition check if it is true
evaluation = True
for condition in rule["conditions"]:
condition = re.sub(' +',' ',condition)
# look for sub expressions and calculate them
expressions = re.findall("\(([^\)]+)\)",condition)
for i in range(len(expressions)):
expression = expressions[i]
placeholder = "%exp_"+str(i)+"%"
exp1,operator,exp2 = expression.split(' ')
# calculate the sub expression
exp_value = sub_expression(definitions[exp1],operator,definitions[exp2])
log.debug("["+module_id+"]["+rule_id+"] resolving "+exp1+" ("+str(definitions[exp1])+") "+operator+" "+exp2+" ("+str(definitions[exp2])+"): "+str(exp_value)+" (alias "+placeholder+")")
# add the sub expressions to the definitions
definitions[placeholder] = exp_value
condition = condition.replace("("+expression+")",placeholder)
# do the comparison and apply the condition
a,operator,b = condition.split(' ')
sub_evaluation = is_true(definitions[a],operator,definitions[b])
log.debug("["+module_id+"]["+rule_id+"] evaluating "+a+" ("+str(definitions[a])+") "+operator+" "+b+" ("+str(definitions[b])+"): "+str(sub_evaluation))
if not sub_evaluation: evaluation = False
log.debug("["+module_id+"]["+rule_id+"] evaluates to "+str(evaluation))
# evaluate the conditions
if not evaluation: continue
# alert has triggered, prepare the alert text
alert_text = utils.lang(rule["display_name"])
# replace the variable if needed
if variable_sensor is not None: alert_text = alert_text.replace("%i%",utils.lang(variable_sensor["display_name"]))
# replace the definitions placeholders
for definition in rule["definitions"]:
value = definitions[definition][0] if isinstance(definitions[definition],list) else definitions[definition]
# apply aliases
if "aliases" in rule:
for to_find,to_replace in rule["aliases"].iteritems():
if str(value) == str(to_find): value = to_replace
# add the suffix
if utils.is_sensor(rule["definitions"][definition]) and "elapsed" in rule["definitions"][definition]:
value = str(value)+" minutes"
if utils.is_sensor(rule["definitions"][definition]) and "timestamp" in rule["definitions"][definition]:
value = utils.timestamp2date(value)
if utils.is_sensor(rule["definitions"][definition]) and "distance" in rule["definitions"][definition]:
if conf["general"]["units"]["imperial"]: value = str(value)+" miles"
else: value = str(value)+" km"
elif utils.is_sensor(rule["definitions"][definition]): value = str(value)+suffix[definition]
alert_text = alert_text.replace("%"+definition+"%",str(value))
# execute an action
if "actions" in rule:
for action in rule["actions"]:
# replace the definitions placeholders
action = action.replace("%i%",variable)
for definition in rule["definitions"]:
value = definitions[definition][0] if isinstance(definitions[definition],list) else definitions[definition]
action = action.replace("%"+definition+"%",str(value))
# parse the action
split = action.split(',')
what = split[0]
key = split[1]
value = split[2]
force = True if len(split) > 3 and split[3] == "force" else False
ifnotexists = True if len(split) > 3 and split[3] == "ifnotexists" else False
# ensure the target sensor exists
sensor = utils.get_sensor_string(key)
if sensor is None:
log.warning("["+rule["rule_id"]+"] invalid sensor "+key)
continue
# execute the requested action
if what == "send": sensors.data_send(sensor["module_id"],sensor["group_id"],sensor["sensor_id"],value,force=force)
elif what == "set": sensors.data_set(sensor["module_id"],sensor["group_id"],sensor["sensor_id"],value,ifnotexists=ifnotexists)
# notify about the alert
if rule["severity"] == "none": notify = False
if notify:
log.info("["+module_id+"]["+rule_id+"]["+rule["severity"]+"] "+alert_text)
if rule["severity"] != "debug":
db.set(conf["constants"]["db_schema"]["alerts"]+":"+rule["severity"],alert_text,utils.now())
output.notify(rule["severity"],alert_text)
except Exception,e:
log.warning("error while running rule "+module_id+":"+rule_id+": "+utils.get_exception(e))
return alert_text
# purge old data from the database
def expire():
total = 0
for stat in [':alert',':warning',':info']:
key = conf["constants"]["db_schema"]["alerts"]+stat
if db.exists(key):
deleted = db.deletebyscore(key,"-inf",utils.now()-conf["alerter"]["data_expire_days"]*conf["constants"]["1_day"])
log.debug("expiring from "+stat+" "+str(total)+" items")
total = total + deleted
log.info("expired "+str(total)+" items")
# run the given schedule
def run_schedule(run_every):
# for each module
log.debug("evaluate all the rules configured to run every "+run_every)
for module in conf["modules"]:
if not module["enabled"]: continue
if "rules" not in module: continue
# for each configured rule
for rule in module["rules"]:
if not rule["enabled"]: continue
if rule["run_every"] != run_every: continue
# if the rule has the given run_every, run it
run(module["module_id"],rule["rule_id"])
# schedule both hourly and daily alerts
def schedule_all():
log.info("starting alerter module...")
# run now startup rules
schedule.add_job(run_schedule,'date',run_date=datetime.datetime.now(),args=["startup"])
# schedule minute, hourly and daily jobs
schedule.add_job(run_schedule,'cron',second="30",args=["minute"])
schedule.add_job(run_schedule,'cron',minute="*/5",args=["5 minutes"])
schedule.add_job(run_schedule,'cron',minute="*/10",args=["10 minutes"])
schedule.add_job(run_schedule,'cron',minute="*/30",args=["30 minutes"])
schedule.add_job(run_schedule,'cron',minute="1",args=["hour"])
schedule.add_job(run_schedule,'cron',hour="1",args=["day"])
# schedule an expire job
schedule.add_job(expire,'cron',hour="1")
# return the latest alerts for a web request
def data_get_alerts(severity,timeframe):
start = utils.recent()
if timeframe == "recent": start = utils.recent(hours=conf["general"]["timeframes"]["alerter_recent_hours"])
if timeframe == "history": start = utils.history(days=conf["general"]["timeframes"]["alerter_history_days"])
return json.dumps(db.rangebyscore(conf["constants"]["db_schema"]["alerts"]+":"+severity,start,utils.now(),withscores=True,format_date=True))
# allow running it both as a module and when called directly
if __name__ == '__main__':
if len(sys.argv) != 3:
# no arguments provided, schedule all alerts
schedule.start()
schedule_all()
while True:
time.sleep(1)
else:
# <module_id> <rule_id>
run(sys.argv[1],sys.argv[2])