forked from ZoneMinder/pyzm
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Events.py
131 lines (106 loc) · 4.62 KB
/
Events.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
"""
Events
=======
Holds a list of Events for a ZM configuration
You can pass different conditions to filter the events
Each invocation results in a new API call as events are very dynamic
You are not expected to use this module directly. It is instantiated when you use the :meth:`pyzm.api.ZMApi.events` method of :class:`pyzm.api`
"""
from pyzm.helpers.Base import Base
from pyzm.helpers.Event import Event
import requests
import dateparser
class Events(Base):
def __init__(self, logger=None, api=None, options=None):
super().__init__(logger)
self.api = api
self.events = []
self.pagination = {}
self._load(options)
def list(self):
"""Returns list of event
Returns:
list -- events of :class:`.Event`
"""
return self.events
def count(self):
"""Returns number of events retrieved in previous invocation
Returns:
int -- number of events
"""
return int(self.pagination.get('count'))
def _load(self, options={}):
self.logger.Debug(1,'Retrieving events via API')
url_filter=''
tz = {}
if options.get('tz'):
tz = {'TIMEZONE': options.get('tz')}
#print ('USING ',tz)
if options.get('from'):
from_list = options.get('to').split(" to ", 1)
if len(from_list) == 2:
url_filter += '/StartTime >=:'+dateparser.parse(from_list[0], settings=tz).strftime('%Y-%m-%d %H:%M:%S')
url_filter += '/StartTime <=:' + dateparser.parse(from_list[1], settings=tz).strftime('%Y-%m-%d %H:%M:%S')
else:
url_filter += '/StartTime >=:' + dateparser.parse(from_list[0], settings=tz).strftime('%Y-%m-%d %H:%M:%S')
if options.get('to'):
to_list = options.get('to').split(" to ", 1)
if len(to_list) == 2:
url_filter += '/EndTime <=:'+dateparser.parse(to_list[1], settings=tz).strftime('%Y-%m-%d %H:%M:%S')
url_filter += '/EndTime >=:' + dateparser.parse(to_list[0], settings=tz).strftime('%Y-%m-%d %H:%M:%S')
else:
url_filter += '/EndTime <=:' + dateparser.parse(to_list[0], settings=tz).strftime('%Y-%m-%d %H:%M:%S')
if options.get('mid'):
url_filter += '/MonitorId =:'+str(options.get('mid'))
if options.get('min_alarmed_frames'):
url_filter += '/AlarmFrames >=:'+str(options.get('min_alarmed_frames'))
if options.get('max_alarmed_frames'):
url_filter += '/AlarmFrames <=:'+str(options.get('max_alarmed_frames'))
if options.get('object_only'):
url_filter += '/Notes REGEXP:detected:'
# catch all
if options.get('raw_filter'):
url_filter+=options.get('raw_filter')
#print ('URL filter: ',url_filter)
# tbd - no need for url_prefix in options
url_prefix = options.get('url_prefix',self.api.api_url + '/events/index' )
url = url_prefix + url_filter +'.json'
params = {
'sort':'StartTime',
'direction': 'desc',
'page': 1
}
for k in options:
if k in params:
params[k] = options[k]
numevents = 100
if options.get('max_events'): numevents = options.get('max_events')
if options.get('limit'): numevents = options.get('limit')
params['limit'] = numevents
currevents = 0
self.events = []
events= []
while True:
r = self.api._make_request(url=url, query=params)
events.extend(r.get('events'))
pagination = r.get('pagination')
self.pagination = pagination
if not pagination:
break
if not pagination.get('nextPage'):
break
currevents += int(pagination.get('current'))
if currevents >= numevents:
break
params['page'] +=1
for event in events:
self.events.append(Event(event=event,api=self.api,logger= self.logger))
def get(self, options={}):
"""Returns the full list of events. Typically useful if you need access to data for which you don't have an easy getter
Keyword Arguments:
- options: dict with same parameters as the one you pass in :meth:`pyzm.api.ZMApi.events`. This is really a convenience instead of re-creating the instance.
Returns:
list -- of :class:`pyzm.helpers.Events`
"""
self._load(options)
return self.events