Starting & stopping streams from within microservice

microservice.py

@@ -110,15 +110,48 @@ def configure(radio, frequency):
     else:
         return "Radio not connected", 400
 
-'''
+
 @app.route('/radio/<radio>/start')
 def start_stream(radio):
-    pass
+    """Tune the radio to a frequency.
+    You must connect to the radio before attempting to configure it.
+    The radio must be configured before streaming can be started.
+    ---
+    parameters:
+      - name: radio
+        description: Radio device driver name, or serial number.
+        in: path
+        type: string
+        required: true
+    """
+    try:
+        radios[radio].start_stream()
+        return "", 200
+    except Exception as e:
+        return str(e), 400
 
 @app.route('/radio/<radio>/end')
 def end_stream(radio):
-    pass
+    """Tune the radio to a frequency.
-'''
+    You must connect to the radio before attempting to configure it.
+    The radio must be configured before streaming can be started.
+    ---
+    parameters:
+      - name: radio
+        description: Radio device driver name, or serial number.
+        in: path
+        type: string
+        required: true
+    """
+    try:
+        radios[radio].end_stream()
+        return "", 200
+    except Exception as e:
+        return str(e), 400
+
 
 if __name__ == '__main__':
-    app.run(debug=True)
+    app.run(
+        threaded=True,
+        debug=True
+    )

radio.py

@@ -1,12 +1,21 @@
+from threading import Thread
 import SoapySDR as soapy
 import prefixed
+from formats import *
+import sys
 
 
 class Radio:
+    FORMAT = 'CS16'
+    SAMPLES = 8192
+
     def __init__(self, device_info):
         self.device_info = device_info
         self.stream = None
         self.buffer = None
+        self.port = None
+        self.run = False
+        self.thread = None
 
         self.device = soapy.Device(device_info)
         if self.device is None:
@@ -31,3 +40,42 @@ class Radio:
             'bandwidth': self.device.getBandwidth(soapy.SOAPY_SDR_RX, 0),
             'gain-mode': 'auto' if self.device.getGainMode(soapy.SOAPY_SDR_RX, 0) else 'manual',
         }
+
+
+    def start_stream(self):
+        if self.thread:
+            raise RuntimeError('Stream thread is already running')
+
+        self.buffer = np.array([0] * Radio.SAMPLES * 2, TYPES[Radio.FORMAT])
+        self.stream = self.device.setupStream(soapy.SOAPY_SDR_RX, FORMATS[Radio.FORMAT])
+        self.device.activateStream(self.stream)
+
+        self.run = True
+        self.thread = Thread(target=self._stream_thread, daemon=True, args=())
+        self.thread.start()
+
+
+    def _stream_thread(self):
+        while self.run:
+            result = self.device.readStream(self.stream, [self.buffer], Radio.SAMPLES)
+
+            if result.ret < 1:
+                print('Stream read failed, exiting.', file=sys.stderr)
+                self.keep_going = False
+
+        self._cleanup_stream()
+
+
+    def end_stream(self):
+        if self.thread is None:
+            raise RuntimeError('No stream thread to terminate')
+
+        self.run = False
+        self.thread.join()
+        self.thread = None
+
+
+    def _cleanup_stream(self):
+        #shutdown the stream
+        self.device.deactivateStream(self.stream)
+        self.device.closeStream(self.stream)