first working nfm chain using pycsdr

csdr/__init__.py

@@ -35,6 +35,9 @@ from owrx.audio.chopper import AudioChopper
 
 from csdr.pipe import Pipe
 
+from csdr.chain.demodulator import DemodulatorChain
+from csdr.chain.fm import Fm
+
 import logging
 
 logger = logging.getLogger(__name__)
@@ -115,6 +118,10 @@ class Dsp(DirewolfConfigSubscriber):
         self.pipe_base_path = "{tmp_dir}/openwebrx_pipe_".format(tmp_dir=self.temporary_directory)
 
     def chain(self, which):
+        if self.pycsdr_enabled and which == "nfm":
+            self.pycsdr_chain = DemodulatorChain(self.samp_rate, self.get_audio_rate(), 0.0, Fm(self.get_audio_rate()))
+            return self.pycsdr_chain
+
         chain = ["nc -v 127.0.0.1 {nc_port}"]
         chain += ["csdr shift_addfast_cc --fifo {shift_pipe}"]
         if self.decimation > 1:
@@ -463,10 +470,7 @@ class Dsp(DirewolfConfigSubscriber):
         self.samp_rate = samp_rate
         self.calculate_decimation()
         if self.running:
-            if self.pycsdr_chain is not None:
+            self.restart()
-                self.pycsdr_chain.setSampleRate(self.samp_rate)
-            else:
-                self.restart()
 
     def calculate_decimation(self):
         (self.decimation, self.last_decimation) = self.get_decimation(self.samp_rate, self.get_audio_rate())
@@ -580,7 +584,11 @@ class Dsp(DirewolfConfigSubscriber):
             return
         self.offset_freq = offset_freq
         if self.running:
-            self.pipes["shift_pipe"].write("%g\n" % (-float(self.offset_freq) / self.samp_rate))
+            if self.pycsdr_chain is not None and isinstance(self.pycsdr_chain, DemodulatorChain):
+                self.pycsdr_chain.setShiftRate(-float(self.offset_freq) / self.samp_rate)
+            else:
+                self.pipes["shift_pipe"].write("%g\n" % (-float(self.offset_freq) / self.samp_rate))
+
 
     def set_center_freq(self, center_freq):
         # dsp only needs to know this to be able to pass it to decoders in the form of get_operating_freq()
@@ -596,9 +604,12 @@ class Dsp(DirewolfConfigSubscriber):
         self.low_cut = low_cut
         self.high_cut = high_cut
         if self.running:
-            self.pipes["bpf_pipe"].write(
+            if self.pycsdr_chain is not None and isinstance(self.pycsdr_chain, DemodulatorChain):
-                "%g %g\n" % (float(self.low_cut) / self.if_samp_rate(), float(self.high_cut) / self.if_samp_rate())
+                self.pycsdr_chain.setBandpass(low_cut, high_cut)
-            )
+            else:
+                self.pipes["bpf_pipe"].write(
+                    "%g %g\n" % (float(self.low_cut) / self.if_samp_rate(), float(self.high_cut) / self.if_samp_rate())
+                )
 
     def get_bpf(self):
         return [self.low_cut, self.high_cut]
@@ -615,7 +626,10 @@ class Dsp(DirewolfConfigSubscriber):
             else self.squelch_level
         )
         if self.running:
-            self.pipes["squelch_pipe"].write("%g\n" % (self.convertToLinear(actual_squelch)))
+            if self.pycsdr_chain is not None and isinstance(self.pycsdr_chain, DemodulatorChain):
+                self.pycsdr_chain.setSquelchLevel(self.convertToLinear(actual_squelch))
+            else:
+                self.pipes["squelch_pipe"].write("%g\n" % (self.convertToLinear(actual_squelch)))
 
     def set_unvoiced_quality(self, q):
         self.unvoiced_quality = q
@@ -708,7 +722,13 @@ class Dsp(DirewolfConfigSubscriber):
                 return
             self.running = True
 
-            command_base = " | ".join(self.chain(self.demodulator))
+            chain = self.chain(self.demodulator)
+            if self.pycsdr_enabled and isinstance(chain, DemodulatorChain):
+                chain.setInput(self.buffer)
+                self.output.send_output("audio", chain.getOutput().read)
+                return
+
+            command_base = " | ".join(chain)
 
             # create control pipes for csdr
             self.try_create_pipes(self.pipe_names, command_base)

csdr/chain/__init__.py

@@ -13,43 +13,57 @@ class Chain:
             self._connect(self.workers[i - 1], self.workers[i])
 
     def _connect(self, w1, w2):
-        buffer = Buffer(w1.getOutputFormat())
+        if isinstance(w1, Chain):
-        w1.setOutput(buffer)
+            buffer = w1.getOutput()
+        else:
+            buffer = Buffer(w1.getOutputFormat())
+            w1.setOutput(buffer)
         w2.setInput(buffer)
 
     def stop(self):
         for w in self.workers:
             w.stop()
         self.setInput(None)
-        self.setOutput(None)
+        if self.output is not None:
+            self.output.stop()
 
     def setInput(self, buffer):
         if self.input == buffer:
             return
         self.input = buffer
-        self.workers[0].setInput(buffer)
+        if self.workers:
+            self.workers[0].setInput(buffer)
+        else:
+            self.output = self.input
 
-    def setOutput(self, buffer):
+    def getOutput(self):
-        if self.output == buffer:
+        if self.output is None:
-            return
+            if self.workers:
-        if self.output is not None:
+                lastWorker = self.workers[-1]
-            self.output.stop()
+                if isinstance(lastWorker, Chain):
-        self.output = buffer
+                    self.output = lastWorker.getOutput()
-        self.workers[-1].setOutput(buffer)
+                else:
+                    self.output = Buffer(self.getOutputFormat())
+                    self.workers[-1].setOutput(self.output)
+            else:
+                self.output = self.input
+        return self.output
 
     def getOutputFormat(self):
-        return self.workers[-1].getOutputFormat()
+        if self.workers:
+            return self.workers[-1].getOutputFormat()
+        else:
+            return self.input.getOutputFormat()
 
     def pump(self, write):
-        if self.output is None:
+        output = self.getOutput()
-            self.setOutput(Buffer(self.getOutputFormat()))
 
         def copy():
             run = True
             while run:
                 data = None
                 try:
-                    data = self.output.read()
+                    data = output.read()
                 except ValueError:
                     pass
                 if data is None or (isinstance(data, bytes) and len(data) == 0):

csdr/chain/demodulator.py

@@ -0,0 +1,56 @@
+from csdr.chain import Chain
+from pycsdr.modules import Shift, FirDecimate, Bandpass, Squelch, FractionalDecimator
+from abc import ABCMeta, abstractmethod
+
+
+class Demodulator(Chain, metaclass=ABCMeta):
+    @abstractmethod
+    def setLastDecimation(self, decimation: Chain):
+        pass
+
+
+class DemodulatorChain(Chain):
+    def __init__(self, samp_rate: int, audioRate: int, shiftRate: float, demodulator: Demodulator):
+        self.shift = Shift(shiftRate)
+
+        decimation, fraction = self._getDecimation(samp_rate, audioRate)
+        if_samp_rate = samp_rate / decimation
+        transition = 0.15 * (if_samp_rate / float(samp_rate))
+        self.decimation = FirDecimate(decimation, transition)
+
+        bp_transition = 320.0 / if_samp_rate
+        self.bandpass = Bandpass(bp_transition, use_fft=True)
+
+        self.squelch = Squelch(5)
+
+        if fraction != 1.0:
+            demodulator.setLastDecimation(Chain(FractionalDecimator(fraction)))
+
+        workers = [
+            self.shift,
+            self.decimation,
+            self.bandpass,
+            self.squelch,
+            demodulator
+        ]
+
+        super().__init__(*workers)
+
+    def setShiftRate(self, rate: float):
+        self.shift.setRate(rate)
+
+    def setSquelchLevel(self, level: float):
+        self.squelch.setSquelchLevel(level)
+
+    def setBandpass(self, low_cut: float, high_cut: float):
+        self.bandpass.setBandpass(low_cut, high_cut)
+
+    def _getDecimation(self, input_rate, output_rate):
+        if output_rate <= 0:
+            raise ValueError("invalid output rate: {rate}".format(rate=output_rate))
+        decimation = 1
+        target_rate = output_rate
+        while input_rate / (decimation + 1) >= target_rate:
+            decimation += 1
+        fraction = float(input_rate / decimation) / output_rate
+        return decimation, fraction

csdr/chain/fm.py

@@ -0,0 +1,22 @@
+from csdr.chain.demodulator import Demodulator, Chain
+from pycsdr.modules import FmDemod, Limit, NfmDeemphasis, Agc, Convert
+from pycsdr.types import Format
+
+
+class Fm(Demodulator):
+    def __init__(self, sampleRate: int):
+        workers = [
+            FmDemod(),
+            Limit(),
+            # empty chain as placeholder for the "last decimation"
+            Chain(),
+            NfmDeemphasis(sampleRate),
+            Agc(Format.FLOAT),
+            Convert(Format.FLOAT, Format.SHORT),
+        ]
+        super().__init__(*workers)
+
+    def setLastDecimation(self, decimation: Chain):
+        # TODO: build api to replace workers
+        # TODO: replace placeholder
+        pass

owrx/dsp.py

@@ -167,6 +167,7 @@ class DspManager(Output, SdrSourceEventClient):
 
     def start(self):
         if self.sdrSource.isAvailable():
+            self.dsp.setBuffer(self.sdrSource.getBuffer())
             self.dsp.start()
         else:
             self.startOnAvailable = True
@@ -209,6 +210,7 @@ class DspManager(Output, SdrSourceEventClient):
         if state is SdrSourceState.RUNNING:
             logger.debug("received STATE_RUNNING, attempting DspSource restart")
             if self.startOnAvailable:
+                self.dsp.setBuffer(self.sdrSource.getBuffer())
                 self.dsp.start()
                 self.startOnAvailable = False
         elif state is SdrSourceState.STOPPING: