-
Notifications
You must be signed in to change notification settings - Fork 2
/
radar.py
478 lines (382 loc) · 18.4 KB
/
radar.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
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
import tuner
import squitter
import basic
import Queue
import threading
import curses
import logging
import logging.handlers
import time
import sys
import server
import operator
import os
import simplejson
import shutil
__author__ = 'Wolfrax'
"""
This module implements the application using 2 classes
1. Radar, it runs in a separate thread, reads from a message queue, decode messages and adds to a display queue
The display queue is read recurrently (once per sec) from a RepeatTimer thread and is displayed
2. TextDisplay, implements a text table format using curses. It reads quitter objects and format the output.
The main application initiates a tuner object and a radar object. The tuner runs the main thread and passes
information to the radar object through a callback function (Radar - tuner_read).
The tuner_read method creates a Squitter object, do some basic parsing of the message and stores it into a
message queue (msgQ). The message passed from the tuner is in the format
[[signal strength (float), message (long)], [signal strength, message], ...]
signal strength is a simple measure of (max sample - min sample) in the preamble divided by max range (65535)
expressed in %. Thus signal_strength = 10 means that the diff of 'max sample' and 'min sample' is 10% of the
dynamic range.
message is the signal found by the tuner encoded as a long integer.
The radar object pick up the basic Squitter objects from the message queue (msgQ), decodes it further
(using Squitter.py) and store the resulting Squitter object into a 'blip dictionary' using the ICAO information as key.
The blip dictionary have the following format
{ICAO24: [{'timestamp': ts, 'cnt': n, 'msg': message}, ...]}
'timestamp' is when the message was seen,
'cnt' is how many times a message have been seen,
'msg' is the decoded Squitter object
The blip dictionary is in turn read recurrently by a separate thread for display.
To collect some (rather pointless) statistics using call signs counts a smple FlightDB class exists.
Call sign statistics is stored into a json structure which is stored recurrently on file (file name is configurable in
the config file, if "flight db name" is "" this function is not used). The database is dumped to file ever 10 min,
before dumping the current file is copied to a backup-file. The backup file is used when the radar is re-staring
and we get an exception trying to read the current database file (the file is probably corrupt).
A simple tool to dump the content of the file exists (flight_db_tool.py), A client can ask for this information using
"GET FLIGHT_DB STR" (implemented in server.py).
"""
class TextDisplay:
"""
This implements a table text display for Squitter messages using curses
Methods are:
add: add a new message + timestamp + count into the msgQ dictionary using ICAO address as key
The methods also traverse the msqQ objects to propagate message items to the latest on
start: initialize the window environment
update_screen: format and add strings to the window form the msgQ
close: close the window environment in an orderly manner
"""
def __init__(self):
self.msgQ = {}
# header is on the format [Column, String]
self.header_text = [[0, "ICAO"], [8, "Mode"], [14, "Sqwk"], [20, "Flight"], [29, "Alt"], [36, "Spd"],
[41, "Hdg"], [48, "Lat"], [57, "Long"], [64, "Sig%"], [70, "Msgs"], [77, "Ti "]]
self.last_pos = self.header_text[-1][0] + len(self.header_text[-1][1])
self.header_underline = "-" * self.last_pos
self.header_spinner = "|/-\\"
self.update_cnt = 0
self.win = None
self.max_row = 0
self.saved_cur = None
curses.wrapper(self.start)
def start(self, screen):
self.win = screen
self.saved_cur = curses.curs_set(0)
self.max_row = self.win.getmaxyx()[0]
curses.init_pair(1, curses.COLOR_RED, curses.COLOR_BLACK)
curses.init_pair(2, curses.COLOR_YELLOW, curses.COLOR_BLACK)
curses.init_pair(3, curses.COLOR_GREEN, curses.COLOR_BLACK)
curses.init_pair(4, curses.COLOR_BLUE, curses.COLOR_BLACK)
self.update_screen()
def update_screen(self):
self.win.clear()
for ind in self.header_text:
self.win.addstr(0, ind[0], ind[1])
self.win.addch(0, self.last_pos, self.header_spinner[self.update_cnt % 4])
self.win.addstr(1, 0, self.header_underline)
self.win.clrtobot()
self.update_cnt += 1
row = 2
for key in self.msgQ.keys():
msg = self.msgQ[key]['msg']
self.win.addstr(row, self.header_text[0][0], msg['ICAO24'], curses.color_pair(3))
self.win.addstr(row, self.header_text[1][0], msg['downlink_format'], curses.color_pair(3))
self.win.addstr(row, self.header_text[2][0], msg['squawk'], curses.color_pair(3))
self.win.addstr(row, self.header_text[3][0], msg['call_sign'], curses.color_pair(3))
self.win.addstr(row, self.header_text[4][0], msg['altitude'], curses.color_pair(3))
self.win.addstr(row, self.header_text[5][0], msg['velocity'], curses.color_pair(3))
self.win.addstr(row, self.header_text[6][0], msg['heading'], curses.color_pair(3))
self.win.addstr(row, self.header_text[7][0], msg['latitude'], curses.color_pair(3))
self.win.addstr(row, self.header_text[8][0], msg['longitude'], curses.color_pair(3))
self.win.addstr(row, self.header_text[9][0], msg['signal_strength'], curses.color_pair(3))
self.win.addstr(row, self.header_text[10][0], self.msgQ[key]['msg_count'], curses.color_pair(2))
self.win.addstr(row, self.header_text[11][0], self.msgQ[key]['timestamp'], curses.color_pair(1))
row += 1
self.win.refresh()
def clear_queue(self):
self.msgQ = {}
def add(self, ts, msg, cnt):
# msgQ = {'abc': {'msg': squitter, 'timestamp': '0', 'msg_count': '1'}, 'def':...
icao = msg['ICAO24']
if icao in self.msgQ:
self.msgQ[icao]['msg'].update(msg)
else:
self.msgQ[icao] = {}
self.msgQ[icao]['msg'] = msg
self.msgQ[icao]['timestamp'] = str(int(time.time() - ts))
self.msgQ[icao]['msg_count'] = str(cnt)
if len(self.msgQ) == self.max_row - 2:
del self.msgQ[icao]
def close(self):
curses.curs_set(self.saved_cur)
curses.endwin()
class FlightDB:
"""
The FlightDB class is a simple persistent storage of flight data.
It stores data int a json file and sends the flights to the client
"""
def __init__(self, loc):
self.logger = logging.getLogger('spots.FlightDB')
self.lock = threading.Lock()
location = os.path.expanduser(loc)
self.loc = loc
self.loc_bck = self.loc + ".1"
init_db = {'version': basic.ADSB.VERSION,
'start_date': basic.statistics['start_time_string'],
'total_cnt': 0,
'flights': {}}
if os.path.exists(location):
try:
self.db = simplejson.load(open(self.loc, 'rb'))
except simplejson.JSONDecodeError:
try:
self.logger.info("Init, DB file corrupt, using backup")
# Current file is corrupt, try to fallback to backup file
self.db = simplejson.load(open(self.loc_bck, 'rb'))
except simplejson.JSONDecodeError:
self.logger.info("Init, DB file and backup corrupt, need to reinitialize")
# No joy, give up and re-initialize
self.db = init_db
else:
self.db = init_db
self.dump()
def add(self, flight):
self.lock.acquire()
if flight in self.db['flights']:
self.db['flights'][flight] += 1
else:
self.db['flights'][flight] = 1
self.db['total_cnt'] += 1
self.lock.release()
def get_head(self):
self.lock.acquire()
res = {k: v for k, v in self.db.iteritems() if k in ('version', 'start_date', 'total_cnt')}
self.lock.release()
return res
def get_flights(self):
# Return a sorted dictionary of flights, descending order
# See https://stackoverflow.com/questions/613183/how-do-i-sort-a-dictionary-by-value
self.lock.acquire()
res = {'flights': sorted(self.db['flights'].items(), key=operator.itemgetter(1), reverse=True)}
self.lock.release()
return res
def dump(self):
shutil.copy2(self.loc, self.loc_bck) # Make a backup
self.lock.acquire()
with open(self.loc, 'wt') as f:
simplejson.dump(self.db, f, skipkeys=True, indent=4*' ')
self.lock.release()
class Radar(basic.ADSB, threading.Thread):
"""
The Radar class is where squitter messages are stored and processed.
It implements a Queue (msgQ) where parsed squitter messages from the tuner are stored for processing.
Radar run through its own thread where the main loop (run) gets messages from msQ
The tuner_read method are a callback used by the tuner, this method is executed from the Tuner thread.
As messages are processed from the Queue they are stored into a dictionary (blips) using the ICAO address as index.
Messages in the dictionary have a timestamp and messages older than MAX_MSG_LIFETIME (seconds) are removed.
When messages are refreshed the timestamp are updated.
The radar use a Text User Interface (TUI) to display information to the end user.
"""
def __init__(self):
threading.Thread.__init__(self, name="Radar")
self.lock = threading.Lock()
self.finished = threading.Event()
self.msgQ = Queue.Queue() # msgQ is where we store messages that the tuner have detected, unlimited queue size
self.blips = {} # blips is a dictionary where we keep the recently seen messages
self.screen = None
self.daemon = True # This is a daemon thread
self.logger = logging.getLogger('spots.Radar')
if basic.ADSB.cfg_use_flight_db:
# We create a simple persistent storage for counting of flights
self.flight_db = FlightDB(basic.ADSB.cfg_flight_db_name)
self.flight_db.dump()
self.flight_timer = basic.RepeatTimer(10 * 60, self._dump_flight_db, "Radar flight DB timer")
self.flight_timer.start()
self.blip_timer = basic.RepeatTimer(1, self._scan_blips, "Radar blip timer")
self.stat_timer = basic.RepeatTimer(3600, self._show_stats, "Radar stat timer")
self.blip_timer.start()
self.stat_timer.start()
def _dump_flight_db(self):
if basic.ADSB.cfg_use_flight_db:
# Save to persistent storage, flights and statistics
self.flight_db.dump()
def _show_stats(self):
self.logger.info("Dumping statistics to file")
self.logger.info(str(basic.statistics))
def get_flight_db(self):
if basic.ADSB.cfg_use_flight_db:
fl = self.flight_db.get_head()
fl.update(self.flight_db.get_flights())
else:
fl = {}
return fl
@staticmethod
def get_statistics():
return basic.statistics.data
def get_blips_serialized(self):
# blips_series: {{'count': x, 'timestamp': y, 'altitude': 0, 'longitude': 13.755, ...},
# {'count': y, ...}}
result = []
self.lock.acquire()
for key in self.blips.keys():
elem = {'count': self.blips[key]['count'],
'timestamp': str(int(time.time() - self.blips[key]['timestamp']))}
for msg_key in self.blips[key]['msg']:
elem.update({msg_key: self.blips[key]['msg'][msg_key]})
result.append(elem)
self.lock.release()
return result
def _scan_blips(self):
"""
Scan the blip dictionary and add to the screen (if used) or print the item if screen not used
Note that the access to the blip dictionary is locked to prevent the producer thread to modify it
"""
self.lock.acquire()
self._remove_old_blips()
for key in self.blips.keys():
if self.cfg_use_text_display:
self.screen.add(self.blips[key]['timestamp'], self.blips[key]['msg'], self.blips[key]['count'])
else:
print self.blips[key]['msg']
self.lock.release()
if self.cfg_use_text_display:
self.screen.update_screen()
self.screen.clear_queue()
def _remove_old_blips(self):
"""
Remove blips that has a timestamp older than specified level (normally 60 secs)
"""
for key in self.blips.keys():
if (time.time() - self.blips[key]['timestamp']) >= self.cfg_max_blip_ttl:
del self.blips[key]
def _blip_add(self, msg):
"""
Decdode and add msg to the blips dictionary with a timestamp using ICAO address as key.
If an entry on the ICAO address exists, update the element.
Note that a lock is needed before the blip dictionary is modified to avoid confusing the reader thread
"""
msg.decode()
icao = msg['ICAO24']
self.lock.acquire()
# blip: {ICAO24: {'timestamp': ts, 'count': n, 'msg': msg},...}
if icao in self.blips:
self.blips[icao]['msg'].update(msg)
self.blips[icao]['timestamp'] = time.time()
self.blips[icao]['count'] += 1
else:
self.blips[icao] = {'msg': msg, 'timestamp': time.time(), 'count': 1}
# if not self.blips[icao]['msg'].decodeCPR_relative():
# self.blips[icao]['msg'].decodeCPR()
self.blips[icao]['msg'].decodeCPR()
# Logic, decodeCPR returns True if
# 1. It successfully decodes one pair of odd+even position messages, not separated by more than 10 seconds
# 2. Give that condition 1) above is fulfilled
#if self.blips[icao]['msg'].decodeCPR():
# self.blips[icao]['msg'].decodeCPR_relative()
if basic.ADSB.cfg_use_flight_db and msg['call_sign'] != "":
self.flight_db.add(msg['call_sign'])
self.lock.release()
if self.cfg_verbose_logging:
self.logger.info("{}".format(str(msg)))
def _blip_exist(self, msg):
return msg['ICAO24'] in self.blips
def run(self):
"""
Main loop for the Radar object, here is were objects from the tuner is retrieved and checked before
adding to the blip dictionary.
Note that the crc is checked here versus different type of messages.
Messages are added to the blip dictionary if:
1. The crc is ok
2. If crc is not ok, check if the the ICAO address exists in blip dictionary and add
a. if msg is of type 11 and crc sum is less than 80
b. we try setting the ICAO address to the crc sum for other messages, for these messages the
ICAO address is xor'ed with the crc
"""
self.logger.info("Radar running")
if self.cfg_use_text_display:
self.screen = TextDisplay()
while not self.finished.is_set():
msg = self.msgQ.get()
if msg.get_downlink_format() == self.DF_ALL_CALL_REPLY_11:
if msg.crc_ok:
self._blip_add(msg)
else:
if basic.ADSB.crc_2_int(msg.crc_sum) < 80 and self._blip_exist(msg):
msg.crc_ok = True
self._blip_add(msg)
elif msg.get_downlink_format() == self.DF_ADSB_MSG_17 and msg.crc_ok:
self._blip_add(msg)
elif msg.get_downlink_format() == self.DF_EXTENDED_SQUITTER_18 and msg.crc_ok:
self._blip_add(msg)
else: # All other DF have CRC xor'ed with ICAO address
if msg.crc_ok:
self._blip_add(msg)
else:
# we use the crc_sum for ICAO, this is tested in the decode method of Squitter
msg['ICAO24'] = msg.crc_sum
msg.crc_ok = True if self._blip_exist(msg) else False
if msg.crc_ok:
self._blip_add(msg)
self.logger.info("Radar stopping")
def _die(self):
self.logger.info("Radar dying")
if basic.ADSB.cfg_use_flight_db:
self._dump_flight_db()
self.flight_timer.cancel()
self.finished.set()
self.blip_timer.cancel()
self.stat_timer.cancel()
if self.cfg_use_text_display:
self.screen.close()
basic.statistics.dump()
def tuner_read(self, msgs, stop=False):
"""
Callback function used by the tuner thread (which is the main thread)
Creates a new Squitter object and do some basic parsing, add to the msgQ
"""
if stop:
self._die()
else:
for m in msgs:
sq = squitter.Squitter()
sq.parse(m)
if sq.msg != 0:
self.msgQ.put(sq)
def run_Radar():
print "Spots version {}".format(basic.ADSB.VERSION)
sys.stderr = open("spots.err", 'w')
logger = logging.getLogger('spots')
logger.setLevel(logging.DEBUG)
fh = logging.handlers.RotatingFileHandler(filename=basic.ADSB.cfg_log_file,
mode='w',
maxBytes=basic.ADSB.cfg_log_max_bytes,
backupCount=basic.ADSB.cfg_log_backup_count)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - (%(threadName)-10s) - %(message)s')
fh.setFormatter(formatter)
logger.addHandler(fh)
fh.setLevel(logging.DEBUG)
logger.info("Spots version: {}".format(basic.ADSB.VERSION))
if basic.ADSB.cfg_read_from_file:
tuner1090 = tuner.Tuner(filename=basic.ADSB.cfg_file_name)
else:
tuner1090 = tuner.Tuner()
tuner1090.start()
radar = Radar()
radar.start()
host = basic.ADSB.cfg_server_address
port = basic.ADSB.cfg_server_port
spots_server = server.SpotsServer((host, port), radar)
spots_server.start()
logger.info("Spots message server running, listening on {}:{}".format(host, port))
tuner1090.read(radar.tuner_read) # This is the main loop and main thread
spots_server.die()
if __name__ == '__main__':
run_Radar()