Mercurial > code > home > repos > homeauto

comparison service/beacon/beaconmap.py @ 291:299ddd7e2070

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
start bt beacon tools Ignore-this: a19bb907ede601562ef44c27ae706dca
author drewp@bigasterisk.com
date Wed, 20 Jul 2016 23:52:03 -0700
parents
children 6ba2c88f9847
comparison
equal deleted inserted replaced
290:7b5cff542078 291:299ddd7e2070
1 from __future__ import division
2 import sys, cyclone.web, json, datetime, time
3 import arrow
4 from twisted.internet import reactor, task
5 from pymongo import MongoClient
6 from dateutil.tz import tzlocal
7 import math
8 import cyclone.sse
9 from locator import Locator, Measurement
10
11 sys.path.append("/my/proj/homeauto/lib")
12 from cycloneerr import PrettyErrorHandler
13 from logsetup import log
14
15 class Devices(PrettyErrorHandler, cyclone.web.RequestHandler):
16 def get(self):
17 devices = []
18 startCount = datetime.datetime.now(tzlocal()) - datetime.timedelta(seconds=60*60*2)
19 filt = {
20 #"addr_type": "Public",
21 }
22 for addr in scan.distinct('addr', filt):
23 filtAddr = filt.copy()
24 filtAddr['addr'] = addr
25 row = scan.find_one(filtAddr, sort=[('t', -1)], limit=1)
26 filtAddrRecent = filtAddr.copy()
27 filtAddrRecent['t'] = {'$gt': startCount}
28 freq = scan.count(filtAddrRecent)
29 if not freq:
30 continue
31 name = None
32 if addr == "00:ea:23:23:c6:c4":
33 name = 'apollo'
34 if addr == "00:ea:23:21:e0:a4":
35 name = 'white'
36 if addr == "00:ea:23:24:f8:d4":
37 name = 'green'
38 if 'Eddystone-URL' in row:
39 name = row['Eddystone-URL']
40 devices.append({
41 'addr': addr,
42 'recentCount': freq,
43 'lastSeen': row['t'].isoformat(),
44 'name': name})
45 devices.sort(key=lambda d: (d['name'] or 'zzz',
46 -d['recentCount'],
47 d['addr']))
48 self.set_header("Content-Type", "application/json")
49 self.write(json.dumps({'devices': devices}))
50
51
52 class Poller(object):
53 def __init__(self):
54 self.listeners = [] # Points handlers
55
56 self.lastPointTime = {} # addr : secs
57 self.lastValues = {} # addr : {sensor: (secs, rssi)}
58 task.LoopingCall(self.poll).start(1)
59
60 def poll(self):
61 addrs = set(l.addr for l in self.listeners if l.addr)
62 seconds = 60 * 20
63 now = datetime.datetime.now(tzlocal())
64 startTime = (now - datetime.timedelta(seconds=seconds))
65 startTimeSec = arrow.get(startTime).timestamp
66 for addr in addrs:
67 points = {} # from: [offsetSec, rssi]
68 for row in scan.find({'addr': addr, 't': {'$gt': startTime},
69 #'addr_type': 'Public',
70 },
71 sort=[('t', 1)]):
72 t = (arrow.get(row['t']) - startTime).total_seconds()
73 points.setdefault(row['from'], []).append([
74 round(t, 2), row['rssi']])
75 self.lastValues.setdefault(addr, {})[row['from']] = (
76 now, row['rssi'])
77
78 for pts in points.values():
79 smooth(pts)
80
81 if not points:
82 continue
83
84 last = max(pts[-1][0] + startTimeSec for pts in points.values())
85 if self.lastPointTime.get(addr, 0) == last:
86 continue
87 self.lastPointTime[addr] = last
88 msg = json.dumps({
89 'addr': addr,
90 'startTime': startTimeSec,
91 'points': [{'from': k, 'points': v}
92 for k,v in sorted(points.items())]})
93 for lis in self.listeners:
94 if lis.addr == addr:
95 lis.sendEvent(msg)
96
97 def lastValue(self, addr, maxSensorAgeSec=30):
98 """note: only considers actively polled addrs"""
99 out = {} # from: rssi
100 now = datetime.datetime.now(tzlocal())
101 for sensor, (t, rssi) in self.lastValues.get(addr, {}).iteritems():
102 print 'consider %s %s' % (t, now)
103 if (now - t).total_seconds() < maxSensorAgeSec:
104 out[sensor] = rssi
105 return out
106
107 def smooth(pts):
108 # see https://filterpy.readthedocs.io/en/latest/kalman/UnscentedKalmanFilter.html
109 for i in range(0, len(pts)):
110 if i == 0:
111 prevT, smoothX = pts[i]
112 else:
113 t, x = pts[i]
114 if t - prevT < 30:
115 smoothX = .8 * smoothX + .2 * x
116 else:
117 smoothX = x
118 pts[i] = [t, round(smoothX, 1)]
119 prevT = t
120
121 class Points(cyclone.sse.SSEHandler):
122 def __init__(self, application, request, **kw):
123 cyclone.sse.SSEHandler.__init__(self, application, request, **kw)
124 if request.headers['accept'] != 'text/event-stream':
125 raise ValueError('ignoring bogus request')
126 self.addr = request.arguments.get('addr', [None])[0]
127
128 def bind(self):
129 if not self.addr:
130 return
131 poller.listeners.append(self)
132 def unbind(self):
133 if not self.addr:
134 return
135 poller.listeners.remove(self)
136
137 class LocatorEstimatesPoller(object):
138 def __init__(self):
139 self.listeners = []
140 self.lastResult = {}
141 self.locator = Locator()
142 task.LoopingCall(self.poll).start(1)
143
144 def poll(self):
145 addrs = set(l.addr for l in self.listeners if l.addr)
146 now = datetime.datetime.now(tzlocal())
147 cutoff = (now - datetime.timedelta(seconds=60))
148
149 for addr in addrs:
150 d = {} # from: [(t, rssi)]
151 for row in scan.find({'addr': addr, 't': {'$gt': cutoff}},
152 sort=[('t', 1)]):
153 d.setdefault(row['from'], []).append((arrow.get(row['t']).timestamp, row['rssi']))
154
155 for pts in d.values():
156 smooth(pts)
157 meas = Measurement(dict((k, v[-1][1]) for k, v in d.items()))
158 nearest = [
159 (dist, coord) for dist, coord in self.locator.nearestPoints(meas) if dist < 25
160 ]
161 if nearest:
162 floors = [row[1][2] for row in nearest]
163 freqs = [(floors.count(z), z) for z in floors]
164 freqs.sort()
165 bestFloor = freqs[-1][1]
166 sameFloorMatches = [(dist, coord) for dist, coord in nearest
167 if coord[2] == bestFloor]
168 weightedCoord = [0, 0, 0]
169 totalWeight = 0
170 for dist, coord in sameFloorMatches:
171 weight = 25 / (dist + .001)
172 totalWeight += weight
173 for i in range(3):
174 weightedCoord[i] += weight * coord[i]
175 for i in range(3):
176 weightedCoord[i] /= totalWeight
177
178 self.lastResult[addr] = {'nearest': nearest, 'weightedCoord': weightedCoord}
179
180 for lis in self.listeners:
181 lis.sendEvent(self.lastResult[addr])
182
183
184 class PositionEstimates(cyclone.sse.SSEHandler):
185 def __init__(self, application, request, **kw):
186 cyclone.sse.SSEHandler.__init__(self, application, request, **kw)
187 if request.headers['accept'] != 'text/event-stream':
188 raise ValueError('ignoring bogus request')
189 self.addr = request.arguments.get('addr', [None])[0]
190
191 def bind(self):
192 if not self.addr:
193 return
194 locatorEstimatesPoller.listeners.append(self)
195 def unbind(self):
196 if not self.addr:
197 return
198 locatorEstimatesPoller.listeners.remove(self)
199
200
201
202 class Sensors(PrettyErrorHandler, cyclone.web.RequestHandler):
203 def get(self):
204 t1 = datetime.datetime.now(tzlocal()) - datetime.timedelta(seconds=60*10)
205 out = []
206 for sens in scan.distinct('from', {'t': {'$gt': t1}}):
207 rssiHist = {} # level: count
208 for row in scan.find({'from': sens, 't': {'$gt': t1}},
209 {'_id': False, 'rssi': True}):
210 bucket = (row['rssi'] // 5) * 5
211 rssiHist[bucket] = rssiHist.get(bucket, 0) + 1
212
213 recent = {}
214 for row in scan.find({'from': sens},
215 {'_id': False,
216 'addr': True,
217 't': True,
218 'rssi': True,
219 'addr_type': True},
220 sort=[('t', -1)],
221 modifiers={'$maxScan': 100000}):
222 addr = row['addr']
223 if addr not in recent:
224 recent[addr] = row
225 recent[addr]['t'] = arrow.get(recent[addr]['t']).timestamp
226
227 out.append({
228 'from': sens,
229 'count': sum(rssiHist.values()),
230 'hist': rssiHist,
231 'recent': sorted(recent.values())
232 })
233
234 self.set_header("Content-Type", "application/json")
235 self.write(json.dumps({'sensors': out}))
236
237
238
239 class Sensor(PrettyErrorHandler, cyclone.web.RequestHandler):
240 def get(self):
241 from_ = self.get_argument('from')
242 if not from_:
243 return
244 seconds = int(self.get_argument('secs', default=60 * 2))
245 startTime = (datetime.datetime.now(tzlocal()) -
246 datetime.timedelta(seconds=seconds))
247 points = {} # addr : [offsetSec, rssi]
248 for row in scan.find({'from': from_, 't': {'$gt': startTime},
249 #'addr_type': 'Public',
250 },
251 {'_id': False,
252 'addr': True,
253 't': True,
254 'rssi': True,
255 },
256 sort=[('t', 1)]):
257 points.setdefault(row['addr'], []).append([
258 round((arrow.get(row['t']) - startTime).total_seconds(), 2),
259 row['rssi']])
260
261 self.set_header("Content-Type", "application/json")
262 self.write(json.dumps({
263 'sensor': from_,
264 'startTime': arrow.get(startTime).timestamp,
265 'points': points}))
266
267 class Save(PrettyErrorHandler, cyclone.web.RequestHandler):
268 def post(self):
269 lines = open('saved_points').readlines()
270 lineNum = len(lines) + 1
271 row = poller.lastValue('00:ea:23:21:e0:a4')
272 with open('saved_points', 'a') as out:
273 out.write('%s %r\n' % (lineNum, row))
274 self.write('wrote line %s: %r' % (lineNum, row))
275
276 scan = MongoClient('bang', 27017, tz_aware=True)['beacon']['scan']
277 poller = Poller()
278 locatorEstimatesPoller = LocatorEstimatesPoller()
279
280 reactor.listenTCP(
281 9113,
282 cyclone.web.Application([
283 (r"/(|.*\.(?:js|html|json))$", cyclone.web.StaticFileHandler, {
284 "path": ".", "default_filename": "beaconmap.html"}),
285 (r"/devices", Devices),
286 (r'/points', Points),
287 (r'/sensors', Sensors),
288 (r'/sensor', Sensor),
289 (r'/save', Save),
290 (r'/positionEstimates', PositionEstimates),
291 ]))
292 log.info('serving on 9113')
293 reactor.run()