restructure the code to have the parser sit where all the parsers sit
This commit is contained in:
Jakob Ketterl
201
owrx/kiss.py
201
owrx/kiss.py
@ -1,6 +1,5 @@
|
||||
import socket
|
||||
import time
|
||||
from owrx.map import Map, LatLngLocation
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -10,23 +9,6 @@ FESC = 0xDB
|
||||
TFEND = 0xDC
|
||||
TFESC = 0XDD
|
||||
|
||||
def group(a, *ns):
|
||||
for n in ns:
|
||||
a = [a[i:i+n] for i in range(0, len(a), n)]
|
||||
return a
|
||||
|
||||
def join(a, *cs):
|
||||
return [cs[0].join(join(t, *cs[1:])) for t in a] if cs else a
|
||||
|
||||
def hexdump(data):
|
||||
toHex = lambda c: '{:02X}'.format(c)
|
||||
toChr = lambda c: chr(c) if 32 <= c < 127 else '.'
|
||||
make = lambda f, *cs: join(group(list(map(f, data)), 8, 2), *cs)
|
||||
hs = make(toHex, ' ', ' ')
|
||||
cs = make(toChr, ' ', '')
|
||||
for i, (h, c) in enumerate(zip(hs, cs)):
|
||||
print ('{:010X}: {:48} {:16}'.format(i * 16, h, c))
|
||||
|
||||
|
||||
class KissClient(object):
|
||||
def __init__(self, port):
|
||||
@ -35,167 +17,32 @@ class KissClient(object):
|
||||
self.socket.connect(("localhost", port))
|
||||
|
||||
def read(self):
|
||||
buf = bytes()
|
||||
escaped = False
|
||||
while True:
|
||||
input = self.socket.recv(1)
|
||||
# EOF
|
||||
if len(input) == 0:
|
||||
return bytes()
|
||||
return self.socket.recv(1)
|
||||
|
||||
if input[0] == FESC:
|
||||
escaped = True
|
||||
elif escaped:
|
||||
if input[0] == TFEND:
|
||||
buf += [FEND]
|
||||
elif input[0] == TFESC:
|
||||
buf += [FESC]
|
||||
|
||||
class KissDeframer(object):
|
||||
def __init__(self):
|
||||
self.escaped = False
|
||||
self.buf = bytearray()
|
||||
|
||||
def parse(self, input):
|
||||
frames = []
|
||||
for b in input:
|
||||
if b == FESC:
|
||||
self.escaped = True
|
||||
elif self.escaped:
|
||||
if b == TFEND:
|
||||
self.buf.append(FEND)
|
||||
elif b == TFESC:
|
||||
self.buf.append(FESC)
|
||||
else:
|
||||
logger.warning("invalid escape char: %s", str(input[0]))
|
||||
escaped = False
|
||||
self.escaped = False
|
||||
elif input[0] == FEND:
|
||||
logger.debug("decoded frame: " + str(buf))
|
||||
if len(buf) > 0:
|
||||
try:
|
||||
return self.parseFrame(buf)
|
||||
except Exception:
|
||||
logger.exception("failed to decode packet data")
|
||||
return {}
|
||||
# data frames start with 0x00
|
||||
if len(self.buf) > 1 and self.buf[0] == 0x00:
|
||||
frames += [self.buf[1:]]
|
||||
self.buf = bytearray()
|
||||
else:
|
||||
buf += input
|
||||
|
||||
def parseFrame(self, frame):
|
||||
# data frames start with 0x00
|
||||
if frame[0] != 0x00:
|
||||
return {}
|
||||
ax25frame = frame[1:]
|
||||
control_pid = ax25frame.find(bytes([0x03, 0xf0]))
|
||||
if control_pid % 7 > 0:
|
||||
logger.warning("aprs packet framing error: control/pid position not aligned with 7-octet callsign data")
|
||||
|
||||
def chunks(l, n):
|
||||
"""Yield successive n-sized chunks from l."""
|
||||
for i in range(0, len(l), n):
|
||||
yield l[i:i + n]
|
||||
|
||||
information = ax25frame[control_pid+2:]
|
||||
|
||||
data = {
|
||||
"destination": self.extractCallsign(ax25frame[0:7]),
|
||||
"source": self.extractCallsign(ax25frame[7:14]),
|
||||
"path": [self.extractCallsign(c) for c in chunks(ax25frame[14:control_pid], 7)]
|
||||
}
|
||||
# TODO how can we tell if this is an APRS frame at all?
|
||||
aprsData = self.parseAprsData(data["destination"], information)
|
||||
data.update(aprsData)
|
||||
|
||||
logger.debug(data)
|
||||
if "lat" in data and "lon" in data:
|
||||
loc = LatLngLocation(data["lat"], data["lon"], data["comment"] if "comment" in data else None)
|
||||
Map.getSharedInstance().updateLocation(data["source"], loc, "APRS")
|
||||
return data
|
||||
|
||||
def hasCompressedCoordinatesx(self, raw):
|
||||
return raw[0] == "/" or raw[0] == "\\"
|
||||
|
||||
def parseUncompressedCoordinates(self, raw):
|
||||
lat = int(raw[0:2]) + float(raw[2:7]) / 60
|
||||
if raw[7] == "S":
|
||||
lat *= -1
|
||||
lon = int(raw[9:12]) + float(raw[12:17]) / 60
|
||||
if raw[17] == "W":
|
||||
lon *= -1
|
||||
return {
|
||||
"lat": lat,
|
||||
"lon": lon,
|
||||
"symbol": raw[18]
|
||||
}
|
||||
|
||||
def parseCompressedCoordinates(self, raw):
|
||||
def decodeBase91(input):
|
||||
base = decodeBase91(input[:-1]) * 91 if len(input) > 1 else 0
|
||||
return base + (ord(input[-1]) - 33)
|
||||
return {
|
||||
"lat": 90 - decodeBase91(raw[1:5]) / 380926,
|
||||
"lon": -180 + decodeBase91(raw[5:9]) / 190463,
|
||||
"symbol": raw[9]
|
||||
}
|
||||
|
||||
def parseMicEFrame(self, destination, information):
|
||||
def extractNumber(input):
|
||||
n = ord(input)
|
||||
if n >= ord("P"):
|
||||
return n - ord("P")
|
||||
if n >= ord("A"):
|
||||
return n - ord("A")
|
||||
return n - ord("0")
|
||||
|
||||
def listToNumber(input):
|
||||
base = listToNumber(input[:-1]) * 10 if len(input) > 1 else 0
|
||||
return base + input[-1]
|
||||
|
||||
logger.debug(destination)
|
||||
rawLatitude = [extractNumber(c) for c in destination[0:6]]
|
||||
logger.debug(rawLatitude)
|
||||
lat = listToNumber(rawLatitude[0:2]) + listToNumber(rawLatitude[2:6]) / 6000
|
||||
if ord(destination[3]) <= ord("9"):
|
||||
lat *= -1
|
||||
|
||||
logger.debug(lat)
|
||||
|
||||
logger.debug(information)
|
||||
lon = information[1] - 28
|
||||
if ord(destination[4]) >= ord("P"):
|
||||
lon += 100
|
||||
if 180 <= lon <= 189:
|
||||
lon -= 80
|
||||
if 190 <= lon <= 199:
|
||||
lon -= 190
|
||||
|
||||
minutes = information[2] - 28
|
||||
if minutes >= 60:
|
||||
minutes -= 60
|
||||
|
||||
lon += minutes / 60 + (information[3] - 28) / 6000
|
||||
|
||||
if ord(destination[5]) >= ord("P"):
|
||||
lon *= -1
|
||||
|
||||
return {
|
||||
"lat": lat,
|
||||
"lon": lon,
|
||||
"comment": information[9:].decode()
|
||||
}
|
||||
|
||||
def parseAprsData(self, destination, information):
|
||||
if information[0] == 0x1c or information[0] == 0x60:
|
||||
return self.parseMicEFrame(destination, information)
|
||||
|
||||
information = information.decode()
|
||||
logger.debug(information)
|
||||
|
||||
if information[0] == "!" or information[0] == "=":
|
||||
# position without timestamp
|
||||
information = information[1:]
|
||||
elif information[0] == "/" or information[0] == "@":
|
||||
# position with timestamp
|
||||
# TODO parse timestamp
|
||||
information = information[8:]
|
||||
else:
|
||||
return {}
|
||||
|
||||
if self.hasCompressedCoordinatesx(information):
|
||||
coords = self.parseCompressedCoordinates(information[0:10])
|
||||
coords["comment"] = information[10:]
|
||||
else:
|
||||
coords = self.parseUncompressedCoordinates(information[0:19])
|
||||
coords["comment"] = information[19:]
|
||||
return coords
|
||||
|
||||
def extractCallsign(self, input):
|
||||
cs = bytes([b >> 1 for b in input[0:6]]).decode().strip()
|
||||
ssid = (input[6] & 0b00011110) >> 1
|
||||
if ssid > 0:
|
||||
return "{callsign}-{ssid}".format(callsign=cs, ssid=ssid)
|
||||
else:
|
||||
return cs
|
||||
self.buf.append(b)
|
||||
return frames
|
||||
|
||||
Reference in New Issue
Block a user