FGFS-4.1/pyModeS
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #47 from junzis/rtlsdr

Browse Source 
Add RTL-SDR support
This commit is contained in:
Junzi Sun
2019-08-23 14:47:50 +02:00
committed by GitHub
parent 7cb75ea8ca 4906a49e9c
commit 81d7cef6e8
6 changed files with 559 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
pyModeS/extra/rtlreader.py Normal file
View File

@@ -0,0 +1,155 @@
import numpy as np
import pyModeS as pms
from rtlsdr import RtlSdr
from threading import Thread
import time
modes_sample_rate = 2e6
modes_frequency = 1090e6
buffer_size = 1024 * 100
read_size = 1024 * 20
pbits = 8
fbits = 112
preamble = [1, 0, 1, 0, 0, 0, 0, 1, 0, 1, 0, 0, 0, 0, 0, 0]
th_amp = 0.2 # signal amplitude threshold for 0 and 1 bit
th_amp_diff = 0.8 # signal amplitude threshold difference between 0 and 1 bit
class RtlReader(Thread):
def __init__(self, debug=False):
super(RtlReader, self).__init__()
self.signal_buffer = []
self.debug = debug
self.sdr = RtlSdr()
self.sdr.sample_rate = modes_sample_rate
self.sdr.center_freq = modes_frequency
self.sdr.gain = "auto"
# sdr.freq_correction = 75
def _process_buffer(self):
messages = []
# signal_array = np.array(self.signal_buffer)
# pulses_array = np.where(np.array(self.signal_buffer) < th_amp, 0, 1)
# pulses = "".join(str(x) for x in pulses_array)
buffer_length = len(self.signal_buffer)
i = 0
while i < buffer_length:
if self.signal_buffer[i] < th_amp:
i += 1
continue
# if pulses[i : i + pbits * 2] == preamble:
if self._check_preamble(self.signal_buffer[i : i + pbits * 2]):
frame_start = i + pbits * 2
frame_end = i + pbits * 2 + (fbits + 1) * 2
frame_length = (fbits + 1) * 2
frame_pulses = self.signal_buffer[frame_start:frame_end]
msgbin = ""
for j in range(0, frame_length, 2):
p2 = frame_pulses[j : j + 2]
if len(p2) < 2:
break
if p2[0] < th_amp and p2[1] < th_amp:
break
elif p2[0] >= p2[1]:
c = "1"
elif p2[0] < p2[1]:
c = "0"
else:
msgbin = ""
break
msgbin += c
# advance i with a jump
i = frame_start + j
if len(msgbin) > 0:
msghex = pms.bin2hex(msgbin)
if self._check_msg(msghex):
messages.append([msghex, time.time()])
if self.debug:
self._debug_msg(msghex)
elif i > buffer_length - 500:
# save some for next process
break
else:
i += 1
# keep reminder of buffer for next iteration
self.signal_buffer = self.signal_buffer[i:]
return messages
def _check_preamble(self, pulses):
if len(pulses) != 16:
return False
for i in range(16):
if abs(pulses[i] - preamble[i]) > th_amp_diff:
return False
return True
def _check_msg(self, msg):
df = pms.df(msg)
msglen = len(msg)
if df == 17 and msglen == 28:
if pms.crc(msg) == 0:
return True
elif df in [20, 21] and msglen == 28:
return True
elif df in [4, 5, 11] and msglen == 14:
return True
def _debug_msg(self, msg):
df = pms.df(msg)
msglen = len(msg)
if df == 17 and msglen == 28:
print(msg, pms.icao(msg), pms.crc(msg))
elif df in [20, 21] and msglen == 28:
print(msg, pms.icao(msg))
elif df in [4, 5, 11] and msglen == 14:
print(msg, pms.icao(msg))
else:
# print("[*]", msg)
pass
def _read_callback(self, data, rtlsdr_obj):
# scaling signal (imporatant)
amp = np.absolute(data)
amp_norm = np.interp(amp, (amp.min(), amp.max()), (0, 1))
self.signal_buffer.extend(amp_norm.tolist())
if len(self.signal_buffer) >= buffer_size:
messages = self._process_buffer()
self.handle_messages(messages)
def handle_messages(self, messages):
"""re-implement this method to handle the messages"""
for msg, t in messages:
# print("%15.9f %s" % (t, msg))
pass
def stop(self):
self.sdr.cancel_read_async()
def run(self):
self.sdr.read_samples_async(self._read_callback, read_size)
# count = 1
# while count < 1000:
# count += 1
# data = self.sdr.read_samples(read_size)
# self._read_callback(data, None)
if __name__ == "__main__":
rtl = RtlReader()
rtl.debug = True
rtl.start()
rtl.join()

141
pyModeS/extra/tcpclient.py
View File

@@ -1,6 +1,6 @@
'''
"""
Stream beast raw data from a TCP server, convert to mode-s messages
'''
"""
from __future__ import print_function, division
import os
import sys
@@ -9,47 +9,44 @@ import time
import pyModeS as pms
from threading import Thread
import traceback
import zmq
if (sys.version_info > (3, 0)):
if sys.version_info > (3, 0):
PY_VERSION = 3
else:
PY_VERSION = 2
class BaseClient(Thread):
def __init__(self, host, port, rawtype):
def __init__(self, host, port, datatype):
Thread.__init__(self)
self.host = host
self.port = port
self.buffer = []
self.rawtype = rawtype
if self.rawtype not in ['avr', 'beast', 'skysense']:
print("rawtype must be either avr, beast or skysense")
self.socket = None
self.datatype = datatype
if self.datatype not in ["raw", "beast", "skysense"]:
print("datatype must be either raw, beast or skysense")
os._exit(1)
def connect(self):
while True:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(10) # 10 second timeout
s.connect((self.host, self.port))
print("Server connected - %s:%s" % (self.host, self.port))
print("collecting ADS-B messages...")
return s
except socket.error as err:
print("Socket connection error: %s. reconnecting..." % err)
time.sleep(3)
self.socket = zmq.Context().socket(zmq.STREAM)
self.socket.setsockopt(zmq.LINGER, 0)
self.socket.setsockopt(zmq.RCVTIMEO, 2000)
self.socket.connect("tcp://%s:%s" % (self.host, self.port))
def read_raw_buffer(self):
""" Read raw ADS-B data type.
def read_avr_buffer(self):
# -- testing --
# for b in self.buffer:
# print(chr(b), b)
# Append message with 0-9,A-F,a-f, until stop sign
String strats with "*" and ends with ";". For example:
*5d484ba898f8c6;
*8d400cd5990d7e9a10043e5e6da0;
*a0001498be800030aa0000c7a75f;
"""
messages = []
msg_stop = False
self.current_msg = ""
for b in self.buffer:
if b == 59:
msg_stop = True
@@ -57,9 +54,11 @@ class BaseClient(Thread):
messages.append([self.current_msg, ts])
if b == 42:
msg_stop = False
self.current_msg = ''
self.current_msg = ""
if (not msg_stop) and (48<=b<=57 or 65<=b<=70 or 97<=b<=102):
if (not msg_stop) and (
48 <= b <= 57 or 65 <= b <= 70 or 97 <= b <= 102
):
self.current_msg = self.current_msg + chr(b)
self.buffer = []
@@ -67,7 +66,8 @@ class BaseClient(Thread):
return messages
def read_beast_buffer(self):
'''
"""Handle mode-s beast data type.
<esc> "1" : 6 byte MLAT timestamp, 1 byte signal level,
2 byte Mode-AC
<esc> "2" : 6 byte MLAT timestamp, 1 byte signal level,
@@ -81,8 +81,7 @@ class BaseClient(Thread):
timestamp:
wiki.modesbeast.com/Radarcape:Firmware_Versions#The_GPS_timestamp
'''
"""
messages_mlat = []
msg = []
i = 0
@@ -91,16 +90,16 @@ class BaseClient(Thread):
# then, reset the self.buffer with the remainder
while i < len(self.buffer):
if (self.buffer[i:i+2] == [0x1a, 0x1a]):
msg.append(0x1a)
if self.buffer[i : i + 2] == [0x1A, 0x1A]:
msg.append(0x1A)
i += 1
elif (i == len(self.buffer) - 1) and (self.buffer[i] == 0x1a):
elif (i == len(self.buffer) - 1) and (self.buffer[i] == 0x1A):
# special case where the last bit is 0x1a
msg.append(0x1a)
elif self.buffer[i] == 0x1a:
msg.append(0x1A)
elif self.buffer[i] == 0x1A:
if i == len(self.buffer) - 1:
# special case where the last bit is 0x1a
msg.append(0x1a)
msg.append(0x1A)
elif len(msg) > 0:
messages_mlat.append(msg)
msg = []
@@ -112,12 +111,12 @@ class BaseClient(Thread):
if len(msg) > 0:
reminder = []
for i, m in enumerate(msg):
if (m == 0x1a) and (i < len(msg)-1):
if (m == 0x1A) and (i < len(msg) - 1):
# rewind 0x1a, except when it is at the last bit
reminder.extend([m, m])
else:
reminder.append(m)
self.buffer = [0x1a] + msg
self.buffer = [0x1A] + msg
else:
self.buffer = []
@@ -131,10 +130,10 @@ class BaseClient(Thread):
if msgtype == 0x32:
# Mode-S Short Message, 7 byte, 14-len hexstr
msg = ''.join('%02X' % i for i in mm[8:15])
msg = "".join("%02X" % i for i in mm[8:15])
elif msgtype == 0x33:
# Mode-S Long Message, 14 byte, 28-len hexstr
msg = ''.join('%02X' % i for i in mm[8:22])
msg = "".join("%02X" % i for i in mm[8:22])
else:
# Other message tupe
continue
@@ -215,25 +214,37 @@ class BaseClient(Thread):
messages = []
while len(self.buffer) > SS_MSGLENGTH:
i = 0
if self.buffer[i] == SS_STARTCHAR and self.buffer[i+SS_MSGLENGTH] == SS_STARTCHAR:
if (
self.buffer[i] == SS_STARTCHAR
and self.buffer[i + SS_MSGLENGTH] == SS_STARTCHAR
):
i += 1
if (self.buffer[i]>>7):
#Long message
payload = self.buffer[i:i+14]
if self.buffer[i] >> 7:
# Long message
payload = self.buffer[i : i + 14]
else:
#Short message
payload = self.buffer[i:i+7]
msg = ''.join('%02X' % j for j in payload)
i += 14 #Both message types use 14 bytes
tsbin = self.buffer[i:i+6]
sec = ( (tsbin[0] & 0x7f) << 10) | (tsbin[1] << 2 ) | (tsbin[2] >> 6)
nano = ( (tsbin[2] & 0x3f) << 24) | (tsbin[3] << 16) | (tsbin[4] << 8) | tsbin[5]
ts = sec + nano*1.0e-9
# Short message
payload = self.buffer[i : i + 7]
msg = "".join("%02X" % j for j in payload)
i += 14 # Both message types use 14 bytes
tsbin = self.buffer[i : i + 6]
sec = (
((tsbin[0] & 0x7F) << 10)
| (tsbin[1] << 2)
| (tsbin[2] >> 6)
)
nano = (
((tsbin[2] & 0x3F) << 24)
| (tsbin[3] << 16)
| (tsbin[4] << 8)
| tsbin[5]
)
ts = sec + nano * 1.0e-9
i += 6
#Signal and noise level - Don't care for now
# Signal and noise level - Don't care for now
i += 3
self.buffer = self.buffer[SS_MSGLENGTH:]
messages.append( [msg,ts] )
messages.append([msg, ts])
else:
self.buffer = self.buffer[1:]
return messages
@@ -244,11 +255,11 @@ class BaseClient(Thread):
print("%15.9f %s" % (t, msg))
def run(self):
sock = self.connect()
self.connect()
while True:
try:
received = sock.recv(1024)
received = [i for i in self.socket.recv(4096)]
if PY_VERSION == 2:
received = [ord(i) for i in received]
@@ -261,11 +272,11 @@ class BaseClient(Thread):
# continue
# -- Removed!! Cause delay in low data rate scenario --
if self.rawtype == 'beast':
if self.datatype == "beast":
messages = self.read_beast_buffer()
elif self.rawtype == 'avr':
messages = self.read_avr_buffer()
elif self.rawtype == 'skysense':
elif self.datatype == "raw":
messages = self.read_raw_buffer()
elif self.datatype == "skysense":
messages = self.read_skysense_buffer()
if not messages:
@@ -279,8 +290,8 @@ class BaseClient(Thread):
# Provides the user an option to supply the environment
# variable PYMODES_DEBUG to halt the execution
# for debugging purposes
debug_intent = os.environ.get('PYMODES_DEBUG', 'false')
if debug_intent.lower() == 'true':
debug_intent = os.environ.get("PYMODES_DEBUG", "false")
if debug_intent.lower() == "true":
traceback.print_exc()
sys.exit()
else:
@@ -292,11 +303,11 @@ class BaseClient(Thread):
print("Unexpected Error:", e)
if __name__ == '__main__':
if __name__ == "__main__":
# for testing purpose only
host = sys.argv[1]
port = int(sys.argv[2])
rawtype = sys.argv[3]
client = BaseClient(host=host, port=port, rawtype=rawtype)
datatype = sys.argv[3]
client = BaseClient(host=host, port=port, datatype=datatype)
client.daemon = True
client.run()

136
pyModeS/streamer/modeslive
View File

@@ -9,6 +9,7 @@ import curses
from threading import Lock
import pyModeS as pms
from pyModeS.extra.tcpclient import BaseClient
from pyModeS.extra.rtlreader import RtlReader
from pyModeS.streamer.stream import Stream
from pyModeS.streamer.screen import Screen
@@ -18,30 +19,77 @@ ADSB_TS = []
COMMB_MSG = []
COMMB_TS = []
support_rawtypes = ["raw", "beast", "skysense"]
parser = argparse.ArgumentParser()
parser.add_argument('--server', help='server address or IP', required=True)
parser.add_argument('--port', help='raw data port', required=True)
parser.add_argument('--rawtype', help='beast, avr or skysense', required=True)
parser.add_argument('--latlon', help='receiver position', nargs=2, metavar=('LAT', 'LON'), required=True)
parser.add_argument('--show-uncertainty', dest='uncertainty', help='display uncertaint values, default off', action='store_true', required=False, default=False)
parser.add_argument('--dumpto', help='folder to dump decoded output', required=False, default=None)
parser.add_argument(
"--source",
help='Choose data source, "rtlsdr" or "net"',
required=True,
default="net",
)
parser.add_argument(
"--connect",
help="Define server, port and data type. Supported data types are: %s"
% support_rawtypes,
nargs=3,
metavar=("SERVER", "PORT", "DATATYPE"),
default=None,
required=False,
)
parser.add_argument(
"--latlon",
help="Receiver latitude and longitude, needed for the surface position, default none",
nargs=2,
metavar=("LAT", "LON"),
default=None,
required=False,
)
parser.add_argument(
"--show-uncertainty",
dest="uncertainty",
help="Display uncertainty values, default off",
action="store_true",
required=False,
default=False,
)
parser.add_argument(
"--dumpto",
help="Folder to dump decoded output, default none",
required=False,
default=None,
)
args = parser.parse_args()
SERVER = args.server
PORT = int(args.port)
RAWTYPE = args.rawtype
LAT0 = float(args.latlon[0])
LON0 = float(args.latlon[1])
SOURCE = args.source
LATLON = args.latlon
UNCERTAINTY = args.uncertainty
DUMPTO = args.dumpto
if SOURCE == "rtlsdr":
pass
elif SOURCE == "net":
if args.connect is None:
print("Error: --connect argument must not be empty.")
else:
SERVER, PORT, DATATYPE = args.connect
if DATATYPE not in support_rawtypes:
print(
"Data type not supported, avaiable ones are %s"
% support_rawtypes
)
else:
print('Source must be "rtlsdr" or "net".')
sys.exit(1)
if DUMPTO is not None:
# append to current folder except root is given
if DUMPTO[0] != '/':
DUMPTO = os.getcwd() + '/' + DUMPTO
if DUMPTO[0] != "/":
DUMPTO = os.getcwd() + "/" + DUMPTO
if not os.path.isdir(DUMPTO):
print('Error: dump folder (%s) does not exist' % DUMPTO)
print("Error: dump folder (%s) does not exist" % DUMPTO)
sys.exit(1)
@@ -56,7 +104,7 @@ class ModesClient(BaseClient):
local_buffer_ehs_ts = []
for msg, t in messages:
if len(msg) < 28: # only process long messages
if len(msg) < 28: # only process long messages
continue
df = pms.df(msg)
@@ -70,7 +118,6 @@ class ModesClient(BaseClient):
else:
continue
LOCK.acquire()
ADSB_MSG.extend(local_buffer_adsb_msg)
ADSB_TS.extend(local_buffer_adsb_ts)
@@ -79,14 +126,57 @@ class ModesClient(BaseClient):
LOCK.release()
class ModesRtlReader(RtlReader):
"""docstring for ModesRtlReader."""
def __init__(self):
super(ModesRtlReader, self).__init__()
def handle_messages(self, messages):
local_buffer_adsb_msg = []
local_buffer_adsb_ts = []
local_buffer_ehs_msg = []
local_buffer_ehs_ts = []
for msg, t in messages:
if len(msg) < 28: # only process long messages
continue
df = pms.df(msg)
if df == 17 or df == 18:
local_buffer_adsb_msg.append(msg)
local_buffer_adsb_ts.append(t)
elif df == 20 or df == 21:
local_buffer_ehs_msg.append(msg)
local_buffer_ehs_ts.append(t)
else:
continue
LOCK.acquire()
ADSB_MSG.extend(local_buffer_adsb_msg)
ADSB_TS.extend(local_buffer_adsb_ts)
COMMB_MSG.extend(local_buffer_ehs_msg)
COMMB_TS.extend(local_buffer_ehs_ts)
# print(len(ADSB_MSG))
# print(len(COMMB_MSG))
LOCK.release()
# redirect all stdout to null, avoiding messing up with the screen
sys.stdout = open(os.devnull, 'w')
sys.stdout = open(os.devnull, "w")
client = ModesClient(host=SERVER, port=PORT, rawtype=RAWTYPE)
client.daemon = True
client.start()
if SOURCE == "net":
client = ModesClient(host=SERVER, port=PORT, rawtype=DATATYPE)
client.daemon = True
client.start()
elif SOURCE == "rtlsdr":
rtl = ModesRtlReader()
# rtl.debug = True
rtl.daemon = True
rtl.start()
stream = Stream(lat0=LAT0, lon0=LON0, dumpto=DUMPTO)
stream = Stream(latlon=LATLON, dumpto=DUMPTO)
try:
screen = Screen(uncertainty=UNCERTAINTY)
@@ -94,7 +184,7 @@ try:
screen.start()
while True:
if len(ADSB_MSG) > 200:
if len(ADSB_MSG) > 1:
LOCK.acquire()
stream.process_raw(ADSB_TS, ADSB_MSG, COMMB_TS, COMMB_MSG)
ADSB_MSG = []
@@ -107,7 +197,7 @@ try:
try:
screen.update_data(acs)
screen.update()
time.sleep(0.02)
time.sleep(1)
except KeyboardInterrupt:
raise
except:

104
pyModeS/streamer/screen.py
View File

@@ -6,38 +6,39 @@ import time
from threading import Thread
COLUMNS = [
('call', 10),
('lat', 10),
('lon', 10),
('alt', 7),
('gs', 5),
('tas', 5),
('ias', 5),
('mach', 7),
('roc', 7),
('trk', 10),
('hdg', 10),
('live', 6),
("call", 10),
("lat", 10),
("lon", 10),
("alt", 7),
("gs", 5),
("tas", 5),
("ias", 5),
("mach", 7),
("roc", 7),
("trk", 10),
("hdg", 10),
("live", 6),
]
UNCERTAINTY_COLUMNS = [
('|', 5),
('ver', 4),
('HPL', 5),
('RCu', 5),
('RCv', 5),
('HVE', 5),
('VVE', 5),
('Rc', 4),
('VPL', 5),
('EPU', 5),
('VEPU', 6),
('HFOMr', 7),
('VFOMr', 7),
('PE_RCu', 8),
('PE_VPL', 8),
("|", 5),
("ver", 4),
("HPL", 5),
("RCu", 5),
("RCv", 5),
("HVE", 5),
("VVE", 5),
("Rc", 4),
("VPL", 5),
("EPU", 5),
("VEPU", 6),
("HFOMr", 7),
("VFOMr", 7),
("PE_RCu", 8),
("PE_VPL", 8),
]
class Screen(Thread):
def __init__(self, uncertainty=False):
Thread.__init__(self)
@@ -55,7 +56,6 @@ class Screen(Thread):
if uncertainty:
self.columns.extend(UNCERTAINTY_COLUMNS)
def reset_cursor_pos(self):
self.screen.move(self.y, self.x)
@@ -64,7 +64,12 @@ class Screen(Thread):
def draw_frame(self):
self.screen.border(0)
self.screen.addstr(0, 2, "Online aircraft [%d] ('Ctrl+C' to exit, 'Enter' to lock one)" % len(self.acs))
self.screen.addstr(
0,
2,
"Online aircraft [%d] ('Ctrl+C' to exit, 'Enter' to lock one)"
% len(self.acs),
)
def update(self):
if len(self.acs) == 0:
@@ -81,21 +86,20 @@ class Screen(Thread):
row = 1
header = ' icao'
header = " icao"
for c, cw in self.columns:
header += (cw-len(c))*' ' + c
header += (cw - len(c)) * " " + c
# fill end with spaces
header += (self.scr_w - 2 - len(header)) * ' '
header += (self.scr_w - 2 - len(header)) * " "
if len(header) > self.scr_w - 2:
header = header[:self.scr_w-3] + '>'
header = header[: self.scr_w - 3] + ">"
self.screen.addstr(row, 1, header)
row +=1
self.screen.addstr(row, 1, '-'*(self.scr_w-2))
row += 1
self.screen.addstr(row, 1, "-" * (self.scr_w - 2))
icaos = np.array(list(self.acs.keys()))
icaos = np.sort(icaos)
@@ -105,10 +109,10 @@ class Screen(Thread):
idx = row + self.offset - 3
if idx > len(icaos) - 1:
line = ' '*(self.scr_w-2)
line = " " * (self.scr_w - 2)
else:
line = ''
line = ""
icao = icaos[idx]
ac = self.acs[icao]
@@ -116,22 +120,22 @@ class Screen(Thread):
line += icao
for c, cw in self.columns:
if c=='|':
val = '|'
elif c=='live':
val = str(int(time.time() - ac[c]))+'s'
if c == "|":
val = "|"
elif c == "live":
val = str(ac[c] - int(time.time())) + "s"
elif ac[c] is None:
val = ''
val = ""
else:
val = ac[c]
val_str = str(val)
line += (cw-len(val_str))*' ' + val_str
line += (cw - len(val_str)) * " " + val_str
# fill end with spaces
line += (self.scr_w - 2 - len(line)) * ' '
line += (self.scr_w - 2 - len(line)) * " "
if len(line) > self.scr_w - 2:
line = line[:self.scr_w-3] + '>'
line = line[: self.scr_w - 3] + ">"
if (icao is not None) and (self.lock_icao == icao):
self.screen.addstr(row, 1, line, curses.A_STANDOUT)
@@ -140,11 +144,13 @@ class Screen(Thread):
else:
self.screen.addstr(row, 1, line)
self.screen.addstr(self.scr_h-3, 1, '-'*(self.scr_w-2))
self.screen.addstr(self.scr_h - 3, 1, "-" * (self.scr_w - 2))
total_page = len(icaos) // (self.scr_h - 4) + 1
current_page = self.offset // (self.scr_h - 4) + 1
self.screen.addstr(self.scr_h-2, 1, '(%d / %d)' % (current_page, total_page))
self.screen.addstr(
self.scr_h - 2, 1, "(%d / %d)" % (current_page, total_page)
)
self.reset_cursor_pos()
@@ -168,7 +174,7 @@ class Screen(Thread):
self.offset = offset_intent
else:
self.offset = 0
elif c == curses.KEY_DOWN :
elif c == curses.KEY_DOWN:
y_intent = self.y + 1
if y_intent < self.scr_h - 3:
self.y = y_intent

229
pyModeS/streamer/stream.py
View File

@@ -5,24 +5,27 @@ import datetime
import csv
import pyModeS as pms
class Stream():
def __init__(self, lat0, lon0, dumpto=None):
class Stream:
def __init__(self, latlon=None, dumpto=None):
self.acs = dict()
self.lat0 = lat0
self.lon0 = lon0
if latlon is not None:
self.lat0 = float(latlon[0])
self.lon0 = float(latlon[1])
else:
self.lat0 = None
self.lon0 = None
self.t = 0
self.cache_timeout = 60 # seconds
self.cache_timeout = 60 # seconds
if dumpto is not None and os.path.isdir(dumpto):
self.dumpto = dumpto
else:
self.dumpto = None
def process_raw(self, adsb_ts, adsb_msgs, commb_ts, commb_msgs, tnow=None):
"""process a chunk of adsb and commb messages recieved in the same
time period.
@@ -42,43 +45,43 @@ class Stream():
if icao not in self.acs:
self.acs[icao] = {
'live': None,
'call': None,
'lat': None,
'lon': None,
'alt': None,
'gs': None,
'trk': None,
'roc': None,
'tas': None,
'roll': None,
'rtrk': None,
'ias': None,
'mach': None,
'hdg': None,
'ver' : None,
'HPL' : None,
'RCu' : None,
'RCv' : None,
'HVE' : None,
'VVE' : None,
'Rc' : None,
'VPL' : None,
'EPU' : None,
'VEPU' : None,
'HFOMr' : None,
'VFOMr' : None,
'PE_RCu' : None,
'PE_VPL' : None,
"live": None,
"call": None,
"lat": None,
"lon": None,
"alt": None,
"gs": None,
"trk": None,
"roc": None,
"tas": None,
"roll": None,
"rtrk": None,
"ias": None,
"mach": None,
"hdg": None,
"ver": None,
"HPL": None,
"RCu": None,
"RCv": None,
"HVE": None,
"VVE": None,
"Rc": None,
"VPL": None,
"EPU": None,
"VEPU": None,
"HFOMr": None,
"VFOMr": None,
"PE_RCu": None,
"PE_VPL": None,
}
self.acs[icao]['t'] = t
self.acs[icao]['live'] = int(t)
self.acs[icao]["t"] = t
self.acs[icao]["live"] = int(t)
if 1 <= tc <= 4:
cs = pms.adsb.callsign(msg)
self.acs[icao]['call'] = cs
output_buffer.append([t, icao, 'cs', cs])
self.acs[icao]["call"] = cs
output_buffer.append([t, icao, "cs", cs])
if (5 <= tc <= 8) or (tc == 19):
vdata = pms.adsb.velocity(msg)
@@ -86,42 +89,47 @@ class Stream():
continue
spd, trk, roc, tag = vdata
if tag != 'GS':
if tag != "GS":
continue
if (spd is None) or (trk is None):
continue
self.acs[icao]['gs'] = spd
self.acs[icao]['trk'] = trk
self.acs[icao]['roc'] = roc
self.acs[icao]['tv'] = t
self.acs[icao]["gs"] = spd
self.acs[icao]["trk"] = trk
self.acs[icao]["roc"] = roc
self.acs[icao]["tv"] = t
output_buffer.append([t, icao, 'gs', spd])
output_buffer.append([t, icao, 'trk', trk])
output_buffer.append([t, icao, 'roc', roc])
output_buffer.append([t, icao, "gs", spd])
output_buffer.append([t, icao, "trk", trk])
output_buffer.append([t, icao, "roc", roc])
if (5 <= tc <= 18):
if 5 <= tc <= 18:
oe = pms.adsb.oe_flag(msg)
self.acs[icao][oe] = msg
self.acs[icao]['t'+str(oe)] = t
self.acs[icao]["t" + str(oe)] = t
if ('tpos' in self.acs[icao]) and (t - self.acs[icao]['tpos'] < 180):
if ("tpos" in self.acs[icao]) and (
t - self.acs[icao]["tpos"] < 180
):
# use single message decoding
rlat = self.acs[icao]['lat']
rlon = self.acs[icao]['lon']
rlat = self.acs[icao]["lat"]
rlon = self.acs[icao]["lon"]
latlon = pms.adsb.position_with_ref(msg, rlat, rlon)
elif ('t0' in self.acs[icao]) and ('t1' in self.acs[icao]) and \
(abs(self.acs[icao]['t0'] - self.acs[icao]['t1']) < 10):
elif (
("t0" in self.acs[icao])
and ("t1" in self.acs[icao])
and (abs(self.acs[icao]["t0"] - self.acs[icao]["t1"]) < 10)
):
# use multi message decoding
try:
latlon = pms.adsb.position(
self.acs[icao][0],
self.acs[icao][1],
self.acs[icao]['t0'],
self.acs[icao]['t1'],
self.lat0, self.lon0
)
self.acs[icao]["t0"],
self.acs[icao]["t1"],
self.lat0,
self.lon0,
)
except:
# mix of surface and airborne position message
continue
@@ -129,16 +137,16 @@ class Stream():
latlon = None
if latlon is not None:
self.acs[icao]['tpos'] = t
self.acs[icao]['lat'] = latlon[0]
self.acs[icao]['lon'] = latlon[1]
self.acs[icao]["tpos"] = t
self.acs[icao]["lat"] = latlon[0]
self.acs[icao]["lon"] = latlon[1]
alt = pms.adsb.altitude(msg)
self.acs[icao]['alt'] = alt
self.acs[icao]["alt"] = alt
output_buffer.append([t, icao, 'lat', latlon[0]])
output_buffer.append([t, icao, 'lon', latlon[1]])
output_buffer.append([t, icao, 'alt', alt])
output_buffer.append([t, icao, "lat", latlon[0]])
output_buffer.append([t, icao, "lon", latlon[1]])
output_buffer.append([t, icao, "alt", alt])
local_updated_acs_buffer.append(icao)
@@ -146,35 +154,42 @@ class Stream():
ac = self.acs[icao]
if 9 <= tc <= 18:
ac['nic_bc'] = pms.adsb.nic_b(msg)
ac["nic_bc"] = pms.adsb.nic_b(msg)
if (5 <= tc <= 8) or (9 <= tc <= 18) or (20 <= tc <= 22):
ac['HPL'], ac['RCu'], ac['RCv'] = pms.adsb.nuc_p(msg)
ac["HPL"], ac["RCu"], ac["RCv"] = pms.adsb.nuc_p(msg)
if (ac['ver'] == 1) and ('nic_s' in ac.keys()):
ac['Rc'], ac['VPL'] = pms.adsb.nic_v1(msg, ac['nic_s'])
elif (ac['ver'] == 2) and ('nic_a' in ac.keys()) and ('nic_bc' in ac.keys()):
ac['Rc'] = pms.adsb.nic_v2(msg, ac['nic_a'], ac['nic_bc'])
if (ac["ver"] == 1) and ("nic_s" in ac.keys()):
ac["Rc"], ac["VPL"] = pms.adsb.nic_v1(msg, ac["nic_s"])
elif (
(ac["ver"] == 2)
and ("nic_a" in ac.keys())
and ("nic_bc" in ac.keys())
):
ac["Rc"] = pms.adsb.nic_v2(msg, ac["nic_a"], ac["nic_bc"])
if tc == 19:
ac['HVE'], ac['VVE'] = pms.adsb.nuc_v(msg)
if ac['ver'] in [1, 2]:
ac['HFOMr'], ac['VFOMr'] = pms.adsb.nac_v(msg)
ac["HVE"], ac["VVE"] = pms.adsb.nuc_v(msg)
if ac["ver"] in [1, 2]:
ac["HFOMr"], ac["VFOMr"] = pms.adsb.nac_v(msg)
if tc == 29:
ac['PE_RCu'], ac['PE_VPL'], ac['base'] = pms.adsb.sil(msg, ac['ver'])
ac['EPU'], ac['VEPU'] = pms.adsb.nac_p(msg)
ac["PE_RCu"], ac["PE_VPL"], ac["base"] = pms.adsb.sil(
msg, ac["ver"]
)
ac["EPU"], ac["VEPU"] = pms.adsb.nac_p(msg)
if tc == 31:
ac['ver'] = pms.adsb.version(msg)
ac['EPU'], ac['VEPU'] = pms.adsb.nac_p(msg)
ac['PE_RCu'], ac['PE_VPL'], ac['sil_base'] = pms.adsb.sil(msg, ac['ver'])
if ac['ver'] == 1:
ac['nic_s'] = pms.adsb.nic_s(msg)
elif ac['ver'] == 2:
ac['nic_a'], ac['nic_bc'] = pms.adsb.nic_a_c(msg)
ac["ver"] = pms.adsb.version(msg)
ac["EPU"], ac["VEPU"] = pms.adsb.nac_p(msg)
ac["PE_RCu"], ac["PE_VPL"], ac["sil_base"] = pms.adsb.sil(
msg, ac["ver"]
)
if ac["ver"] == 1:
ac["nic_s"] = pms.adsb.nic_s(msg)
elif ac["ver"] == 2:
ac["nic_a"], ac["nic_bc"] = pms.adsb.nic_a_c(msg)
# process commb message
for t, msg in zip(commb_ts, commb_msgs):
@@ -183,32 +198,34 @@ class Stream():
if icao not in self.acs:
continue
self.acs[icao]["live"] = int(t)
bds = pms.bds.infer(msg)
if bds == 'BDS50':
if bds == "BDS50":
roll50 = pms.commb.roll50(msg)
trk50 = pms.commb.trk50(msg)
rtrk50 = pms.commb.rtrk50(msg)
gs50 = pms.commb.gs50(msg)
tas50 = pms.commb.tas50(msg)
self.acs[icao]['t50'] = t
self.acs[icao]["t50"] = t
if tas50:
self.acs[icao]['tas'] = tas50
output_buffer.append([t, icao, 'tas50', tas50])
self.acs[icao]["tas"] = tas50
output_buffer.append([t, icao, "tas50", tas50])
if roll50:
self.acs[icao]['roll'] = roll50
output_buffer.append([t, icao, 'roll50', roll50])
self.acs[icao]["roll"] = roll50
output_buffer.append([t, icao, "roll50", roll50])
if rtrk50:
self.acs[icao]['rtrk'] = rtrk50
output_buffer.append([t, icao, 'rtrk50', rtrk50])
self.acs[icao]["rtrk"] = rtrk50
output_buffer.append([t, icao, "rtrk50", rtrk50])
if trk50:
output_buffer.append([t, icao, 'trk50', trk50])
output_buffer.append([t, icao, "trk50", trk50])
if gs50:
output_buffer.append([t, icao, 'gs50', gs50])
output_buffer.append([t, icao, "gs50", gs50])
elif bds == 'BDS60':
elif bds == "BDS60":
ias60 = pms.commb.ias60(msg)
hdg60 = pms.commb.hdg60(msg)
mach60 = pms.commb.mach60(msg)
@@ -216,28 +233,28 @@ class Stream():
roc60ins = pms.commb.vr60ins(msg)
if ias60 or hdg60 or mach60:
self.acs[icao]['t60'] = t
self.acs[icao]["t60"] = t
if ias60:
self.acs[icao]['ias'] = ias60
self.acs[icao]["ias"] = ias60
if hdg60:
self.acs[icao]['hdg'] = hdg60
self.acs[icao]["hdg"] = hdg60
if mach60:
self.acs[icao]['mach'] = mach60
self.acs[icao]["mach"] = mach60
if roc60baro:
output_buffer.append([t, icao, 'roc60baro', roc60baro])
output_buffer.append([t, icao, "roc60baro", roc60baro])
if roc60ins:
output_buffer.append([t, icao, 'roc60ins', roc60ins])
output_buffer.append([t, icao, "roc60ins", roc60ins])
# clear up old data
for icao in list(self.acs.keys()):
if self.t - self.acs[icao]['live'] > self.cache_timeout:
if self.t - self.acs[icao]["live"] > self.cache_timeout:
del self.acs[icao]
continue
if self.dumpto is not None:
dh = str(datetime.datetime.now().strftime("%Y%m%d_%H"))
fn = self.dumpto + '/pymodes_dump_%s.csv' % dh
fn = self.dumpto + "/pymodes_dump_%s.csv" % dh
output_buffer.sort(key=lambda x: x[0])
with open(fn, "a") as f:
writer = csv.writer(f)
@@ -248,8 +265,4 @@ class Stream():
def get_aircraft(self):
"""all aircraft that are stored in memeory"""
acs = self.acs
icaos = list(acs.keys())
for icao in icaos:
if acs[icao]['lat'] is None:
acs.pop(icao)
return acs