openwebrx-clone/owrx/aprs.py
2019-09-23 03:15:24 +02:00

578 lines
22 KiB
Python

from owrx.kiss import KissDeframer
from owrx.map import Map, LatLngLocation
from owrx.bands import Bandplan
from owrx.metrics import Metrics, CounterMetric
from datetime import datetime, timezone
import re
import logging
logger = logging.getLogger(__name__)
# speed is in knots... convert to metric (km/h)
knotsToKilometers = 1.852
feetToMeters = 0.3048
milesToKilometers = 1.609344
inchesToMilimeters = 25.4
def fahrenheitToCelsius(f):
return (f - 32) * 5 / 9
# not sure what the correct encoding is. it seems TAPR has set utf-8 as a standard, but not everybody is following it.
encoding = "utf-8"
# regex for altitute in comment field
altitudeRegex = re.compile("(^.*)\\/A=([0-9]{6})(.*$)")
# regex for parsing third-party headers
thirdpartyeRegex = re.compile("^([a-zA-Z0-9-]+)>((([a-zA-Z0-9-]+\\*?,)*)([a-zA-Z0-9-]+\\*?)):(.*)$")
# regex for getting the message id out of message
messageIdRegex = re.compile("^(.*){([0-9]{1,5})$")
def decodeBase91(input):
base = decodeBase91(input[:-1]) * 91 if len(input) > 1 else 0
return base + (ord(input[-1]) - 33)
def getSymbolData(symbol, table):
return {"symbol": symbol, "table": table, "index": ord(symbol) - 33, "tableindex": ord(table) - 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(encoding, "replace").strip()
ssid = (input[6] & 0b00011110) >> 1
if ssid > 0:
return "{callsign}-{ssid}".format(callsign=cs, ssid=ssid)
else:
return cs
class WeatherMapping(object):
def __init__(self, char, key, length, scale=None):
self.char = char
self.key = key
self.length = length
self.scale = scale
def matches(self, input):
return self.char == input[0] and len(input) > self.length
def updateWeather(self, weather, input):
def deepApply(obj, key, v):
keys = key.split(".")
if len(keys) > 1:
if not keys[0] in obj:
obj[keys[0]] = {}
deepApply(obj[keys[0]], ".".join(keys[1:]), v)
else:
obj[key] = v
try:
value = int(input[1 : 1 + self.length])
if self.scale:
value = self.scale(value)
deepApply(weather, self.key, value)
except ValueError:
pass
remain = input[1 + self.length :]
return weather, remain
class WeatherParser(object):
mappings = [
WeatherMapping("c", "wind.direction", 3),
WeatherMapping("s", "wind.speed", 3, lambda x: x * milesToKilometers),
WeatherMapping("g", "wind.gust", 3, lambda x: x * milesToKilometers),
WeatherMapping("t", "temperature", 3, fahrenheitToCelsius),
WeatherMapping("r", "rain.hour", 3, lambda x: x / 100 * inchesToMilimeters),
WeatherMapping("p", "rain.day", 3, lambda x: x / 100 * inchesToMilimeters),
WeatherMapping("P", "rain.sincemidnight", 3, lambda x: x / 100 * inchesToMilimeters),
WeatherMapping("h", "humidity", 2),
WeatherMapping("b", "barometricpressure", 5, lambda x: x / 10),
WeatherMapping("s", "snowfall", 3, lambda x: x * 25.4),
]
def __init__(self, data, weather={}):
self.data = data
self.weather = weather
def getWeather(self):
doWork = True
weather = self.weather
while doWork:
mapping = next((m for m in WeatherParser.mappings if m.matches(self.data)), None)
if mapping:
(weather, remain) = mapping.updateWeather(weather, self.data)
self.data = remain
doWork = len(self.data) > 0
else:
doWork = False
return weather
def getRemainder(self):
return self.data
class AprsLocation(LatLngLocation):
def __init__(self, data):
super().__init__(data["lat"], data["lon"])
self.data = data
def __dict__(self):
res = super(AprsLocation, self).__dict__()
for key in ["comment", "symbol", "course", "speed"]:
if key in self.data:
res[key] = self.data[key]
return res
class AprsParser(object):
def __init__(self, handler):
self.ax25parser = Ax25Parser()
self.deframer = KissDeframer()
self.dial_freq = None
self.band = None
self.handler = handler
self.metric = self.getMetric()
def setDialFrequency(self, freq):
self.dial_freq = freq
self.band = Bandplan.getSharedInstance().findBand(freq)
self.metric = self.getMetric()
def getMetric(self):
band = "unknown"
if self.band is not None:
band = self.band.getName()
name = "aprs.decodes.{band}.aprs".format(band=band)
metrics = Metrics.getSharedInstance()
metric = metrics.getMetric(name)
if metric is None:
metric = CounterMetric()
metrics.addMetric(name, metric)
return metric
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("decoded APRS data: %s", aprsData)
self.updateMap(aprsData)
self.metric.inc()
self.handler.write_aprs_data(aprsData)
except Exception:
logger.exception("exception while parsing aprs data")
def updateMap(self, mapData):
if "type" in mapData and mapData["type"] == "thirdparty" and "data" in mapData:
mapData = mapData["data"]
if "lat" in mapData and "lon" in mapData:
loc = AprsLocation(mapData)
source = mapData["source"]
if "type" in mapData:
if mapData["type"] == "item":
source = mapData["item"]
elif mapData["type"] == "object":
source = mapData["object"]
Map.getSharedInstance().updateLocation(source, loc, "APRS", self.band)
def hasCompressedCoordinates(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": getSymbolData(raw[18], raw[8])}
def parseCompressedCoordinates(self, raw):
return {
"lat": 90 - decodeBase91(raw[1:5]) / 380926,
"lon": -180 + decodeBase91(raw[5:9]) / 190463,
"symbol": getSymbolData(raw[9], raw[0]),
}
def parseTimestamp(self, raw):
now = datetime.now()
if raw[6] == "h":
ts = datetime.strptime(raw[0:6], "%H%M%S")
ts = ts.replace(year=now.year, month=now.month, day=now.month, tzinfo=timezone.utc)
else:
ts = datetime.strptime(raw[0:6], "%d%H%M")
ts = ts.replace(year=now.year, month=now.month)
if raw[6] == "z":
ts = ts.replace(tzinfo=timezone.utc)
elif raw[6] == "/":
ts = ts.replace(tzinfo=now.tzinfo)
else:
logger.warning("invalid timezone info byte: %s", raw[6])
return int(ts.timestamp() * 1000)
def parseStatusUpate(self, raw):
res = {"type": "status"}
if raw[6] == "z":
res["timestamp"] = self.parseTimestamp(raw[0:7])
res["comment"] = raw[7:]
else:
res["comment"] = raw
return res
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] == ord("`") or information[0] == ord("'"):
aprsData.update(MicEParser().parse(data))
return aprsData
information = information.decode(encoding, "replace")
# APRS data type identifier
dti = information[0]
if dti == "!" or dti == "=":
# position without timestamp
aprsData.update(self.parseRegularAprsData(information[1:]))
elif dti == "/" or dti == "@":
# position with timestamp
aprsData["timestamp"] = self.parseTimestamp(information[1:8])
aprsData.update(self.parseRegularAprsData(information[8:]))
elif dti == ">":
# status update
aprsData.update(self.parseStatusUpate(information[1:]))
elif dti == "}":
# third party
aprsData.update(self.parseThirdpartyAprsData(information[1:]))
elif dti == ":":
# message
aprsData.update(self.parseMessage(information[1:]))
elif dti == ";":
# object
aprsData.update(self.parseObject(information[1:]))
elif dti == ")":
# item
aprsData.update(self.parseItem(information[1:]))
return aprsData
def parseObject(self, information):
result = {"type": "object"}
if len(information) > 16:
result["object"] = information[0:9].strip()
result["live"] = information[9] == "*"
result["timestamp"] = self.parseTimestamp(information[10:17])
result.update(self.parseRegularAprsData(information[17:]))
# override type, losing information about compression
result["type"] = "object"
return result
def parseItem(self, information):
result = {"type": "item"}
if len(information) > 3:
indexes = [information[0:10].find(p) for p in ["!", "_"]]
filtered = [i for i in indexes if i >= 3]
filtered.sort()
if len(filtered):
index = filtered[0]
result["item"] = information[0:index]
result["live"] = information[index] == "!"
result.update(self.parseRegularAprsData(information[index + 1 :]))
# override type, losing information about compression
result["type"] = "item"
return result
def parseMessage(self, information):
result = {"type": "message"}
if len(information) > 9 and information[9] == ":":
result["adressee"] = information[0:9]
message = information[10:]
if len(message) > 3 and message[0:3] == "ack":
result["type"] = "messageacknowledgement"
result["messageid"] = int(message[3:8])
elif len(message) > 3 and message[0:3] == "rej":
result["type"] = "messagerejection"
result["messageid"] = int(message[3:8])
else:
matches = messageIdRegex.match(message)
if matches:
result["messageid"] = int(matches.group(2))
message = matches.group(1)
result["message"] = message
return result
def parseThirdpartyAprsData(self, information):
matches = thirdpartyeRegex.match(information)
if matches:
path = matches.group(2).split(",")
destination = next((c.strip("*").upper() for c in path if c.endswith("*")), None)
data = self.parseAprsData(
{
"source": matches.group(1).upper(),
"destination": destination,
"path": path,
"data": matches.group(6).encode(encoding),
}
)
return {"type": "thirdparty", "data": data}
return {"type": "thirdparty"}
def parseRegularAprsData(self, information):
if self.hasCompressedCoordinates(information):
aprsData = self.parseCompressedCoordinates(information[0:10])
aprsData["type"] = "compressed"
if information[10] != " ":
if information[10] == "{":
# pre-calculated radio range
aprsData["range"] = 2 * 1.08 ** (ord(information[11]) - 33) * milesToKilometers
else:
aprsData["course"] = (ord(information[10]) - 33) * 4
# speed is in knots... convert to metric (km/h)
aprsData["speed"] = (1.08 ** (ord(information[11]) - 33) - 1) * knotsToKilometers
# compression type
t = ord(information[12])
aprsData["fix"] = (t & 0b00100000) > 0
sources = ["other", "GLL", "GGA", "RMC"]
aprsData["nmeasource"] = sources[(t & 0b00011000) >> 3]
origins = [
"Compressed",
"TNC BText",
"Software",
"[tbd]",
"KPC3",
"Pico",
"Other tracker",
"Digipeater conversion",
]
aprsData["compressionorigin"] = origins[t & 0b00000111]
comment = information[13:]
else:
aprsData = self.parseUncompressedCoordinates(information[0:19])
aprsData["type"] = "regular"
comment = information[19:]
def decodeHeightGainDirectivity(comment):
res = {"height": 2 ** int(comment[4]) * 10 * feetToMeters, "gain": int(comment[5])}
directivity = int(comment[6])
if directivity == 0:
res["directivity"] = "omni"
elif 0 < directivity < 9:
res["directivity"] = directivity * 45
return res
# aprs data extensions
# yes, weather stations are officially identified by their symbols. go figure...
if "symbol" in aprsData and aprsData["symbol"]["index"] == 62:
# weather report
weather = {}
if len(comment) > 6 and comment[3] == "/":
try:
weather["wind"] = {"direction": int(comment[0:3]), "speed": int(comment[4:7]) * milesToKilometers}
except ValueError:
pass
comment = comment[7:]
parser = WeatherParser(comment, weather)
aprsData["weather"] = parser.getWeather()
comment = parser.getRemainder()
elif len(comment) > 6:
if comment[3] == "/":
# course and speed
# for a weather report, this would be wind direction and speed
try:
aprsData["course"] = int(comment[0:3])
aprsData["speed"] = int(comment[4:7]) * knotsToKilometers
except ValueError:
pass
comment = comment[7:]
elif comment[0:3] == "PHG":
# station power and effective antenna height/gain/directivity
try:
powerCodes = [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
aprsData["power"] = powerCodes[int(comment[3])]
aprsData.update(decodeHeightGainDirectivity(comment))
except ValueError:
pass
comment = comment[7:]
elif comment[0:3] == "RNG":
# pre-calculated radio range
try:
aprsData["range"] = int(comment[3:7]) * milesToKilometers
except ValueError:
pass
comment = comment[7:]
elif comment[0:3] == "DFS":
# direction finding signal strength and antenna height/gain
try:
aprsData["strength"] = int(comment[3])
aprsData.update(decodeHeightGainDirectivity(comment))
except ValueError:
pass
comment = comment[7:]
matches = altitudeRegex.match(comment)
if matches:
aprsData["altitude"] = int(matches.group(2)) * feetToMeters
comment = matches.group(1) + matches.group(3)
aprsData["comment"] = comment
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"]
rawLatitude = [self.extractNumber(c) for c in destination[0:6]]
lat = self.listToNumber(rawLatitude[0:2]) + self.listToNumber(rawLatitude[2:6]) / 6000
if ord(destination[3]) <= ord("9"):
lat *= -1
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
speed = (information[4] - 28) * 10
dc28 = information[5] - 28
speed += int(dc28 / 10)
course = (dc28 % 10) * 100
course += information[6] - 28
if speed >= 800:
speed -= 800
if course >= 400:
course -= 400
# speed is in knots... convert to metric (km/h)
speed *= knotsToKilometers
comment = information[9:].decode(encoding, "replace").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 {
"fix": information[0] == ord("`") or information[0] == 0x1C,
"lat": lat,
"lon": lon,
"comment": comment,
"altitude": altitude,
"speed": speed,
"course": course,
"device": device,
"type": "Mic-E",
"symbol": getSymbolData(chr(information[7]), chr(information[8])),
}