openwebrx-clone/owrx/aprs.py

583 lines
22 KiB
Python
Raw Normal View History

from owrx.kiss import KissDeframer
from owrx.map import Map, LatLngLocation
from owrx.bands import Bandplan
2019-09-12 21:23:50 +00:00
from owrx.metrics import Metrics, CounterMetric
from datetime import datetime, timezone
2019-08-17 18:35:32 +00:00
import re
import logging
logger = logging.getLogger(__name__)
2019-08-15 20:10:58 +00:00
# speed is in knots... convert to metric (km/h)
2019-08-17 18:35:32 +00:00
knotsToKilometers = 1.852
feetToMeters = 0.3048
2019-08-17 20:04:45 +00:00
milesToKilometers = 1.609344
2019-08-27 09:32:50 +00:00
inchesToMilimeters = 25.4
2019-09-13 21:03:05 +00:00
2019-08-27 09:42:48 +00:00
def fahrenheitToCelsius(f):
return (f - 32) * 5 / 9
2019-08-15 20:10:58 +00:00
2019-09-13 21:03:05 +00:00
2019-08-16 05:29:31 +00:00
# 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"
2019-08-17 18:35:32 +00:00
# regex for altitute in comment field
altitudeRegex = re.compile("(^.*)\\/A=([0-9]{6})(.*$)")
2019-08-25 22:10:43 +00:00
# regex for parsing third-party headers
2019-08-26 09:41:22 +00:00
thirdpartyeRegex = re.compile("^([a-zA-Z0-9-]+)>((([a-zA-Z0-9-]+\\*?,)*)([a-zA-Z0-9-]+\\*?)):(.*)$")
2019-08-25 22:10:43 +00:00
2019-08-27 21:13:26 +00:00
# regex for getting the message id out of message
messageIdRegex = re.compile("^(.*){([0-9]{1,5})$")
2019-08-17 18:35:32 +00:00
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,
}
2019-08-15 16:08:20 +00:00
class Ax25Parser(object):
def parse(self, ax25frame):
2019-08-15 17:56:59 +00:00
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):
2019-08-15 17:56:59 +00:00
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)],
2019-08-15 17:56:59 +00:00
"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
2019-08-26 21:43:08 +00:00
class WeatherMapping(object):
2019-09-13 21:03:05 +00:00
def __init__(self, char, key, length, scale=None):
2019-08-26 21:43:08 +00:00
self.char = char
self.key = key
self.length = length
self.scale = scale
def matches(self, input):
2019-08-27 09:32:50 +00:00
return self.char == input[0] and len(input) > self.length
2019-08-26 21:43:08 +00:00
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
2019-09-13 21:03:05 +00:00
2019-08-27 09:32:50 +00:00
try:
2019-09-13 21:03:05 +00:00
value = int(input[1 : 1 + self.length])
2019-08-27 09:32:50 +00:00
if self.scale:
value = self.scale(value)
deepApply(weather, self.key, value)
except ValueError:
pass
2019-09-13 21:03:05 +00:00
remain = input[1 + self.length :]
2019-08-26 21:43:08 +00:00
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),
2019-08-27 09:42:48 +00:00
WeatherMapping("t", "temperature", 3, fahrenheitToCelsius),
2019-08-27 09:32:50 +00:00
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),
2019-08-26 21:43:08 +00:00
WeatherMapping("h", "humidity", 2),
2019-09-13 21:03:05 +00:00
WeatherMapping("b", "barometricpressure", 5, lambda x: x / 10),
2019-08-26 21:43:08 +00:00
WeatherMapping("s", "snowfall", 3, lambda x: x * 25.4),
]
2019-08-27 21:52:51 +00:00
def __init__(self, data, weather={}):
2019-08-26 21:43:08 +00:00
self.data = data
2019-08-27 21:52:51 +00:00
self.weather = weather
2019-08-26 21:43:08 +00:00
def getWeather(self):
doWork = True
2019-08-27 21:52:51 +00:00
weather = self.weather
2019-08-26 21:43:08 +00:00
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
2019-08-27 09:32:50 +00:00
doWork = len(self.data) > 0
2019-08-26 21:43:08 +00:00
else:
doWork = False
return weather
def getRemainder(self):
return self.data
2019-09-18 16:50:48 +00:00
class AprsLocation(LatLngLocation):
def __init__(self, data):
super().__init__(data["lat"], data["lon"])
2019-09-19 00:25:32 +00:00
self.data = data
2019-09-18 16:50:48 +00:00
def __dict__(self):
res = super(AprsLocation, self).__dict__()
2019-09-19 00:25:32 +00:00
for key in ["comment", "symbol", "course", "speed"]:
if key in self.data:
res[key] = self.data[key]
2019-09-18 16:50:48 +00:00
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
2019-09-12 21:23:50 +00:00
self.metric = self.getMetric()
def setDialFrequency(self, freq):
self.dial_freq = freq
self.band = Bandplan.getSharedInstance().findBand(freq)
2019-09-12 21:23:50 +00:00
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):
2019-08-15 16:08:20 +00:00
try:
data = self.ax25parser.parse(frame)
2019-08-15 16:08:20 +00:00
# TODO how can we tell if this is an APRS frame at all?
aprsData = self.parseAprsData(data)
2019-08-15 21:33:02 +00:00
logger.debug("decoded APRS data: %s", aprsData)
2019-08-25 22:10:43 +00:00
self.updateMap(aprsData)
2019-09-12 21:23:50 +00:00
self.metric.inc()
2019-08-15 16:08:20 +00:00
self.handler.write_aprs_data(aprsData)
except Exception:
logger.exception("exception while parsing aprs data")
2019-08-25 22:10:43 +00:00
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:
2019-09-18 16:50:48 +00:00
loc = AprsLocation(mapData)
2019-08-28 20:01:01 +00:00
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)
2019-08-25 22:10:43 +00:00
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
2019-08-15 17:56:59 +00:00
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")
2019-08-17 18:20:28 +00:00
# APRS data type identifier
dti = information[0]
if dti == "!" or dti == "=":
# position without timestamp
2019-08-15 13:53:55 +00:00
aprsData.update(self.parseRegularAprsData(information[1:]))
2019-08-17 18:20:28 +00:00
elif dti == "/" or dti == "@":
# position with timestamp
aprsData["timestamp"] = self.parseTimestamp(information[1:8])
2019-08-15 13:53:55 +00:00
aprsData.update(self.parseRegularAprsData(information[8:]))
2019-08-17 18:20:28 +00:00
elif dti == ">":
# status update
aprsData.update(self.parseStatusUpate(information[1:]))
2019-08-17 18:20:28 +00:00
elif dti == "}":
# third party
2019-08-25 22:10:43 +00:00
aprsData.update(self.parseThirdpartyAprsData(information[1:]))
2019-08-27 21:13:26 +00:00
elif dti == ":":
# message
aprsData.update(self.parseMessage(information[1:]))
elif dti == ";":
# object
2019-08-28 19:56:50 +00:00
aprsData.update(self.parseObject(information[1:]))
2019-08-27 21:13:26 +00:00
elif dti == ")":
# item
2019-08-28 19:56:50 +00:00
aprsData.update(self.parseItem(information[1:]))
2019-08-15 13:53:55 +00:00
return aprsData
2019-08-28 19:56:50 +00:00
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:
2019-08-28 20:09:52 +00:00
indexes = [information[0:10].find(p) for p in ["!", "_"]]
2019-08-28 19:56:50 +00:00
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] == "!"
2019-09-13 21:03:05 +00:00
result.update(self.parseRegularAprsData(information[index + 1 :]))
2019-08-28 19:56:50 +00:00
# override type, losing information about compression
result["type"] = "item"
return result
2019-08-27 21:13:26 +00:00
def parseMessage(self, information):
result = {"type": "message"}
2019-08-27 21:32:21 +00:00
if len(information) > 9 and information[9] == ":":
2019-08-27 21:13:26 +00:00
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:
2019-09-02 15:20:49 +00:00
result["messageid"] = int(matches.group(2))
message = matches.group(1)
2019-08-27 21:13:26 +00:00
result["message"] = message
return result
2019-08-25 22:10:43 +00:00
def parseThirdpartyAprsData(self, information):
matches = thirdpartyeRegex.match(information)
if matches:
2019-09-02 15:20:49 +00:00
path = matches.group(2).split(",")
2019-08-26 09:41:22 +00:00
destination = next((c.strip("*").upper() for c in path if c.endswith("*")), None)
2019-09-13 21:03:05 +00:00
data = self.parseAprsData(
{
"source": matches.group(1).upper(),
"destination": destination,
"path": path,
"data": matches.group(6).encode(encoding),
}
)
return {"type": "thirdparty", "data": data}
2019-08-25 22:10:43 +00:00
2019-09-13 21:03:05 +00:00
return {"type": "thirdparty"}
2019-08-25 22:10:43 +00:00
2019-08-15 13:53:55 +00:00
def parseRegularAprsData(self, information):
if self.hasCompressedCoordinates(information):
2019-08-15 13:53:55 +00:00
aprsData = self.parseCompressedCoordinates(information[0:10])
aprsData["type"] = "compressed"
if information[10] != " ":
if information[10] == "{":
# pre-calculated radio range
2019-08-17 20:04:45 +00:00
aprsData["range"] = 2 * 1.08 ** (ord(information[11]) - 33) * milesToKilometers
else:
2019-08-16 14:43:16 +00:00
aprsData["course"] = (ord(information[10]) - 33) * 4
2019-08-15 20:10:58 +00:00
# speed is in knots... convert to metric (km/h)
2019-08-17 18:35:32 +00:00
aprsData["speed"] = (1.08 ** (ord(information[11]) - 33) - 1) * knotsToKilometers
# compression type
2019-08-16 14:43:16 +00:00
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]
2019-08-17 20:04:45 +00:00
comment = information[13:]
else:
2019-08-15 13:53:55 +00:00
aprsData = self.parseUncompressedCoordinates(information[0:19])
aprsData["type"] = "regular"
2019-08-17 20:04:45 +00:00
comment = information[19:]
def decodeHeightGainDirectivity(comment):
2019-08-17 22:16:08 +00:00
res = {"height": 2 ** int(comment[4]) * 10 * feetToMeters, "gain": int(comment[5])}
2019-08-17 20:04:45 +00:00
directivity = int(comment[6])
if directivity == 0:
res["directivity"] = "omni"
elif 0 < directivity < 9:
res["directivity"] = directivity * 45
return res
# aprs data extensions
2019-09-18 16:50:48 +00:00
# yes, weather stations are officially identified by their symbols. go figure...
if "symbol" in aprsData and aprsData["symbol"]["index"] == 62:
2019-08-26 21:43:08 +00:00
# weather report
weather = {}
if len(comment) > 6 and comment[3] == "/":
try:
2019-09-13 21:03:05 +00:00
weather["wind"] = {"direction": int(comment[0:3]), "speed": int(comment[4:7]) * milesToKilometers}
2019-08-26 21:43:08 +00:00
except ValueError:
pass
comment = comment[7:]
2019-08-27 21:52:51 +00:00
parser = WeatherParser(comment, weather)
aprsData["weather"] = parser.getWeather()
2019-08-26 21:43:08 +00:00
comment = parser.getRemainder()
elif len(comment) > 6:
2019-08-17 20:04:45 +00:00
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
2019-08-17 20:04:45 +00:00
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
2019-08-17 20:04:45 +00:00
comment = comment[7:]
elif comment[0:3] == "RNG":
# pre-calculated radio range
try:
aprsData["range"] = int(comment[3:7]) * milesToKilometers
except ValueError:
pass
2019-08-17 20:04:45 +00:00
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
2019-08-17 20:04:45 +00:00
comment = comment[7:]
matches = altitudeRegex.match(comment)
2019-08-17 18:35:32 +00:00
if matches:
2019-09-02 15:20:49 +00:00
aprsData["altitude"] = int(matches.group(2)) * feetToMeters
comment = matches.group(1) + matches.group(3)
2019-08-17 20:04:45 +00:00
aprsData["comment"] = comment
2019-08-17 18:35:32 +00:00
return aprsData
2019-08-15 16:08:20 +00:00
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 = {
2019-08-15 17:56:59 +00:00
"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] == "(":
2019-08-15 17:56:59 +00:00
devices = {"5": "D578UV", "8": "D878UV"}
return comment[1:-2], {"manufacturer": "Anytone", "device": devices.get(comment[-1], "Unknown")}
if comment[-2] == "|":
2019-08-15 17:56:59 +00:00
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] == ":":
2019-08-15 17:56:59 +00:00
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
2019-08-15 16:08:20 +00:00
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
2019-08-15 20:10:58 +00:00
# speed is in knots... convert to metric (km/h)
2019-08-17 18:35:32 +00:00
speed *= knotsToKilometers
comment = information[9:].decode(encoding, "replace").strip()
2019-08-15 16:08:20 +00:00
(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)
2019-08-15 18:28:24 +00:00
return {
2019-08-17 22:16:08 +00:00
"fix": information[0] == ord("`") or information[0] == 0x1C,
2019-08-15 18:28:24 +00:00
"lat": lat,
"lon": lon,
"comment": comment,
"altitude": altitude,
"speed": speed,
"course": course,
2019-08-15 18:28:24 +00:00
"device": device,
"type": "Mic-E",
"symbol": getSymbolData(chr(information[7]), chr(information[8]))
2019-08-15 18:28:24 +00:00
}