openwebrx-clone/owrx/pskreporter.py
2021-01-14 20:52:56 +01:00

223 lines
7.4 KiB
Python

import logging
import threading
import time
import random
import socket
from functools import reduce
from operator import and_
from owrx.config import Config
from owrx.version import openwebrx_version
from owrx.locator import Locator
from owrx.metrics import Metrics, CounterMetric
from owrx.reporting import Reporter
logger = logging.getLogger(__name__)
class PskReporter(Reporter):
interval = 300
def getSupportedModes(self):
return ["FT8", "FT4", "JT9", "JT65", "FST4", "FST4W", "JS8"]
def stop(self):
self.cancelTimer()
def __init__(self):
self.spots = []
self.spotLock = threading.Lock()
self.uploader = Uploader()
self.timer = None
metrics = Metrics.getSharedInstance()
self.dupeCounter = CounterMetric()
metrics.addMetric("pskreporter.duplicates", self.dupeCounter)
self.spotCounter = CounterMetric()
metrics.addMetric("pskreporter.spots", self.spotCounter)
def scheduleNextUpload(self):
if self.timer:
return
delay = PskReporter.interval + random.uniform(0, 30)
logger.debug("scheduling next pskreporter upload in %f seconds", delay)
self.timer = threading.Timer(delay, self.upload)
self.timer.start()
def spotEquals(self, s1, s2):
keys = ["callsign", "timestamp", "locator", "mode", "msg"]
return reduce(and_, map(lambda key: s1[key] == s2[key], keys))
def spot(self, spot):
with self.spotLock:
if any(x for x in self.spots if self.spotEquals(spot, x)):
# dupe
self.dupeCounter.inc()
else:
self.spotCounter.inc()
self.spots.append(spot)
self.scheduleNextUpload()
def upload(self):
try:
with self.spotLock:
self.timer = None
spots = self.spots
self.spots = []
if spots:
self.uploader.upload(spots)
except Exception:
logger.exception("Failed to upload spots")
def cancelTimer(self):
if self.timer:
self.timer.cancel()
class Uploader(object):
receieverDelimiter = [0x99, 0x92]
senderDelimiter = [0x99, 0x93]
def __init__(self):
self.sequence = 0
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
def upload(self, spots):
logger.debug("uploading %i spots", len(spots))
for packet in self.getPackets(spots):
self.socket.sendto(packet, ("report.pskreporter.info", 4739))
def getPackets(self, spots):
encoded = [self.encodeSpot(spot) for spot in spots]
# filter out any erroneous encodes
encoded = [e for e in encoded if e is not None]
def chunks(l, n):
"""Yield successive n-sized chunks from l."""
for i in range(0, len(l), n):
yield l[i : i + n]
rHeader = self.getReceiverInformationHeader()
rInfo = self.getReceiverInformation()
sHeader = self.getSenderInformationHeader()
packets = []
# 50 seems to be a safe bet
for chunk in chunks(encoded, 50):
sInfo = self.getSenderInformation(chunk)
length = 16 + len(rHeader) + len(sHeader) + len(rInfo) + len(sInfo)
header = self.getHeader(length)
packets.append(header + rHeader + sHeader + rInfo + sInfo)
return packets
def getHeader(self, length):
self.sequence += 1
return bytes(
# protocol version
[0x00, 0x0A]
+ list(length.to_bytes(2, "big"))
+ list(int(time.time()).to_bytes(4, "big"))
+ list(self.sequence.to_bytes(4, "big"))
+ list((id(self) & 0xFFFFFFFF).to_bytes(4, "big"))
)
def encodeString(self, s):
return [len(s)] + list(s.encode("utf-8"))
def encodeSpot(self, spot):
try:
return bytes(
self.encodeString(spot["callsign"])
+ list(int(spot["freq"]).to_bytes(4, "big"))
+ list(int(spot["db"]).to_bytes(1, "big", signed=True))
+ self.encodeString(spot["mode"])
+ self.encodeString(spot["locator"])
# informationsource. 1 means "automatically extracted
+ [0x01]
+ list(int(spot["timestamp"] / 1000).to_bytes(4, "big"))
)
except Exception:
logger.exception("Error while encoding spot for pskreporter")
return None
def getReceiverInformationHeader(self):
pm = Config.get()
with_antenna = "pskreporter_antenna_information" in pm and pm["pskreporter_antenna_information"] is not None
num_fields = 4 if with_antenna else 3
length = 12 + num_fields * 8
return bytes(
# id
[0x00, 0x03]
# length
+ list(length.to_bytes(2, 'big'))
+ Uploader.receieverDelimiter
# number of fields
+ list(num_fields.to_bytes(2, 'big'))
# padding
+ [0x00, 0x00]
# receiverCallsign
+ [0x80, 0x02, 0xFF, 0xFF, 0x00, 0x00, 0x76, 0x8F]
# receiverLocator
+ [0x80, 0x04, 0xFF, 0xFF, 0x00, 0x00, 0x76, 0x8F]
# decodingSoftware
+ [0x80, 0x08, 0xFF, 0xFF, 0x00, 0x00, 0x76, 0x8F]
# antennaInformation
+ (
[0x80, 0x09, 0xFF, 0xFF, 0x00, 0x00, 0x76, 0x8F] if with_antenna else []
)
# padding
+ [0x00, 0x00]
)
def getReceiverInformation(self):
pm = Config.get()
bodyFields = [
# callsign
pm["pskreporter_callsign"],
# locator
Locator.fromCoordinates(pm["receiver_gps"]),
# decodingSoftware
"OpenWebRX " + openwebrx_version,
]
if "pskreporter_antenna_information" in pm and pm["pskreporter_antenna_information"] is not None:
bodyFields += [pm["pskreporter_antenna_information"]]
body = [b for s in bodyFields for b in self.encodeString(s)]
body = self.pad(body, 4)
body = bytes(Uploader.receieverDelimiter + list((len(body) + 4).to_bytes(2, "big")) + body)
return body
def getSenderInformationHeader(self):
return bytes(
# id, length
[0x00, 0x02, 0x00, 0x3C]
+ Uploader.senderDelimiter
# number of fields
+ [0x00, 0x07]
# senderCallsign
+ [0x80, 0x01, 0xFF, 0xFF, 0x00, 0x00, 0x76, 0x8F]
# frequency
+ [0x80, 0x05, 0x00, 0x04, 0x00, 0x00, 0x76, 0x8F]
# sNR
+ [0x80, 0x06, 0x00, 0x01, 0x00, 0x00, 0x76, 0x8F]
# mode
+ [0x80, 0x0A, 0xFF, 0xFF, 0x00, 0x00, 0x76, 0x8F]
# senderLocator
+ [0x80, 0x03, 0xFF, 0xFF, 0x00, 0x00, 0x76, 0x8F]
# informationSource
+ [0x80, 0x0B, 0x00, 0x01, 0x00, 0x00, 0x76, 0x8F]
# flowStartSeconds
+ [0x00, 0x96, 0x00, 0x04]
)
def getSenderInformation(self, chunk):
sInfo = self.padBytes(b"".join(chunk), 4)
sInfoLength = len(sInfo) + 4
return bytes(Uploader.senderDelimiter) + sInfoLength.to_bytes(2, "big") + sInfo
def pad(self, b, l):
return b + [0x00 for _ in range(0, -1 * len(b) % l)]
def padBytes(self, b, l):
return b + bytes([0x00 for _ in range(0, -1 * len(b) % l)])