From 0ef770435343872425bf1909121d41072ada3ac3 Mon Sep 17 00:00:00 2001
From: mesca <github@clisson.org>
Date: Mon, 6 Nov 2023 17:25:40 +0530
Subject: [PATCH] Add audio input node

---
 examples/passthrough.yaml      | 21 +++++++++++
 timeflux_audio/nodes/device.py | 69 +++++++++++++++++++++++++++++++++-
 2 files changed, 88 insertions(+), 2 deletions(-)
 create mode 100644 examples/passthrough.yaml

diff --git a/examples/passthrough.yaml b/examples/passthrough.yaml
new file mode 100644
index 0000000..4860ef0
--- /dev/null
+++ b/examples/passthrough.yaml
@@ -0,0 +1,21 @@
+graphs:
+  - nodes:
+      - id: input
+        module: timeflux_audio.nodes.device
+        class: Input
+      - id: ui
+        module: timeflux_ui.nodes.ui
+        class: UI
+        params:
+          settings:
+            monitor:
+              lineWidth: 1
+      - id: output
+        module: timeflux_audio.nodes.device
+        class: Output
+    edges:
+      - source: input
+        target: ui:audio
+      - source: input
+        target: output
+    rate: 10
diff --git a/timeflux_audio/nodes/device.py b/timeflux_audio/nodes/device.py
index 46626b3..8a0b0bd 100644
--- a/timeflux_audio/nodes/device.py
+++ b/timeflux_audio/nodes/device.py
@@ -2,11 +2,73 @@
 
 import time
 import numpy as np
+import pandas as pd
 import sounddevice as sd
 from threading import Thread, Lock
 from timeflux.core.node import Node
 
 
+class Input(Node):
+    """Audio input.
+
+    Attributes:
+        o (Port): Default output, provides DataFrame.
+
+    Args:
+        device (int|string): input device (numeric or string ID).
+            If none specified, will use the system default. Default: ``None``.
+
+    Example:
+        .. literalinclude:: /../examples/passthrough.yaml
+           :language: yaml
+
+    """
+
+    def __init__(self, device=None):
+        self.device = device
+        self.logger.info(f"Available audio interfaces:\n{sd.query_devices()}")
+        self._data = np.empty((0, 1))
+        self._running = True
+        self._lock = Lock()
+        self._thread = Thread(target=self._loop).start()
+
+    def _callback(self, indata, frames, time, status):
+        if status:
+            self.logger.warning(status)
+        size = indata.shape[0]
+        if size > 0:
+            self._lock.acquire()
+            self._data = np.vstack((self._data, indata))
+            self._stop = pd.Timestamp.now(tz="UTC")
+            self._lock.release()
+
+    def _loop(self):
+        samplerate = sd.query_devices(self.device, "input")["default_samplerate"]
+        self.meta = {"rate": samplerate}
+        self._delta = 1 / samplerate
+        with sd.InputStream(
+            device=self.device,
+            channels=1,
+            callback=self._callback,
+            samplerate=samplerate,
+        ):
+            while self._running:
+                sd.sleep(1)
+
+    def update(self):
+        self._lock.acquire()
+        if self._data.shape[0] > 0:
+            periods = self._data.shape[0]
+            start = self._stop - pd.Timedelta(self._delta * periods, "s")
+            timestamps = pd.date_range(start=start, end=self._stop, periods=periods)
+            self.o.set(self._data, timestamps, meta=self.meta)
+            self._data = np.empty((0, 1))
+        self._lock.release()
+
+    def terminate(self):
+        self._running = False
+
+
 class Output(Node):
     """Audio output.
 
@@ -16,14 +78,17 @@ class Output(Node):
     Args:
         device (int|string): output device (numeric or string ID).
             If none specified, will use the system default. Default: ``None``.
+        amplitude (float): audio volume.
+            Default: 1
 
     Example:
         .. literalinclude:: /../examples/sine.yaml
            :language: yaml
     """
 
-    def __init__(self, device=None):
+    def __init__(self, device=None, amplitude=1):
         self.device = device
+        self.amplitude = amplitude
         self.logger.info(f"Available audio interfaces:\n{sd.query_devices()}")
         self._data = np.empty((0, 1))
         self._running = True
@@ -44,7 +109,6 @@ def _callback(self, outdata, frames, time, status):
 
     def _loop(self):
         samplerate = sd.query_devices(self.device, "output")["default_samplerate"]
-        self._start_idx = 0
         with sd.OutputStream(
             device=self.device,
             channels=1,
@@ -56,6 +120,7 @@ def _loop(self):
 
     def update(self):
         if self.i.ready():
+            self.i.data *= self.amplitude
             self._lock.acquire()
             self._data = np.vstack((self._data, self.i.data.values))
             self._lock.release()