new modificitions for owrx_connector support

owrx/feature.py

@@ -19,6 +19,7 @@ class FeatureDetector(object):
     features = {
         "core": ["csdr", "nmux", "nc"],
         "rtl_sdr": ["rtl_sdr"],
+        "rtl_sdr_socket": ["owrx_connector"],
         "sdrplay": ["rx_tools"],
         "hackrf": ["hackrf_transfer"],
         "airspy": ["airspy_rx"],
@@ -163,7 +164,10 @@ class FeatureDetector(object):
         def check_digiham_version(command):
             try:
                 process = subprocess.Popen([command, "--version"], stdout=subprocess.PIPE)
-                version = LooseVersion(digiham_version_regex.match(process.stdout.readline().decode()).group(1))
+                matches = digiham_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:
@@ -185,6 +189,40 @@ class FeatureDetector(object):
             True,
         )
 
+    def has_owrx_connector(self):
+        """
+        The owrx_connector package offers direct interfacing between your hardware and openwebrx. It allows quicker
+        frequency switching, uses less CPU and can even provide more stability in some cases.
+
+        You can get it here: https://github.com/jketterl/owrx_connector
+        """
+        required_version = LooseVersion("0.1")
+
+        owrx_connector_version_regex = re.compile("^owrx-connector version (.*)$")
+
+        def check_owrx_connector_version(command):
+            try:
+                process = subprocess.Popen([command, "--version"], stdout=subprocess.PIPE)
+                matches = owrx_connector_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
+
+        return reduce(
+            and_,
+            map(
+                check_owrx_connector_version,
+                [
+                    "rtl_connector",
+                ],
+            ),
+            True,
+        )
+
     def has_dsd(self):
         """
         The digital voice modes NXDN and D-Star can be decoded by the dsd project. Please note that you need the version

owrx/service.py

@@ -1,5 +1,5 @@
 import threading
-import socket
+from owrx.socket import getAvailablePort
 from datetime import datetime, timezone, timedelta
 from owrx.source import SdrService
 from owrx.bands import Bandplan
@@ -245,14 +245,6 @@ class ServiceHandler(object):
         self.startupTimer = threading.Timer(10, self.updateServices)
         self.startupTimer.start()
 
-    def getAvailablePort(self):
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.bind(("", 0))
-        s.listen(1)
-        port = s.getsockname()[1]
-        s.close()
-        return port
-
     def updateServices(self):
         logger.debug("re-scheduling services due to sdr changes")
         self.stopServices()
@@ -293,7 +285,7 @@ class ServiceHandler(object):
                     resampler_props["center_freq"] = cf
                     # TODO the + 24000 is a temporary fix since the resampling optimizer does not account for required bandwidths
                     resampler_props["samp_rate"] = bw + 24000
-                    resampler = Resampler(resampler_props, self.getAvailablePort(), self.source)
+                    resampler = Resampler(resampler_props, getAvailablePort(), self.source)
                     resampler.start()
 
                     for dial in group:

owrx/socket.py

@@ -0,0 +1,10 @@
+import socket
+
+
+def getAvailablePort():
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.bind(("", 0))
+    s.listen(1)
+    port = s.getsockname()[1]
+    s.close()
+    return port

owrx/source.py

@@ -5,6 +5,7 @@ from owrx.meta import MetaParser
 from owrx.wsjt import WsjtParser
 from owrx.aprs import AprsParser
 from owrx.metrics import Metrics, DirectMetric
+from owrx.socket import getAvailablePort
 import threading
 import csdr
 import time
@@ -105,15 +106,10 @@ class SdrSource(object):
         self.profile_id = None
         self.activateProfile()
         self.rtlProps = self.props.collect(
-            "samp_rate", "nmux_memory", "center_freq", "ppm", "rf_gain", "lna_gain", "rf_amp", "antenna", "if_gain"
+            *self.getEventNames()
         ).defaults(PropertyManager.getSharedInstance())
+        self.wireEvents()
 
-        def restart(name, value):
-            logger.debug("restarting sdr source due to property change: {0} changed to {1}".format(name, value))
-            self.stop()
-            self.start()
-
-        self.rtlProps.wire(restart)
         self.port = port
         self.monitor = None
         self.clients = []
@@ -123,6 +119,17 @@ class SdrSource(object):
         self.modificationLock = threading.Lock()
         self.failed = False
 
+    def getEventNames(self):
+        return ["samp_rate", "nmux_memory", "center_freq", "ppm", "rf_gain", "lna_gain", "rf_amp", "antenna", "if_gain"]
+
+    def wireEvents(self):
+        def restart(name, value):
+            logger.debug("restarting sdr source due to property change: {0} changed to {1}".format(name, value))
+            self.stop()
+            self.start()
+
+        self.rtlProps.wire(restart)
+
     # override this in subclasses
     def getCommand(self):
         pass
@@ -164,6 +171,9 @@ class SdrSource(object):
     def getPort(self):
         return self.port
 
+    def useNmux(self):
+        return True
+
     def start(self):
         self.modificationLock.acquire()
         if self.monitor:
@@ -172,7 +182,7 @@ class SdrSource(object):
 
         props = self.rtlProps
 
-        start_sdr_command = self.getCommand().format(
+        cmd = self.getCommand().format(
             **props.collect(
                 "samp_rate", "center_freq", "ppm", "rf_gain", "lna_gain", "rf_amp", "antenna", "if_gain"
             ).__dict__()
@@ -180,8 +190,9 @@ class SdrSource(object):
 
         format_conversion = self.getFormatConversion()
         if format_conversion is not None:
-            start_sdr_command += " | " + format_conversion
+            cmd += " | " + format_conversion
 
+        if self.useNmux():
             nmux_bufcnt = nmux_bufsize = 0
             while nmux_bufsize < props["samp_rate"] / 4:
                 nmux_bufsize += 4096
@@ -194,11 +205,12 @@ class SdrSource(object):
                 self.modificationLock.release()
                 return
             logger.debug("nmux_bufsize = %d, nmux_bufcnt = %d" % (nmux_bufsize, nmux_bufcnt))
-        cmd = start_sdr_command + " | nmux --bufsize %d --bufcnt %d --port %d --address 127.0.0.1" % (
+            cmd = cmd + " | nmux --bufsize %d --bufcnt %d --port %d --address 127.0.0.1" % (
                 nmux_bufsize,
                 nmux_bufcnt,
                 self.port,
             )
+
         self.process = subprocess.Popen(cmd, shell=True, preexec_fn=os.setpgrp)
         logger.info("Started rtl source: " + cmd)
 
@@ -229,6 +241,8 @@ class SdrSource(object):
         if not available:
             self.failed = True
 
+        self.postStart()
+
         self.modificationLock.release()
 
         for c in self.clients:
@@ -237,6 +251,9 @@ class SdrSource(object):
             else:
                 c.onSdrAvailable()
 
+    def postStart(self):
+        pass
+
     def isAvailable(self):
         return self.monitor is not None
 
@@ -390,6 +407,43 @@ class Resampler(SdrSource):
         pass
 
 
+class RtlSdrSocketSource(SdrSource):
+    def __init__(self, id, props, port):
+        super().__init__(id, props, port)
+        self.controlSocket = None
+        self.controlPort = getAvailablePort()
+
+    def getEventNames(self):
+        return ["samp_rate", "center_freq", "ppm", "rf_gain"]
+
+    def wireEvents(self):
+        def reconfigure(prop, value):
+            logger.debug("sending property change over control socket: {0} changed to {1}".format(prop, value))
+            self.controlSocket.send("{prop}:{value}\n".format(prop=prop, value=value).encode())
+
+        self.rtlProps.wire(reconfigure)
+
+    def postStart(self):
+        logger.debug("opening control socket...")
+        self.controlSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.controlSocket.connect(("localhost", self.controlPort))
+
+    def stop(self):
+        super().stop()
+        if self.controlSocket:
+            self.controlSocket.close()
+            self.controlSocket = None
+
+    def getCommand(self):
+        return "rtl_connector -p {port} -c {controlPort}".format(port=self.port, controlPort=self.controlPort) + " -s {samp_rate} -f {center_freq} -g {rf_gain}"
+
+    def getFormatConversion(self):
+        return None
+
+    def useNmux(self):
+        return False
+
+
 class RtlSdrSource(SdrSource):
     def getCommand(self):
         return "rtl_sdr -s {samp_rate} -f {center_freq} -p {ppm} -g {rf_gain} -"