232 lines
8.8 KiB
Python
232 lines
8.8 KiB
Python
from owrx.kiss import KissDeframer
|
|
from owrx.map import Map, LatLngLocation
|
|
from owrx.bands import Bandplan
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def decodeBase91(input):
|
|
base = decodeBase91(input[:-1]) * 91 if len(input) > 1 else 0
|
|
return base + (ord(input[-1]) - 33)
|
|
|
|
|
|
class Ax25Parser(object):
|
|
def parse(self, ax25frame):
|
|
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]
|
|
|
|
return {
|
|
"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)],
|
|
"data": ax25frame[control_pid + 2 :],
|
|
}
|
|
|
|
def extractCallsign(self, input):
|
|
cs = bytes([b >> 1 for b in input[0:6]]).decode("us-ascii").strip()
|
|
ssid = (input[6] & 0b00011110) >> 1
|
|
if ssid > 0:
|
|
return "{callsign}-{ssid}".format(callsign=cs, ssid=ssid)
|
|
else:
|
|
return cs
|
|
|
|
|
|
class AprsParser(object):
|
|
def __init__(self, handler):
|
|
self.ax25parser = Ax25Parser()
|
|
self.deframer = KissDeframer()
|
|
self.dial_freq = None
|
|
self.band = None
|
|
self.handler = handler
|
|
|
|
def setDialFrequency(self, freq):
|
|
self.dial_freq = freq
|
|
self.band = Bandplan.getSharedInstance().findBand(freq)
|
|
|
|
def parse(self, raw):
|
|
for frame in self.deframer.parse(raw):
|
|
try:
|
|
data = self.ax25parser.parse(frame)
|
|
|
|
# TODO how can we tell if this is an APRS frame at all?
|
|
aprsData = self.parseAprsData(data)
|
|
|
|
logger.debug(aprsData)
|
|
if "lat" in aprsData and "lon" in aprsData:
|
|
loc = LatLngLocation(
|
|
aprsData["lat"], aprsData["lon"], aprsData["comment"] if "comment" in aprsData else None
|
|
)
|
|
Map.getSharedInstance().updateLocation(data["source"], loc, "APRS", self.band)
|
|
|
|
self.handler.write_aprs_data(aprsData)
|
|
except Exception:
|
|
logger.exception("exception while parsing aprs 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):
|
|
return {
|
|
"lat": 90 - decodeBase91(raw[1:5]) / 380926,
|
|
"lon": -180 + decodeBase91(raw[5:9]) / 190463,
|
|
"symbol": raw[9],
|
|
}
|
|
|
|
def parseAprsData(self, data):
|
|
information = data["data"]
|
|
|
|
# forward some of the ax25 data
|
|
aprsData = {"source": data["source"], "destination": data["destination"], "path": data["path"]}
|
|
|
|
if information[0] == 0x1C or information[0] == 0x60:
|
|
aprsData.update(MicEParser().parse(data))
|
|
return aprsData
|
|
|
|
information = information.decode("us-ascii")
|
|
logger.debug(information)
|
|
|
|
if information[0] == "!" or information[0] == "=":
|
|
# position without timestamp
|
|
aprsData.update(self.parseRegularAprsData(information[1:]))
|
|
elif information[0] == "/" or information[0] == "@":
|
|
# position with timestamp
|
|
# TODO parse timestamp
|
|
aprsData.update(self.parseRegularAprsData(information[8:]))
|
|
|
|
return aprsData
|
|
|
|
def parseRegularAprsData(self, information):
|
|
if self.hasCompressedCoordinatesx(information):
|
|
aprsData = self.parseCompressedCoordinates(information[0:10])
|
|
aprsData["type"] = "compressed"
|
|
aprsData["comment"] = information[10:]
|
|
else:
|
|
aprsData = self.parseUncompressedCoordinates(information[0:19])
|
|
aprsData["type"] = "regular"
|
|
aprsData["comment"] = information[19:]
|
|
return aprsData
|
|
|
|
|
|
class MicEParser(object):
|
|
def extractNumber(self, 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(self, input):
|
|
base = self.listToNumber(input[:-1]) * 10 if len(input) > 1 else 0
|
|
return base + input[-1]
|
|
|
|
def extractAltitude(self, comment):
|
|
if len(comment) < 4 or comment[3] != "}":
|
|
return (comment, None)
|
|
return comment[4:], decodeBase91(comment[:3]) - 10000
|
|
|
|
def extractDevice(self, comment):
|
|
if len(comment) > 0:
|
|
if comment[0] == ">":
|
|
if len(comment) > 1:
|
|
if comment[-1] == "=":
|
|
return comment[1:-1], {"manufacturer": "Kenwood", "device": "TH-D72"}
|
|
if comment[-1] == "^":
|
|
return comment[1:-1], {"manufacturer": "Kenwood", "device": "TH-D74"}
|
|
return comment[1:], {"manufacturer": "Kenwood", "device": "TH-D7A"}
|
|
if comment[0] == "]":
|
|
if len(comment) > 1 and comment[-1] == "=":
|
|
return comment[1:-1], {"manufacturer": "Kenwood", "device": "TM-D710"}
|
|
return comment[1:], {"manufacturer": "Kenwood", "device": "TM-D700"}
|
|
if len(comment) > 2 and (comment[0] == "`" or comment[0] == "'"):
|
|
if comment[-2] == "_":
|
|
devices = {
|
|
"b": "VX-8",
|
|
'"': "FTM-350",
|
|
"#": "VX-8G",
|
|
"$": "FT1D",
|
|
"%": "FTM-400DR",
|
|
")": "FTM-100D",
|
|
"(": "FT2D",
|
|
"0": "FT3D",
|
|
}
|
|
return comment[1:-2], {"manufacturer": "Yaesu", "device": devices.get(comment[-1], "Unknown")}
|
|
if comment[-2:] == " X":
|
|
return comment[1:-2], {"manufacturer": "SainSonic", "device": "AP510"}
|
|
if comment[-2] == "(":
|
|
devices = {"5": "D578UV", "8": "D878UV"}
|
|
return comment[1:-2], {"manufacturer": "Anytone", "device": devices.get(comment[-1], "Unknown")}
|
|
if comment[-2] == "|":
|
|
devices = {"3": "TinyTrack3", "4": "TinyTrack4"}
|
|
return comment[1:-2], {"manufacturer": "Byonics", "device": devices.get(comment[-1], "Unknown")}
|
|
if comment[-2:] == "^v":
|
|
return comment[1:-2], {"manufacturer": "HinzTec", "device": "anyfrog"}
|
|
if comment[-2] == ":":
|
|
devices = {"4": "P4dragon DR-7400 modem", "8": "P4dragon DR-7800 modem"}
|
|
return (
|
|
comment[1:-2],
|
|
{"manufacturer": "SCS GmbH & Co.", "device": devices.get(comment[-1], "Unknown")},
|
|
)
|
|
if comment[-2:] == "~v":
|
|
return comment[1:-2], {"manufacturer": "Other", "device": "Other"}
|
|
return comment[1:-2], None
|
|
return comment, None
|
|
|
|
def parse(self, data):
|
|
information = data["data"]
|
|
destination = data["destination"]
|
|
|
|
logger.debug(destination)
|
|
rawLatitude = [self.extractNumber(c) for c in destination[0:6]]
|
|
logger.debug(rawLatitude)
|
|
lat = self.listToNumber(rawLatitude[0:2]) + self.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
|
|
|
|
comment = information[9:].decode("us-ascii").strip()
|
|
(comment, altitude) = self.extractAltitude(comment)
|
|
|
|
(comment, device) = self.extractDevice(comment)
|
|
|
|
# altitude might be inside the device string, so repeat and choose one
|
|
(comment, insideAltitude) = self.extractAltitude(comment)
|
|
altitude = next((a for a in [altitude, insideAltitude] if a is not None), None)
|
|
|
|
return {"lat": lat, "lon": lon, "comment": comment, "altitude": altitude, "device": device, "type": "Mic-E"}
|