joachim/openwebrx-clone
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'develop' into pycsdr

Browse Source
This commit is contained in:
Jakob Ketterl
2021-01-17 18:16:32 +01:00
parent f4629804ff e917b920c8
commit 297d6b540d
32 changed files with 758 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
owrx/__main__.py
View File

@ -11,8 +11,9 @@ from owrx.sdr import SdrService
from socketserver import ThreadingMixIn
from owrx.service import Services
from owrx.websocket import WebSocketConnection
from owrx.pskreporter import PskReporter
from owrx.reporting import ReportingEngine
from owrx.version import openwebrx_version
from owrx.audio import DecoderQueue
class ThreadedHttpServer(ThreadingMixIn, HTTPServer):
@ -67,4 +68,5 @@ Support and info: https://groups.io/g/openwebrx
except KeyboardInterrupt:
WebSocketConnection.closeAll()
Services.stop()
PskReporter.stop()
ReportingEngine.stopAll()
DecoderQueue.stopAll()

62
owrx/audio.py
View File

@ -7,7 +7,7 @@ import subprocess
import os
from multiprocessing.connection import Pipe, wait
from datetime import datetime, timedelta
from queue import Queue, Full
from queue import Queue, Full, Empty
import logging
@ -32,22 +32,30 @@ class QueueJob(object):
pass
PoisonPill = object()
class QueueWorker(threading.Thread):
def __init__(self, queue):
self.queue = queue
self.doRun = True
super().__init__(daemon=True)
super().__init__()
def run(self) -> None:
while self.doRun:
job = self.queue.get()
try:
job.run()
except Exception:
logger.exception("failed to decode job")
self.queue.onError()
finally:
job.unlink()
if job is PoisonPill:
self.doRun = False
# put the poison pill back on the queue for the next worker
self.queue.put(PoisonPill)
else:
try:
job.run()
except Exception:
logger.exception("failed to decode job")
self.queue.onError()
finally:
job.unlink()
self.queue.task_done()
@ -64,6 +72,13 @@ class DecoderQueue(Queue):
DecoderQueue.sharedInstance = DecoderQueue(maxsize=pm["decoding_queue_length"], workers=pm["decoding_queue_workers"])
return DecoderQueue.sharedInstance
@staticmethod
def stopAll():
with DecoderQueue.creationLock:
if DecoderQueue.sharedInstance is not None:
DecoderQueue.sharedInstance.stop()
DecoderQueue.sharedInstance = None
def __init__(self, maxsize, workers):
super().__init__(maxsize)
metrics = Metrics.getSharedInstance()
@ -78,6 +93,18 @@ class DecoderQueue(Queue):
metrics.addMetric("decoding.queue.error", self.errorCounter)
self.workers = [self.newWorker() for _ in range(0, workers)]
def stop(self):
logger.debug("shutting down the queue")
try:
# purge all remaining jobs
while not self.empty():
job = self.get()
job.unlink()
except Empty:
pass
# put() PoisonPill to tell workers to shut down
self.put(PoisonPill)
def put(self, item, **kwars):
self.inCounter.inc()
try:
@ -161,11 +188,10 @@ class AudioWriter(object):
self.timer.start()
def switchFiles(self):
self.switchingLock.acquire()
file = self.wavefile
filename = self.wavefilename
(self.wavefilename, self.wavefile) = self.getWaveFile()
self.switchingLock.release()
with self.switchingLock:
file = self.wavefile
filename = self.wavefilename
(self.wavefilename, self.wavefile) = self.getWaveFile()
file.close()
job = QueueJob(self, filename, self.dsp.get_operating_freq())
@ -205,9 +231,8 @@ class AudioWriter(object):
self._scheduleNextSwitch()
def write(self, data):
self.switchingLock.acquire()
self.wavefile.writeframes(data)
self.switchingLock.release()
with self.switchingLock:
self.wavefile.writeframes(data)
def stop(self):
self.outputWriter.close()
@ -229,7 +254,8 @@ class AudioWriter(object):
except Exception:
logger.exception("error closing wave file")
try:
os.unlink(self.wavefilename)
with self.switchingLock:
os.unlink(self.wavefilename)
except Exception:
logger.exception("error removing undecoded file")
self.wavefile = None

2
owrx/connection.py
View File

@ -1,4 +1,3 @@
from owrx.config import Config
from owrx.details import ReceiverDetails
from owrx.dsp import DspManager
from owrx.cpu import CpuUsageThread
@ -110,7 +109,6 @@ class OpenWebRxClient(Client, metaclass=ABCMeta):
class OpenWebRxReceiverClient(OpenWebRxClient, SdrSourceEventClient):
sdr_config_keys = [
"waterfall_min_level",
"waterfall_min_level",
"waterfall_max_level",
"samp_rate",

1
owrx/controllers/assets.py
View File

@ -129,6 +129,7 @@ class CompiledAssetsController(GzipMixin, ModificationAwareController):
"lib/MessagePanel.js",
"lib/Js8Threads.js",
"lib/Modes.js",
"lib/MetaPanel.js",
],
"map.js": [
"lib/jquery-3.2.1.min.js",

3
owrx/cpu.py
View File

@ -71,6 +71,7 @@ class CpuUsageThread(threading.Thread):
self.shutdown()
def shutdown(self):
CpuUsageThread.sharedInstance = None
with CpuUsageThread.creationLock:
CpuUsageThread.sharedInstance = None
self.doRun = False
self.endEvent.set()

21
owrx/feature.py
View File

@ -77,6 +77,7 @@ class FeatureDetector(object):
"digital_voice_freedv": ["freedv_rx", "sox"],
"digital_voice_m17": ["m17_demod", "sox"],
"wsjt-x": ["wsjtx", "sox"],
"wsjt-x-2-3": ["wsjtx_2_3", "sox"],
"packet": ["direwolf", "sox"],
"pocsag": ["digiham", "sox"],
"js8call": ["js8", "sox"],
@ -459,6 +460,26 @@ class FeatureDetector(object):
"""
return reduce(and_, map(self.command_is_runnable, ["jt9", "wsprd"]), True)
def _has_wsjtx_version(self, required_version):
wsjt_version_regex = re.compile("^WSJT-X (.*)$")
try:
process = subprocess.Popen(["wsjtx_app_version", "--version"], stdout=subprocess.PIPE)
matches = wsjt_version_regex.match(process.stdout.readline().decode())
if matches is None:
return False
version = LooseVersion(matches.group(1))
process.wait(1)
return version >= required_version
except FileNotFoundError:
return False
def has_wsjtx_2_3(self):
"""
Newer digital modes (e.g. FST4, FST4) require WSJT-X in at least version 2.3.
"""
return self.has_wsjtx() and self._has_wsjtx_version(LooseVersion("2.3"))
def has_js8(self):
"""
To decode JS8, you will need to install [JS8Call](http://js8call.com/)

4
owrx/js8.py
View File

@ -4,10 +4,10 @@ import re
from js8py import Js8
from js8py.frames import Js8FrameHeartbeat, Js8FrameCompound
from .map import Map, LocatorLocation
from .pskreporter import PskReporter
from .metrics import Metrics, CounterMetric
from .config import Config
from abc import ABCMeta, abstractmethod
from owrx.reporting import ReportingEngine
import logging
@ -102,7 +102,7 @@ class Js8Parser(Parser):
Map.getSharedInstance().updateLocation(
frame.callsign, LocatorLocation(frame.grid), "JS8", self.band
)
PskReporter.getSharedInstance().spot({
ReportingEngine.getSharedInstance().spot({
"callsign": frame.callsign,
"mode": "JS8",
"locator": frame.grid,

6
owrx/modes.py
View File

@ -54,7 +54,7 @@ class DigitalMode(Mode):
class Modes(object):
mappings = [
AnalogMode("nfm", "FM", bandpass=Bandpass(-4000, 4000)),
AnalogMode("wfm", "WFM", bandpass=Bandpass(-50000, 50000)),
AnalogMode("wfm", "WFM", bandpass=Bandpass(-75000, 75000)),
AnalogMode("am", "AM", bandpass=Bandpass(-4000, 4000)),
AnalogMode("lsb", "LSB", bandpass=Bandpass(-3000, -300)),
AnalogMode("usb", "USB", bandpass=Bandpass(300, 3000)),
@ -75,8 +75,8 @@ class Modes(object):
DigitalMode(
"wspr", "WSPR", underlying=["usb"], bandpass=Bandpass(1350, 1650), requirements=["wsjt-x"], service=True
),
DigitalMode("fst4", "FST4", underlying=["usb"], bandpass=Bandpass(0, 3000), requirements=["wsjt-x"], service=True),
DigitalMode("fst4w", "FST4W", underlying=["usb"], bandpass=Bandpass(1350, 1650), requirements=["wsjt-x"], service=True),
DigitalMode("fst4", "FST4", underlying=["usb"], bandpass=Bandpass(0, 3000), requirements=["wsjt-x-2-3"], service=True),
DigitalMode("fst4w", "FST4W", underlying=["usb"], bandpass=Bandpass(1350, 1650), requirements=["wsjt-x-2-3"], service=True),
DigitalMode("js8", "JS8Call", underlying=["usb"], bandpass=Bandpass(0, 3000), requirements=["js8call"], service=True),
DigitalMode(
"packet",

38
owrx/pskreporter.py
View File

@ -9,43 +9,19 @@ 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 PskReporterDummy(object):
"""
used in place of the PskReporter when reporting is disabled.
does nothing.
"""
def spot(self, spot):
pass
def cancelTimer(self):
pass
class PskReporter(object):
sharedInstance = None
creationLock = threading.Lock()
class PskReporter(Reporter):
interval = 300
supportedModes = ["FT8", "FT4", "JT9", "JT65", "FST4", "FST4W", "JS8"]
@staticmethod
def getSharedInstance():
with PskReporter.creationLock:
if PskReporter.sharedInstance is None:
if Config.get()["pskreporter_enabled"]:
PskReporter.sharedInstance = PskReporter()
else:
PskReporter.sharedInstance = PskReporterDummy()
return PskReporter.sharedInstance
def getSupportedModes(self):
return ["FT8", "FT4", "JT9", "JT65", "FST4", "JS8"]
@staticmethod
def stop():
if PskReporter.sharedInstance:
PskReporter.sharedInstance.cancelTimer()
def stop(self):
self.cancelTimer()
def __init__(self):
self.spots = []
@ -72,8 +48,6 @@ class PskReporter(object):
return reduce(and_, map(lambda key: s1[key] == s2[key], keys))
def spot(self, spot):
if not spot["mode"] in PskReporter.supportedModes:
return
with self.spotLock:
if any(x for x in self.spots if self.spotEquals(spot, x)):
# dupe

56
owrx/reporting.py Normal file
View File

@ -0,0 +1,56 @@
import threading
from abc import ABC, abstractmethod
from owrx.config import Config
class Reporter(ABC):
@abstractmethod
def stop(self):
pass
@abstractmethod
def spot(self, spot):
pass
@abstractmethod
def getSupportedModes(self):
return []
class ReportingEngine(object):
creationLock = threading.Lock()
sharedInstance = None
@staticmethod
def getSharedInstance():
with ReportingEngine.creationLock:
if ReportingEngine.sharedInstance is None:
ReportingEngine.sharedInstance = ReportingEngine()
return ReportingEngine.sharedInstance
@staticmethod
def stopAll():
with ReportingEngine.creationLock:
if ReportingEngine.sharedInstance is not None:
ReportingEngine.sharedInstance.stop()
def __init__(self):
self.reporters = []
config = Config.get()
if "pskreporter_enabled" in config and config["pskreporter_enabled"]:
# inline import due to circular dependencies
from owrx.pskreporter import PskReporter
self.reporters += [PskReporter()]
if "wsprnet_enabled" in config and config["wsprnet_enabled"]:
# inline import due to circular dependencies
from owrx.wsprnet import WsprnetReporter
self.reporters += [WsprnetReporter()]
def stop(self):
for r in self.reporters:
r.stop()
def spot(self, spot):
for r in self.reporters:
if spot["mode"] in r.getSupportedModes():
r.spot(spot)

36
owrx/source/__init__.py
View File

@ -75,9 +75,38 @@ class SdrSource(ABC):
self.state = SdrSource.STATE_STOPPED
self.busyState = SdrSource.BUSYSTATE_IDLE
self.validateProfiles()
if self.isAlwaysOn():
self.start()
def validateProfiles(self):
props = PropertyStack()
props.addLayer(1, self.props)
for id, p in self.props["profiles"].items():
props.replaceLayer(0, self._getProfilePropertyLayer(p))
if "center_freq" not in props:
logger.warning("Profile \"%s\" does not specify a center_freq", id)
continue
if "samp_rate" not in props:
logger.warning("Profile \"%s\" does not specify a samp_rate", id)
continue
if "start_freq" in props:
start_freq = props["start_freq"]
srh = props["samp_rate"] / 2
center_freq = props["center_freq"]
if start_freq < center_freq - srh or start_freq > center_freq + srh:
logger.warning("start_freq for profile \"%s\" is out of range", id)
def _getProfilePropertyLayer(self, profile):
layer = PropertyLayer()
for (key, value) in profile.items():
# skip the name, that would overwrite the source name.
if key == "name":
continue
layer[key] = value
return layer
def isAlwaysOn(self):
return "always-on" in self.props and self.props["always-on"]
@ -119,12 +148,7 @@ class SdrSource(ABC):
profile = profiles[profile_id]
self.profile_id = profile_id
layer = PropertyLayer()
for (key, value) in profile.items():
# skip the name, that would overwrite the source name.
if key == "name":
continue
layer[key] = value
layer = self._getProfilePropertyLayer(profile)
self.props.replaceLayer(0, layer)
def getId(self):

4
owrx/source/resampler.py
View File

@ -32,3 +32,7 @@ class Resampler(DirectSource):
def activateProfile(self, profile_id=None):
logger.warning("Resampler does not support setting profiles")
pass
def validateProfiles(self):
# resampler does not support profiles
pass

58
owrx/wsjt.py
View File

@ -2,7 +2,7 @@ from datetime import datetime, timezone
from owrx.map import Map, LocatorLocation
import re
from owrx.metrics import Metrics, CounterMetric
from owrx.pskreporter import PskReporter
from owrx.reporting import ReportingEngine
from owrx.parser import Parser
from owrx.audio import AudioChopperProfile
from abc import ABC, ABCMeta, abstractmethod
@ -156,19 +156,24 @@ class WsjtParser(Parser):
return
mode = profile.getMode()
if mode == "WSPR":
decoder = WsprDecoder(profile)
if mode in ["WSPR", "FST4W"]:
messageParser = BeaconMessageParser()
else:
decoder = Jt9Decoder(profile)
messageParser = QsoMessageParser()
if mode == "WSPR":
decoder = WsprDecoder(profile, messageParser)
else:
decoder = Jt9Decoder(profile, messageParser)
out = decoder.parse(msg, freq)
out["mode"] = mode
out["interval"] = profile.getInterval()
self.pushDecode(mode)
if "callsign" in out and "locator" in out:
Map.getSharedInstance().updateLocation(
out["callsign"], LocatorLocation(out["locator"]), mode, self.band
)
PskReporter.getSharedInstance().spot(out)
ReportingEngine.getSharedInstance().spot(out)
self.handler.write_wsjt_message(out)
except (ValueError, IndexError):
@ -195,10 +200,9 @@ class WsjtParser(Parser):
class Decoder(ABC):
locator_pattern = re.compile(".*\\s([A-Z0-9/]{2,})(\\sR)?\\s([A-R]{2}[0-9]{2})$")
def __init__(self, profile):
def __init__(self, profile, messageParser):
self.profile = profile
self.messageParser = messageParser
def parse_timestamp(self, instring):
dateformat = self.profile.getTimestampFormat()
@ -215,8 +219,19 @@ class Decoder(ABC):
def parse(self, msg, dial_freq):
pass
def parseMessage(self, msg):
m = Decoder.locator_pattern.match(msg)
class MessageParser(ABC):
@abstractmethod
def parse(self, msg):
pass
# Used in QSO-style modes (FT8, FT4, FST4)
class QsoMessageParser(MessageParser):
locator_pattern = re.compile(".*\\s([A-Z0-9/]{2,})(\\sR)?\\s([A-R]{2}[0-9]{2})$")
def parse(self, msg):
m = QsoMessageParser.locator_pattern.match(msg)
if m is None:
return {}
# this is a valid locator in theory, but it's somewhere in the arctic ocean, near the north pole, so it's very
@ -226,6 +241,17 @@ class Decoder(ABC):
return {"callsign": m.group(1), "locator": m.group(3)}
# Used in propagation reporting / beacon modes (WSPR / FST4W)
class BeaconMessageParser(MessageParser):
wspr_splitter_pattern = re.compile("([A-Z0-9/]*)\\s([A-R]{2}[0-9]{2})\\s([0-9]+)")
def parse(self, msg):
m = BeaconMessageParser.wspr_splitter_pattern.match(msg)
if m is None:
return {}
return {"callsign": m.group(1), "locator": m.group(2), "dbm": m.group(3)}
class Jt9Decoder(Decoder):
def parse(self, msg, dial_freq):
# ft8 sample
@ -245,13 +271,11 @@ class Jt9Decoder(Decoder):
"freq": dial_freq + int(msg[9:13]),
"msg": wsjt_msg,
}
result.update(self.parseMessage(wsjt_msg))
result.update(self.messageParser.parse(wsjt_msg))
return result
class WsprDecoder(Decoder):
wspr_splitter_pattern = re.compile("([A-Z0-9/]*)\\s([A-R]{2}[0-9]{2})\\s([0-9]+)")
def parse(self, msg, dial_freq):
# wspr sample
# '2600 -24 0.4 0.001492 -1 G8AXA JO01 33'
@ -266,11 +290,5 @@ class WsprDecoder(Decoder):
"drift": int(msg[20:23]),
"msg": wsjt_msg,
}
result.update(self.parseMessage(wsjt_msg))
result.update(self.messageParser.parse(wsjt_msg))
return result
def parseMessage(self, msg):
m = WsprDecoder.wspr_splitter_pattern.match(msg)
if m is None:
return {}
return {"callsign": m.group(1), "locator": m.group(2)}

90
owrx/wsprnet.py Normal file
View File

@ -0,0 +1,90 @@
from owrx.reporting import Reporter
from owrx.version import openwebrx_version
from owrx.config import Config
from owrx.locator import Locator
from owrx.metrics import Metrics, CounterMetric
from queue import Queue, Full
from urllib import request, parse
import threading
import logging
from datetime import datetime, timezone
logger = logging.getLogger(__name__)
class Worker(threading.Thread):
def __init__(self, queue: Queue):
self.queue = queue
self.doRun = True
# some constants that we don't expect to change
config = Config.get()
self.callsign = config["wsprnet_callsign"]
self.locator = Locator.fromCoordinates(config["receiver_gps"])
super().__init__(daemon=True)
def run(self):
while self.doRun:
try:
spot = self.queue.get()
self.uploadSpot(spot)
self.queue.task_done()
except Exception:
logger.exception("Exception while uploading WSPRNet spot")
def _getMode(self, spot):
interval = round(spot["interval"] / 60)
# FST4W modes are mapped not to conflict with WSPR modes 2 and 15:
if spot["mode"] != "WSPR" and interval in [2, 15]:
return interval + 1
return interval
def uploadSpot(self, spot):
# function=wspr&date=210114&time=1732&sig=-15&dt=0.5&drift=0&tqrg=7.040019&tcall=DF2UU&tgrid=JN48&dbm=37&version=2.3.0-rc3&rcall=DD5JFK&rgrid=JN58SC&rqrg=7.040047&mode=2
# {'timestamp': 1610655960000, 'db': -23.0, 'dt': 0.3, 'freq': 7040048, 'drift': -1, 'msg': 'LA3JJ JO59 37', 'callsign': 'LA3JJ', 'locator': 'JO59', 'mode': 'WSPR'}
date = datetime.fromtimestamp(spot["timestamp"] / 1000, tz=timezone.utc)
data = parse.urlencode({
"function": "wspr",
"date": date.strftime("%y%m%d"),
"time": date.strftime("%H%M"),
"sig": spot["db"],
"dt": spot["dt"],
# FST4W does not have drift
"drift": spot["drift"] if "drift" in spot else 0,
"tqrg": spot["freq"] / 1E6,
"tcall": spot["callsign"],
"tgrid": spot["locator"],
"dbm": spot["dbm"],
"version": openwebrx_version,
"rcall": self.callsign,
"rgrid": self.locator,
# mode 2 = WSPR 2 minutes
"mode": self._getMode(spot)
}).encode()
request.urlopen("http://wsprnet.org/post/", data)
class WsprnetReporter(Reporter):
def __init__(self):
# max 100 entries
self.queue = Queue(100)
# single worker
Worker(self.queue).start()
# metrics
metrics = Metrics.getSharedInstance()
self.spotCounter = CounterMetric()
metrics.addMetric("wsprnet.spots", self.spotCounter)
def stop(self):
pass
def spot(self, spot):
try:
self.queue.put(spot, block=False)
self.spotCounter.inc()
except Full:
logger.warning("WSPRNet Queue overflow, one spot lost")
def getSupportedModes(self):
return ["WSPR", "FST4W"]