From 8a8a271c9c8788e2bc2391eea7dd4136bb88f5fc Mon Sep 17 00:00:00 2001
From: Sebastien Bourdeauducq <sb@m-labs.hk>
Date: Mon, 27 Jan 2020 17:24:59 +0800
Subject: [PATCH] hacks to stop SDR buffer overflows

---
 dmi.py     | 32 ++++++++++++++++----------------
 noptica.py | 49 +++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 65 insertions(+), 16 deletions(-)

diff --git a/dmi.py b/dmi.py
index 1d1b254..edc8f18 100644
--- a/dmi.py
+++ b/dmi.py
@@ -1,10 +1,14 @@
 import SoapySDR
 import numpy as np
+import gc
 
-from noptica import InductionHeater, Stabilizer, PositionTracker
+from noptica import *
 
 
 def main():
+    # For this PoC code, a small memory leak is less harmful than random overflows.
+    gc.disable()
+
     freq_sample = 5e6
     freq_base = 1086e6
     bufsize = 4096
@@ -20,31 +24,27 @@ def main():
             sdr.setSampleRate(SoapySDR.SOAPY_SDR_RX, channel, freq_sample)
             sdr.setFrequency(SoapySDR.SOAPY_SDR_RX, channel, freq_base)
 
-        samples_ref = np.array([0]*bufsize, np.complex64)
-        samples_meas = np.array([0]*bufsize, np.complex64)
-        rx_stream = sdr.setupStream(SoapySDR.SOAPY_SDR_RX, SoapySDR.SOAPY_SDR_CF32, [0, 1])
+        buf_sdr = BufferedSDR(sdr, [0, 1], bufsize, 256)
+        buf_sdr.start()
         try:
-            sdr.activateStream(rx_stream)
-            try:
-                stabilizer_throttle = 0
-                while True:
-                    sr = sdr.readStream(rx_stream, [samples_ref, samples_meas], bufsize)
-                    if sr.ret != bufsize:
-                        print("SDR sampling error")
-                        return
+            stabilizer_throttle = 0
+            while True:
+                buffers = buf_sdr.get()
+                try:
+                    samples_ref, samples_meas = buffers
 
                     # We can't update faster than the MHS5200A serial interface
                     stabilizer_throttle += 1
-                    if stabilizer_throttle == 30:
+                    if stabilizer_throttle == 40:
                         stabilizer.input(samples_ref)
                         stabilizer_throttle = 0
 
                     position, leakage = position_tracker.input(samples_ref, samples_meas)
                     print(np.sum(position)/len(position), leakage)
-            finally:
-                sdr.deactivateStream(rx_stream)
+                finally:
+                    buf_sdr.dispose(buffers)
         finally:
-            sdr.closeStream(rx_stream)
+            buf_sdr.stop()
     finally:
         induction.stop()
 
diff --git a/noptica.py b/noptica.py
index b8130bf..f8a32ed 100644
--- a/noptica.py
+++ b/noptica.py
@@ -2,10 +2,59 @@ import serial
 import queue
 import threading
 
+import SoapySDR
 import numpy as np
 from scipy.signal import blackmanharris
 
 
+class BufferedSDR:
+    def __init__(self, sdr, channels, bufsize, nbufs):
+        self.sdr = sdr
+        self.channels = channels
+        self.bufsize = bufsize
+
+        self.stream = None
+        self.terminate = False
+        self.thread = None
+        self.queue = queue.Queue(nbufs)
+        self.available_buffers = queue.Queue(nbufs)
+        for _ in range(nbufs):
+            self.available_buffers.put([np.array([0]*bufsize, np.complex64) for _ in self.channels])
+
+    def start(self):
+        self.stream = self.sdr.setupStream(SoapySDR.SOAPY_SDR_RX, SoapySDR.SOAPY_SDR_CF32, self.channels)
+        try:
+            self.thread = threading.Thread(target=self.thread_target)
+            self.thread.start()
+        except:
+            self.sdr.closeStream(self.stream)
+            raise
+
+    def stop(self):
+        self.terminate = True
+        self.thread.join()
+        self.sdr.closeStream(self.stream)
+
+    def get(self):
+        return self.queue.get()
+
+    def dispose(self, buffers):
+        self.available_buffers.put(buffers)
+
+    def thread_target(self):
+        self.sdr.activateStream(self.stream)
+        try:
+            while not self.terminate:
+                buffers = self.available_buffers.get()
+                sr = self.sdr.readStream(self.stream, buffers, self.bufsize)
+                if sr.ret != self.bufsize:
+                    print("SDR sampling error")
+                    return
+                self.queue.put(buffers)
+        finally:
+            self.sdr.deactivateStream(self.stream)
+
+
 class InductionHeater:
     """Interface to the MHS5200A function generator driving the LC tank"""
     def __init__(self, port, induction_min, induction_max):