Enable multi-channel acquisition

Author: mesca

README.rst

@@ -44,10 +44,12 @@ Download the `Arduino IDE <https://www.arduino.cc/en/Main/Software>`__ on your c
 
 You're all set!
 
-An example graph can be found `here <https://github.com/timeflux/timeflux_upsidedownlabs/blob/master/examples/uart.yaml>`__.
+An example graph for the default A0 analog port can be found `here <https://github.com/timeflux/timeflux_upsidedownlabs/blob/master/examples/uart.yaml>`__.
 You can run it like this:
 
 ::
 
     $ conda activate timeflux
-    $ timeflux -d examples/uart.yaml
+    $ timeflux -d examples/uart.yaml
+
+For an example of querying multiple ports and naming the channels, see `this graph <https://github.com/timeflux/timeflux_upsidedownlabs/blob/master/examples/channels.yaml>`__.

examples/channels.yaml

@@ -0,0 +1,26 @@
+graphs:
+  - id: demo
+    nodes:
+      - id: data
+        module: timeflux_upsidedownlabs.nodes.driver
+        class: UpsideDownLabs
+        params:
+          rate: 100
+          channels:
+            0: ECG
+            2: EMG1
+            3: EMG2
+            100: EEG # will be removed (invalid pin)
+      - id: display
+        module: timeflux.nodes.debug
+        class: Display
+      - id: ui
+        module: timeflux_ui.nodes.ui
+        class: UI
+    edges:
+      - source: data
+        target: ui:data
+      - source: data
+        target: display
+    rate: 10
+

timeflux_upsidedownlabs/nodes/driver.py

@@ -17,6 +17,9 @@ class UpsideDownLabs(Node):
             e.g. ``COM3`` on Windows;  ``/dev/cu.usbmodem14601`` on MacOS;
             ``/dev/ttyUSB0`` on GNU/Linux.
             Default: autodetect
+        channels (dict): The pin/channel mapping.
+            Keys are pin numbers and values are channel names.
+            Default: {1: "signal"}
         rate (int): The device rate in Hz.
             Default: ``500``.
 
@@ -26,23 +29,32 @@ class UpsideDownLabs(Node):
 
     """
 
-    def __init__(self, port=None, rate=500):
+    def __init__(self, port=None, channels=None, rate=500):
         if not port:
             port = Arduino.AUTODETECT
+        if not channels:
+            channels = {1: "signal"}
+        self.channels = channels
         self.rate = rate
-        self.pin = 0
         self.timestamp = 0
-        self.timestamps = []
-        self.data = []
         self.board = Arduino(port)
         self.meta = {"rate": rate}
         self._lock = Lock()
         self._blink()
-        self.board.analog[self.pin].register_callback(self._callback)
         self.board.samplingOn(1000 / self.rate)
-        self.board.analog[self.pin].enable_reporting()
+        for pin in list(self.channels.keys()):
+            if pin >= len(self.board.analog):
+                self.logger.warning(f"Removing invalid pin {pin}")
+                del self.channels[pin]
+        self._reset_buffer()
+        self._reset_sample()
+        for pin, channel in self.channels.items():
+            # See: https://docs.python-guide.org/writing/gotchas/#late-binding-closures
+            self.board.analog[pin].register_callback(lambda data, channel=channel : self._callback(data, channel))
+            self.board.analog[pin].enable_reporting()
 
     def _blink(self):
+        """Show a cool led animation"""
         pins = [8, 9, 10, 11, 12, 13]
         for i in range(10):
             for pin in pins:
@@ -51,24 +63,48 @@ def _blink(self):
                 self.board.digital[pin].write(0)
             pins.reverse()
 
-    def _callback(self, data):
+    def _callback(self, data, channel):
         """Acquire and cache data."""
         if self.timestamp == 0:
             self.timestamp = time.time()
         self._lock.acquire()
-        # print(f"{self.timestamp:.3f}\t{data}")
-        self.timestamps.append(self.timestamp)
-        self.data.append(data)
+        self.sample["received"] += 1
+        self.sample["data"][channel].append(data)
+        if not self.sample["timestamp"]:
+            self.sample["timestamp"] = self.timestamp
+        if self.sample["received"] == len(self.channels):
+            self._commit_sample()
+            self.timestamp += 1 / self.rate
         self._lock.release()
-        self.timestamp += 1 / self.rate
+
+    def _reset_buffer(self):
+        """Reset the buffer"""
+        self.timestamps = []
+        self.data = {}
+        for channel in self.channels.values():
+            self.data[channel] = []
+
+    def _reset_sample(self):
+        """Reset the sample"""
+        self.sample = {
+            "timestamp": None,
+            "data": { channel: [] for channel in self.channels.values() },
+            "received": 0
+        }
+
+    def _commit_sample(self):
+        """Append the sample"""
+        self.timestamps.append(self.sample["timestamp"])
+        for channel in self.channels.values():
+            self.data[channel] += self.sample["data"][channel]
+        self._reset_sample()
 
     def update(self):
         """Update the node output."""
         self._lock.acquire()
         index = pd.to_datetime(self.timestamps, unit="s")
         self.o.set(self.data, index, meta=self.meta)
-        self.data = []
-        self.timestamps = []
+        self._reset_buffer()
         self._lock.release()
 
     def terminate(self):