update to match pycsdr chaanges

csdr/__init__.py

@@ -35,6 +35,7 @@ from owrx.audio.chopper import AudioChopper
 
 from csdr.pipe import Pipe
 
+from pycsdr.modules import Buffer
 from csdr.chain.demodulator import DemodulatorChain
 from csdr.chain.fm import Fm
 from csdr.chain.am import Am
@@ -49,6 +50,7 @@ class Dsp(DirewolfConfigSubscriber):
     def __init__(self, output: Output):
         self.pycsdr_enabled = True
         self.pycsdr_chain = None
+        self.pycsdr_reader = None
         self.buffer = None
 
         self.samp_rate = 250000
@@ -729,7 +731,10 @@ class Dsp(DirewolfConfigSubscriber):
                 self.set_bpf(self.low_cut, self.high_cut)
                 self.set_offset_freq(self.offset_freq)
                 chain.setInput(self.buffer)
-                self.output.send_output("audio", chain.getOutput().read)
+                outputBuffer = Buffer(chain.getOutputFormat())
+                chain.setWriter(outputBuffer)
+                self.pycsdr_reader = outputBuffer.getReader()
+                self.output.send_output("audio", self.pycsdr_reader.read)
                 return
 
             command_base = " | ".join(chain)
@@ -824,6 +829,8 @@ class Dsp(DirewolfConfigSubscriber):
             if self.pycsdr_enabled and self.pycsdr_chain is not None:
                 self.pycsdr_chain.stop()
                 self.pycsdr_chain = None
+                self.pycsdr_reader.stop()
+                self.pycsdr_reader = None
             if self.process is not None:
                 try:
                     os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)

csdr/chain/__init__.py

@@ -1,56 +1,65 @@
-from pycsdr.modules import Buffer
+from pycsdr.modules import Buffer, Writer
 
 
 class Chain:
     def __init__(self, *workers):
         self.input = None
         self.output = None
+        self.reader = None
         self.workers = list(workers)
-        for i in range(1, len(self.workers)):
+        filtered = self._filterWorkers()
-            self._connect(self.workers[i - 1], self.workers[i])
+        for i in range(1, len(filtered)):
+            self._connect(filtered[i - 1], filtered[i])
+
+    def _filterWorkers(self):
+        return [w for w in self.workers if not isinstance(w, Chain) or not w.empty()]
+
+    def empty(self):
+        return len(self.workers) <= 0
 
     def _connect(self, w1, w2):
-        if isinstance(w1, Chain):
+        writer = Buffer(w1.getOutputFormat())
-            buffer = w1.getOutput()
+        w1.setWriter(writer)
+        if isinstance(w2, Chain):
+            w2.setInput(writer)
         else:
-            buffer = Buffer(w1.getOutputFormat())
+            w2.setReader(writer.getReader())
-            w1.setOutput(buffer)
-        w2.setInput(buffer)
 
     def stop(self):
         for w in self.workers:
             w.stop()
-        self.setInput(None)
+        if self.reader is not None:
-        if self.output is not None:
+            self.reader.stop()
-            self.output.stop()
+            self.reader = None
 
-    def setInput(self, buffer):
+    def setInput(self, buffer: Buffer):
         if self.input == buffer:
             return
         self.input = buffer
         if self.workers:
-            self.workers[0].setInput(buffer)
+            firstWorker = self.workers[0]
+            if isinstance(firstWorker, Chain):
+                firstWorker.setInput(buffer)
+            else:
+                firstWorker.setReader(buffer.getReader())
         else:
             self.output = self.input
 
-    def getOutput(self):
+    def setWriter(self, writer: Writer):
-        if self.output is None:
+        if self.output == writer:
-            if self.workers:
+            return
-                lastWorker = self.workers[-1]
+        self.output = writer
-                if isinstance(lastWorker, Chain):
+        if self.workers:
-                    self.output = lastWorker.getOutput()
+            lastWorker = self.workers[-1]
-                else:
+            lastWorker.setWriter(self.output)
-                    self.output = Buffer(self.getOutputFormat())
+        else:
-                    self.workers[-1].setOutput(self.output)
+            raise BufferError("setOutput on empty chain")
-            else:
-                self.output = self.input
-        return self.output
 
     def getOutputFormat(self):
         if self.workers:
             return self.workers[-1].getOutputFormat()
         else:
-            return self.input.getOutputFormat()
+            raise BufferError("getOutputFormat on empty chain")
 
     def replace(self, index, newWorker):
         if index >= len(self.workers):
@@ -60,36 +69,41 @@ class Chain:
         self.workers[index] = newWorker
 
         if index == 0:
-            newWorker.setInput(self.input)
+            writer = self.input
         else:
             previousWorker = self.workers[index - 1]
-            if isinstance(previousWorker, Chain):
+            writer = Buffer(previousWorker.getOutputFormat())
-                newWorker.setInput(previousWorker.getOutput())
+            previousWorker.setWriter(writer)
+
+        if writer is not None:
+            if isinstance(newWorker, Chain):
+                newWorker.setInput(writer)
             else:
-                buffer = Buffer(previousWorker.getOutputFormat())
+                newWorker.setReader(writer.getReader())
-                previousWorker.setOutput(buffer)
-                newWorker.setInput(buffer)
 
         if index < len(self.workers) - 1:
             nextWorker = self.workers[index + 1]
-            if isinstance(newWorker, Chain):
+            writer = Buffer(newWorker.getOutputFormat())
-                nextWorker.setInput(newWorker.getOutput())
+            newWorker.setWriter(writer)
+            if isinstance(nextWorker, Chain):
+                nextWorker.setInput(writer)
             else:
-                buffer = Buffer(newWorker.getOutputFormat())
+                nextWorker.setReader(writer.getReader())
-                newWorker.setOutput(buffer)
-                nextWorker.setInput(buffer)
         else:
-            newWorker.setOutput(self.output)
+            if self.output is not None:
+                newWorker.setWriter(self.output)
 
     def pump(self, write):
-        output = self.getOutput()
+        if self.output is None:
+            self.setWriter(Buffer(self.getOutputFormat()))
+        self.reader = self.output.getReader()
 
         def copy():
             run = True
             while run:
                 data = None
                 try:
-                    data = output.read()
+                    data = self.reader.read()
                 except ValueError:
                     pass
                 if data is None or (isinstance(data, bytes) and len(data) == 0):

owrx/source/__init__.py

@@ -252,13 +252,13 @@ class SdrSource(ABC):
     def _getTcpSource(self):
         with self.modificationLock:
             if self.tcpSource is None:
-                self.tcpSource = TcpSource(self.port)
+                self.tcpSource = TcpSource(self.port, Format.COMPLEX_FLOAT)
         return self.tcpSource
 
     def getBuffer(self):
         if self.buffer is None:
             self.buffer = Buffer(Format.COMPLEX_FLOAT)
-            self._getTcpSource().setOutput(self.buffer)
+            self._getTcpSource().setWriter(self.buffer)
         return self.buffer
 
     def getCommandValues(self):