openwebrx-clone/owrx/kiss.py

149 lines
4.9 KiB
Python
Raw Normal View History

import socket
import time
import logging
import random
from owrx.config import Config
logger = logging.getLogger(__name__)
FEND = 0xC0
FESC = 0xDB
TFEND = 0xDC
2019-08-15 17:56:59 +00:00
TFESC = 0xDD
2020-12-30 17:33:21 +00:00
FEET_PER_METER = 3.28084
2021-01-20 16:01:46 +00:00
2019-08-17 22:15:07 +00:00
class DirewolfConfig(object):
def getConfig(self, port, is_service):
pm = Config.get()
2019-08-17 22:15:07 +00:00
config = """
2019-08-17 23:39:23 +00:00
ACHANNELS 1
ADEVICE stdin null
2019-08-17 23:39:23 +00:00
CHANNEL 0
2019-08-17 22:15:07 +00:00
MYCALL {callsign}
MODEM 1200
2019-08-17 23:39:23 +00:00
2019-08-17 22:15:07 +00:00
KISSPORT {port}
AGWPORT off
""".format(
2019-08-17 22:16:08 +00:00
port=port, callsign=pm["aprs_callsign"]
2019-08-17 22:15:07 +00:00
)
if is_service and pm["aprs_igate_enabled"]:
2019-08-17 22:15:07 +00:00
config += """
IGSERVER {server}
IGLOGIN {callsign} {password}
""".format(
2019-08-17 22:16:08 +00:00
server=pm["aprs_igate_server"], callsign=pm["aprs_callsign"], password=pm["aprs_igate_password"]
2019-08-17 22:15:07 +00:00
)
if pm["aprs_igate_beacon"]:
2021-01-20 16:01:46 +00:00
# Format beacon lat/lon
2020-03-29 16:08:26 +00:00
lat = pm["receiver_gps"]["lat"]
lon = pm["receiver_gps"]["lon"]
direction_ns = "N" if lat > 0 else "S"
direction_we = "E" if lon > 0 else "W"
lat = abs(lat)
lon = abs(lon)
lat = "{0:02d}^{1:05.2f}{2}".format(int(lat), (lat - int(lat)) * 60, direction_ns)
lon = "{0:03d}^{1:05.2f}{2}".format(int(lon), (lon - int(lon)) * 60, direction_we)
2019-08-17 22:15:07 +00:00
2021-01-20 16:01:46 +00:00
# Format beacon details
symbol = str(pm["aprs_igate_symbol"]) if "aprs_igate_symbol" in pm else "R&"
gain = "GAIN=" + str(pm["aprs_igate_gain"]) if "aprs_igate_gain" in pm else ""
adir = "DIR=" + str(pm["aprs_igate_dir"]) if "aprs_igate_dir" in pm else ""
comment = str(pm["aprs_igate_comment"]) if "aprs_igate_comment" in pm else '"OpenWebRX APRS gateway"'
2020-12-10 05:54:06 +00:00
2021-01-20 16:01:46 +00:00
# Convert height from meters to feet if specified
2020-12-30 17:33:21 +00:00
height = ""
2020-12-30 19:16:12 +00:00
if "aprs_igate_height" in pm:
2020-12-30 17:33:21 +00:00
try:
height_m = float(pm["aprs_igate_height"])
height_ft = round(height_m * FEET_PER_METER)
2020-12-30 17:33:21 +00:00
height = "HEIGHT=" + str(height_ft)
except:
2021-01-20 16:01:46 +00:00
logger.error(
"Cannot parse 'aprs_igate_height', expected float: " + str(pm["aprs_igate_height"])
)
2020-12-30 17:33:21 +00:00
2021-01-20 16:01:46 +00:00
if (len(comment) > 0) and ((comment[0] != '"') or (comment[len(comment) - 1] != '"')):
comment = '"' + comment + '"'
elif len(comment) == 0:
comment = '""'
2020-12-12 16:59:34 +00:00
2020-12-30 19:21:07 +00:00
pbeacon = "PBEACON sendto=IG delay=0:30 every=60:00 symbol={symbol} lat={lat} long={lon} {height} {gain} {adir} comment={comment}".format(
2021-01-20 16:01:46 +00:00
symbol=symbol, lat=lat, lon=lon, height=height, gain=gain, adir=adir, comment=comment
)
2019-08-17 22:15:07 +00:00
2020-12-10 06:05:04 +00:00
logger.info("APRS PBEACON String: " + pbeacon)
2021-01-20 16:01:46 +00:00
2020-12-30 17:33:21 +00:00
config += "\n" + pbeacon + "\n"
2020-12-10 06:05:04 +00:00
2019-08-17 22:15:07 +00:00
return config
class KissClient(object):
@staticmethod
def getFreePort():
# direwolf has some strange hardcoded port ranges
while True:
try:
port = random.randrange(1024, 49151)
# test if port is available for use
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
2019-08-17 22:16:08 +00:00
s.bind(("localhost", port))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.close()
return port
except OSError:
pass
def __init__(self, port):
2021-01-20 16:01:46 +00:00
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):
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]))
self.escaped = False
elif input[0] == FEND:
# data frames start with 0x00
if len(self.buf) > 1 and self.buf[0] == 0x00:
frames += [self.buf[1:]]
self.buf = bytearray()
else:
self.buf.append(b)
return frames