Run recorder in separate thread

saraswati/cli.py

@@ -71,7 +71,7 @@ def validate_channel(ctx, param, value):
     multiple=True,
     callback=validate_channel,
     help="Define channels to record, add multiple times to define more channels. "
-         "The expression to define an audio source should be a GStreamer pipeline element syntax.",
+    "The expression to define an audio source should be a GStreamer pipeline element syntax.",
 )
 spool_opt = click.option(
     "--spool",
@@ -187,4 +187,5 @@ def record(
         recorder.add_channel(name=channel.name, source=channel.source)
 
     # Invoke recording pipelines.
-    recorder.run()
+    recorder.start()
+    recorder.record()

saraswati/model.py

@@ -23,7 +23,7 @@ class SaraswatiSettings:
     chunk_max_files: Optional[int] = 9999
 
     # Where to store the recordings.
-    spool_path: Optional[Path] = None
+    spool_path: Optional[str] = None
     spool_filename_pattern: Optional[
         str
     ] = "recording_{channel}_{timestamp}_{fragment:04d}.mka"

saraswati/recorder.py

@@ -3,6 +3,7 @@
 # Saraswati is a robust, multi-channel audio recording, transmission and storage system
 # (c) 2018-2021 Andreas Motl <andreas@hiveeyes.org>
 # (c) 2019 Diren Senger <diren@diren.de>
+import threading
 from functools import partial
 from typing import List
 
@@ -22,7 +23,7 @@
 logger = logging.getLogger(__name__)
 
 
-class SaraswatiRecorder:
+class SaraswatiRecorder(threading.Thread):
     """
     This implements an audio encoding GStreamer pipeline in Python similar to this one::
 
@@ -41,6 +42,8 @@ class SaraswatiRecorder:
 
     def __init__(self, settings: SaraswatiSettings):
 
+        super().__init__()
+
         logger.info("Setting up audio recorder")
 
         self.settings = settings
@@ -54,16 +57,38 @@ def __init__(self, settings: SaraswatiSettings):
         # list for muxer
         self.muxer = []
 
-        self.setup()
-        self.print_status()
-
-    def setup(self):
         # Setup PyGObject and GStreamer
         Gst.init(None)
 
         # Create main event loop object
         self.mainloop = GLib.MainLoop()
 
+        self.print_status()
+
+    # Invoke the pipeline.
+    def run(self):
+        logger.info("Starting audio recorder")
+        self.mainloop.run()
+
+    def record(self):
+        try:
+            self.play()
+        except Exception as ex:
+            logger.error(f"Recording suspended: {ex}")
+        finally:
+            GLib.timeout_add_seconds(self.SERVICE_TASK_INTERVAL, self.record)
+
+    def play(self):
+        success = False
+        for i, pipeline in enumerate(self.pipelines):
+            (outcome, state, pending) = pipeline.gst.get_state(timeout=Gst.SECOND)
+            if state != Gst.State.PLAYING:
+                logger.info(f"Starting pipeline: {pipeline}")
+                pipeline.gst.set_state(Gst.State.PLAYING)
+                success = True
+        if success:
+            logger.info("Started recording")
+
     @property
     def output_location(self):
         """
@@ -131,14 +156,6 @@ def add_channel(self, name: str, source: str):
 
             current_muxer.connect(signal_name, signal_callback, user_data)
 
-    # Running the shit
-    def run(self):
-        logger.info("Starting audio recorder")
-        for i, pipeline in enumerate(self.pipelines):
-            logger.info(f"Starting pipeline: {pipeline}")
-            pipeline.gst.set_state(Gst.State.PLAYING)
-        self.mainloop.run()
-
     # Bus message handler
     def on_message(self, pipeline: Pipeline, bus, message):
 

tests/test_recorder.py

@@ -5,7 +5,7 @@
 from saraswati.util import setup_logging
 
 
-def test_spike():
+def test_dryrun():
 
     # Setup logging.
     setup_logging(level=logging.DEBUG)
@@ -19,3 +19,5 @@ def test_spike():
     # Run a basic pipeline test.
     recorder.add_channel(name="channel1", source="audiotestsrc")
     recorder.add_channel(name="channel2", source="audiotestsrc")
+
+    # recorder.play()