restore aprs functionality
This commit is contained in:
Jakob Ketterl
@ -1,10 +1,15 @@
|
||||
from owrx.aprs.kiss import KissDeframer
|
||||
from owrx.map import Map, LatLngLocation
|
||||
from owrx.metrics import Metrics, CounterMetric
|
||||
from owrx.parser import Parser
|
||||
from owrx.bands import Bandplan
|
||||
from datetime import datetime, timezone
|
||||
from csdr.module import Module
|
||||
from pycsdr.modules import Reader
|
||||
from pycsdr.types import Format
|
||||
from threading import Thread
|
||||
from io import BytesIO
|
||||
import re
|
||||
import logging
|
||||
import pickle
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -45,7 +50,40 @@ def getSymbolData(symbol, table):
|
||||
return {"symbol": symbol, "table": table, "index": ord(symbol) - 33, "tableindex": ord(table) - 33}
|
||||
|
||||
|
||||
class Ax25Parser(object):
|
||||
class Ax25Parser(Module, Thread):
|
||||
def __init__(self):
|
||||
self.doRun = True
|
||||
super().__init__()
|
||||
|
||||
def getInputFormat(self) -> Format:
|
||||
return Format.CHAR
|
||||
|
||||
def getOutputFormat(self) -> Format:
|
||||
return Format.CHAR
|
||||
|
||||
def setReader(self, reader: Reader) -> None:
|
||||
super().setReader(reader)
|
||||
self.start()
|
||||
|
||||
def stop(self):
|
||||
self.doRun = False
|
||||
self.reader.stop()
|
||||
|
||||
def run(self):
|
||||
while self.doRun:
|
||||
data = self.reader.read()
|
||||
if data is None:
|
||||
self.doRun = False
|
||||
break
|
||||
io = BytesIO(data.tobytes())
|
||||
try:
|
||||
while True:
|
||||
frame = self.parse(pickle.load(io))
|
||||
if frame is not None:
|
||||
self.writer.write(pickle.dumps(frame))
|
||||
except EOFError:
|
||||
pass
|
||||
|
||||
def parse(self, ax25frame):
|
||||
control_pid = ax25frame.find(bytes([0x03, 0xF0]))
|
||||
if control_pid % 7 > 0:
|
||||
@ -54,7 +92,7 @@ class Ax25Parser(object):
|
||||
def chunks(l, n):
|
||||
"""Yield successive n-sized chunks from l."""
|
||||
for i in range(0, len(l), n):
|
||||
yield l[i : i + n]
|
||||
yield l[i:i + n]
|
||||
|
||||
return {
|
||||
"destination": self.extractCallsign(ax25frame[0:7]),
|
||||
@ -117,9 +155,9 @@ class WeatherParser(object):
|
||||
WeatherMapping("s", "snowfall", 3, lambda x: x * 25.4),
|
||||
]
|
||||
|
||||
def __init__(self, data, weather={}):
|
||||
def __init__(self, data, weather=None):
|
||||
self.data = data
|
||||
self.weather = weather
|
||||
self.weather = {} if weather is None else weather
|
||||
|
||||
def getWeather(self):
|
||||
doWork = True
|
||||
@ -151,16 +189,44 @@ class AprsLocation(LatLngLocation):
|
||||
return res
|
||||
|
||||
|
||||
class AprsParser(Parser):
|
||||
def __init__(self, handler):
|
||||
super().__init__(handler)
|
||||
self.ax25parser = Ax25Parser()
|
||||
self.deframer = KissDeframer()
|
||||
class AprsParser(Module, Thread):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.metrics = {}
|
||||
self.doRun = True
|
||||
self.band = None
|
||||
|
||||
def setDialFrequency(self, freq):
|
||||
super().setDialFrequency(freq)
|
||||
self.metrics = {}
|
||||
self.band = Bandplan.getSharedInstance().findBand(freq)
|
||||
|
||||
def setReader(self, reader: Reader) -> None:
|
||||
super().setReader(reader)
|
||||
self.start()
|
||||
|
||||
def getInputFormat(self) -> Format:
|
||||
return Format.CHAR
|
||||
|
||||
def getOutputFormat(self) -> Format:
|
||||
return Format.CHAR
|
||||
|
||||
def run(self):
|
||||
while self.doRun:
|
||||
data = self.reader.read()
|
||||
if data is None:
|
||||
self.doRun = False
|
||||
break
|
||||
io = BytesIO(data.tobytes())
|
||||
try:
|
||||
while True:
|
||||
frame = self.parse(pickle.load(io))
|
||||
if frame is not None:
|
||||
self.writer.write(pickle.dumps(frame))
|
||||
except EOFError:
|
||||
pass
|
||||
|
||||
def stop(self):
|
||||
self.doRun = False
|
||||
self.reader.stop()
|
||||
|
||||
def getMetric(self, category):
|
||||
if category not in self.metrics:
|
||||
@ -184,22 +250,22 @@ class AprsParser(Parser):
|
||||
return False
|
||||
return True
|
||||
|
||||
def parse(self, raw):
|
||||
for frame in self.deframer.parse(raw):
|
||||
try:
|
||||
data = self.ax25parser.parse(frame)
|
||||
def parse(self, data):
|
||||
try:
|
||||
# TODO how can we tell if this is an APRS frame at all?
|
||||
aprsData = self.parseAprsData(data)
|
||||
|
||||
# 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.getMetric("total").inc()
|
||||
if self.isDirect(aprsData):
|
||||
self.getMetric("direct").inc()
|
||||
|
||||
logger.debug("decoded APRS data: %s", aprsData)
|
||||
self.updateMap(aprsData)
|
||||
self.getMetric("total").inc()
|
||||
if self.isDirect(aprsData):
|
||||
self.getMetric("direct").inc()
|
||||
self.handler.write_aprs_data(aprsData)
|
||||
except Exception:
|
||||
logger.exception("exception while parsing aprs data")
|
||||
# the frontend uses this to distinguis hessages from the different parsers
|
||||
aprsData["mode"] = "APRS"
|
||||
return 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:
|
||||
|
||||
@ -1,5 +1,10 @@
|
||||
from pycsdr.modules import Reader
|
||||
from pycsdr.types import Format
|
||||
from csdr.module import Module
|
||||
from threading import Thread
|
||||
import socket
|
||||
import time
|
||||
import pickle
|
||||
|
||||
import logging
|
||||
|
||||
@ -11,33 +16,37 @@ TFEND = 0xDC
|
||||
TFESC = 0xDD
|
||||
|
||||
|
||||
class KissClient(object):
|
||||
def __init__(self, port):
|
||||
delay = 0.5
|
||||
retries = 0
|
||||
while True:
|
||||
try:
|
||||
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.socket.connect(("localhost", port))
|
||||
break
|
||||
except ConnectionError:
|
||||
if retries > 20:
|
||||
logger.error("maximum number of connection attempts reached. did direwolf start up correctly?")
|
||||
raise
|
||||
retries += 1
|
||||
time.sleep(delay)
|
||||
|
||||
def read(self):
|
||||
return self.socket.recv(1)
|
||||
|
||||
|
||||
class KissDeframer(object):
|
||||
class KissDeframer(Module, Thread):
|
||||
def __init__(self):
|
||||
self.escaped = False
|
||||
self.buf = bytearray()
|
||||
self.doRun = True
|
||||
super().__init__()
|
||||
|
||||
def getInputFormat(self) -> Format:
|
||||
return Format.CHAR
|
||||
|
||||
def getOutputFormat(self) -> Format:
|
||||
return Format.CHAR
|
||||
|
||||
def setReader(self, reader: Reader) -> None:
|
||||
super().setReader(reader)
|
||||
self.start()
|
||||
|
||||
def run(self):
|
||||
while self.doRun:
|
||||
data = self.reader.read()
|
||||
if data is None:
|
||||
self.doRun = False
|
||||
else:
|
||||
for frame in self.parse(data):
|
||||
self.writer.write(pickle.dumps(frame))
|
||||
|
||||
def stop(self):
|
||||
self.doRun = False
|
||||
self.reader.stop()
|
||||
|
||||
def parse(self, input):
|
||||
frames = []
|
||||
for b in input:
|
||||
if b == FESC:
|
||||
self.escaped = True
|
||||
@ -49,11 +58,10 @@ class KissDeframer(object):
|
||||
else:
|
||||
logger.warning("invalid escape char: %s", str(input[0]))
|
||||
self.escaped = False
|
||||
elif input[0] == FEND:
|
||||
elif b == FEND:
|
||||
# data frames start with 0x00
|
||||
if len(self.buf) > 1 and self.buf[0] == 0x00:
|
||||
frames += [self.buf[1:]]
|
||||
yield self.buf[1:]
|
||||
self.buf = bytearray()
|
||||
else:
|
||||
self.buf.append(b)
|
||||
return frames
|
||||
|
||||
94
owrx/aprs/module.py
Normal file
94
owrx/aprs/module.py
Normal file
@ -0,0 +1,94 @@
|
||||
from csdr.module import Module
|
||||
from pycsdr.types import Format
|
||||
from pycsdr.modules import Reader, Writer, TcpSource
|
||||
from subprocess import Popen, PIPE
|
||||
from owrx.aprs.direwolf import DirewolfConfig
|
||||
from owrx.config.core import CoreConfig
|
||||
import threading
|
||||
import time
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DirewolfModule(Module):
|
||||
def __init__(self, service: bool = False):
|
||||
self.process = None
|
||||
self.inputReader = None
|
||||
self.tcpSource = None
|
||||
self.service = service
|
||||
super().__init__()
|
||||
|
||||
def setReader(self, reader: Reader) -> None:
|
||||
super().setReader(reader)
|
||||
self.start()
|
||||
|
||||
def setWriter(self, writer: Writer) -> None:
|
||||
super().setWriter(writer)
|
||||
if self.tcpSource is not None:
|
||||
self.tcpSource.setWriter(writer)
|
||||
|
||||
def getInputFormat(self) -> Format:
|
||||
return Format.SHORT
|
||||
|
||||
def getOutputFormat(self) -> Format:
|
||||
return Format.CHAR
|
||||
|
||||
def start(self):
|
||||
temporary_directory = CoreConfig().get_temporary_directory()
|
||||
direwolf_config_path = "{tmp_dir}/openwebrx_direwolf_{myid}.conf".format(
|
||||
tmp_dir=temporary_directory, myid=id(self)
|
||||
)
|
||||
direwolf_config = DirewolfConfig()
|
||||
# TODO
|
||||
# direwolf_config.wire(self)
|
||||
|
||||
file = open(direwolf_config_path, "w")
|
||||
file.write(direwolf_config.getConfig(self.service))
|
||||
file.close()
|
||||
|
||||
# direwolf -c {direwolf_config} -r {audio_rate} -t 0 -q d -q h 1>&2
|
||||
self.process = Popen(
|
||||
["direwolf", "-c", direwolf_config_path, "-r", "48000", "-t", "0", "-q", "d", "-q", "h"],
|
||||
start_new_session=True,
|
||||
stdin=PIPE,
|
||||
)
|
||||
|
||||
threading.Thread(target=self.pump(self.reader.read, self.process.stdin.write)).start()
|
||||
|
||||
delay = 0.5
|
||||
retries = 0
|
||||
while True:
|
||||
try:
|
||||
self.tcpSource = TcpSource(direwolf_config.getPort(), Format.CHAR)
|
||||
if self.writer:
|
||||
self.tcpSource.setWriter(self.writer)
|
||||
break
|
||||
except ConnectionError:
|
||||
if retries > 20:
|
||||
logger.error("maximum number of connection attempts reached. did direwolf start up correctly?")
|
||||
raise
|
||||
retries += 1
|
||||
time.sleep(delay)
|
||||
|
||||
def stop(self):
|
||||
if self.process is not None:
|
||||
self.process.terminate()
|
||||
self.process.wait()
|
||||
self.process = None
|
||||
self.reader.stop()
|
||||
|
||||
def pump(self, read, write):
|
||||
def copy():
|
||||
while True:
|
||||
data = None
|
||||
try:
|
||||
data = read()
|
||||
except ValueError:
|
||||
pass
|
||||
if data is None or isinstance(data, bytes) and len(data) == 0:
|
||||
break
|
||||
write(data)
|
||||
|
||||
return copy
|
||||
Reference in New Issue
Block a user