[MRG] LSLClient (#6120)

* working on the realtime client

* new base class for rtclient

* updates to the base client

* add import

* first pass at lsl in mne

* UPDATE lsl client, base_client, FIX epochs

* decouple the refactor from the LSLClient implementation

this reverts the attempt to both add an LSLClient and create a BaseClient class. For this upcoming PR, the focus will be adding support to LSL. subsequent PRs can be done to refactor the realtime module

* update to lsl client

at the moment, I am able to connect to a stream.
I need to test whether I am able to read the data from the stream.
opened an issue on pylsl to inquire whether i can both send and receive on same machine for local dev purpose (labstreaminglayer/pylsl#10)

once I can confirm the data is coming in properly, i will add some tests

* Update lsl_client.py

mistakenly tried to create inlets twice. this is the mvp for the LSL

* code cleanup and the beginning of tests for the lslclient

* Update test_lsl_client.py

* starting to remember the pythonic way of coding

been living in the world of javascript. feeling like a wayward youth

* simplify time

* Update lsl_client.py

* address some comment in the pr

change the pattern of checking for the import following common pattern seen in mne

* [FIX] _testing.py: copy pasted but never corrected

* Update test_lsl_client.py

forgot that I changed the class signature. If this doesn't work, I will implement multiprocessing to create the LSL Stream for the test

* testing is very useful

just caught the fact that I didn't change the base class to take the private connect function

* FIX arg in _connect; Implement thread to test

Fix a typo in the LSLClient._connect which led to downstream errors.

* address styling

* replace thread with process for clean exit

* linting complete

* FIX: Add to two other req lists

* FIX: Not on PyPi for Linux

Author: teonbrooks

README.rst

@@ -93,6 +93,7 @@ For full functionality, some functions require:
 - Picard >= 0.3
 - CuPy >= 4.0 (for NVIDIA CUDA acceleration)
 - DIPY >= 0.10.1
+- PyLSL >= 1.13.1
 
 Contributing to MNE-Python
 ^^^^^^^^^^^^^^^^^^^^^^^^^^

doc/python_reference.rst

@@ -948,10 +948,11 @@ Realtime
 .. autosummary::
    :toctree: generated/
 
+   FieldTripClient
+   LSLClient
+   MockRtClient
    RtEpochs
    RtClient
-   MockRtClient
-   FieldTripClient
    StimServer
    StimClient
 

doc/whats_new.rst

@@ -95,6 +95,8 @@ Changelog
 
 - Improved clicking in :meth:`mne.io.Raw.plot` (left click on trace toggles bad, left click on background sets green line, right click anywhere removes green line) by `Clemens Brunner`_
 
+- Add :class:`mne.realtime.LSLClient` for realtime data acquisition with LSL streams of data by `Teon Brooks`_ and `Mainak Jas`_
+
 Bug
 ~~~
 - Fix filtering functions (e.g., :meth:`mne.io.Raw.filter`) to properly take into account the two elements in ``n_pad`` parameter by `Bruno Nicenboim`_

environment.yml

@@ -45,3 +45,4 @@ dependencies:
   - pydocstyle
   - codespell
   - python-picard
+  - pylsl>=1.13.1

mne/realtime/__init__.py

@@ -9,6 +9,7 @@
 
 from .client import RtClient
 from .epochs import RtEpochs
+from .lsl_client import LSLClient
 from .mockclient import MockRtClient
 from .fieldtrip_client import FieldTripClient
 from .stim_server_client import StimServer, StimClient

mne/realtime/base_client.py

@@ -0,0 +1,199 @@
+# Authors: Teon Brooks <teon.brooks@gmail.com>
+#          Mainak Jas <mainakjas@gmail.com>
+#
+# License: BSD (3-clause)
+
+import threading
+import time
+import numpy as np
+
+from ..utils import logger, fill_doc
+
+
+def _buffer_recv_worker(client):
+    """Worker thread that constantly receives buffers."""
+    try:
+        for raw_buffer in client.iter_raw_buffers():
+            client._push_raw_buffer(raw_buffer)
+    except RuntimeError as err:
+        # something is wrong, the server stopped (or something)
+        client._recv_thread = None
+        print('Buffer receive thread stopped: %s' % err)
+
+
+class _BaseClient(object):
+    """Base Realtime Client.
+
+    Parameters
+    ----------
+    info : instance of mne.Info | None
+        The measurement info read in from a file. If None, it is generated from
+        the realtime stream. This method may result in less info than expected.
+    host : str
+        The identifier of the server. IP address, LSL id, or raw filename.
+    port : int | None
+        Port to use for the connection.
+    wait_max : float
+        Maximum time (in seconds) to wait for real-time buffer to start.
+    tmin : float | None
+        Time instant to start receiving buffers. If None, start from the latest
+        samples available.
+    tmax : float
+        Time instant to stop receiving buffers.
+    buffer_size : int
+        Size of each buffer in terms of number of samples.
+    verbose : bool, str, int, or None
+        If not None, override default verbose level (see :func:`mne.verbose`
+        and :ref:`Logging documentation <tut_logging>` for more).
+    """
+
+    def __init__(self, info=None, host='localhost', port=None,
+                 wait_max=10., tmin=None, tmax=np.inf,
+                 buffer_size=1000, verbose=None):  # noqa: D102
+        self.info = info
+        self.host = host
+        self.port = port
+        self.wait_max = wait_max
+        self.tmin = tmin
+        self.tmax = tmax
+        self.buffer_size = buffer_size
+        self.verbose = verbose
+
+    def __enter__(self):  # noqa: D105
+
+        # connect to buffer
+        logger.info("Client: Waiting for server to start")
+        start_time = time.time()
+        while time.time() < (start_time + self.wait_max):
+            try:
+                self._connect()
+                logger.info("Client: Connected")
+                break
+            except Exception:
+                time.sleep(0.1)
+        else:
+            raise RuntimeError('Could not connect to Client.')
+
+        if not self.info:
+            self.info = self._create_info()
+        self._enter_extra()
+
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self._disconnect()
+
+        return self
+
+    @fill_doc
+    def get_data_as_epoch(self, n_samples=1024, picks=None):
+        """Return last n_samples from current time.
+
+        Parameters
+        ----------
+        n_samples : int
+            Number of samples to fetch.
+        %(picks_all)s
+
+        Returns
+        -------
+        epoch : instance of Epochs
+            The samples fetched as an Epochs object.
+
+        See Also
+        --------
+        mne.Epochs.iter_evoked
+        """
+        pass
+
+    def get_measurement_info(self):
+        """Return the measurement info.
+
+        Returns
+        -------
+        self.info : dict
+            The measurement info.
+        """
+        return self.info
+
+    def iter_raw_buffers(self):
+        """Return an iterator over raw buffers."""
+        pass
+
+    def register_receive_callback(self, callback):
+        """Register a raw buffer receive callback.
+
+        Parameters
+        ----------
+        callback : callable
+            The callback. The raw buffer is passed as the first parameter
+            to callback.
+        """
+        if callback not in self._recv_callbacks:
+            self._recv_callbacks.append(callback)
+
+    def start_receive_thread(self, nchan):
+        """Start the receive thread.
+
+        If the measurement has not been started, it will also be started.
+
+        Parameters
+        ----------
+        nchan : int
+            The number of channels in the data.
+        """
+        if self._recv_thread is None:
+
+            self._recv_thread = threading.Thread(target=_buffer_recv_worker,
+                                                 args=(self, ))
+            self._recv_thread.daemon = True
+            self._recv_thread.start()
+
+    def stop_receive_thread(self, stop_measurement=False):
+        """Stop the receive thread.
+
+        Parameters
+        ----------
+        stop_measurement : bool
+            Also stop the measurement.
+        """
+        if self._recv_thread is not None:
+            self._recv_thread.stop()
+            self._recv_thread = None
+
+    def unregister_receive_callback(self, callback):
+        """Unregister a raw buffer receive callback.
+
+        Parameters
+        ----------
+        callback : callable
+            The callback to unregister.
+        """
+        if callback in self._recv_callbacks:
+            self._recv_callbacks.remove(callback)
+
+    def _connect(self):
+        """Connect to client device."""
+        pass
+
+    def _create_info(self):
+        """Create an mne.Info class for connection to client."""
+        pass
+
+    def _disconnect(self):
+        """Disconnect the client device."""
+        pass
+
+    def _enter_extra(self):
+        """Run additional commands in __enter__.
+
+        For system-specific loading and initializing after connect but
+        during the enter.
+
+        """
+        pass
+
+    def _push_raw_buffer(self, raw_buffer):
+        """Push raw buffer to clients using callbacks."""
+        for callback in self._recv_callbacks:
+            callback(raw_buffer)

mne/realtime/fieldtrip_client.py

@@ -107,7 +107,7 @@ def __enter__(self):  # noqa: D105
         else:
             logger.info("FieldTripClient: Header retrieved")
 
-        self.info = self._guess_measurement_info()
+        self.info = self._create_info()
         self.ch_names = self.ft_header.labels
 
         # find start and end samples
@@ -129,7 +129,7 @@ def __enter__(self):  # noqa: D105
     def __exit__(self, type, value, traceback):  # noqa: D105
         self.ft_client.disconnect()
 
-    def _guess_measurement_info(self):
+    def _create_info(self):
         """Create a minimal Info dictionary for epoching, averaging, etc."""
         if self.info is None:
             warn('Info dictionary not provided. Trying to guess it from '

mne/realtime/lsl_client.py

@@ -0,0 +1,120 @@
+# Authors: Teon Brooks <teon.brooks@gmail.com>
+#          Mainak Jas <mainakjas@gmail.com>
+#
+# License: BSD (3-clause)
+
+import numpy as np
+
+from .base_client import _BaseClient
+from ..epochs import EpochsArray
+from ..io.meas_info import create_info
+from ..io.pick import _picks_to_idx, pick_info
+from ..utils import fill_doc, _check_pylsl_installed
+
+
+class LSLClient(_BaseClient):
+    """LSL Realtime Client.
+
+    Parameters
+    ----------
+    info : instance of mne.Info | None
+        The measurement info read in from a file. If None, it is generated from
+        the LSL stream. This method may result in less info than expected. If
+        the channel type is EEG, the `standard_1005` montage is used for
+        electrode location.
+    host : str
+        The LSL identifier of the server.
+    port : int | None
+        Port to use for the connection.
+    wait_max : float
+        Maximum time (in seconds) to wait for real-time buffer to start
+    tmin : float | None
+        Time instant to start receiving buffers. If None, start from the latest
+        samples available.
+    tmax : float
+        Time instant to stop receiving buffers.
+    buffer_size : int
+        Size of each buffer in terms of number of samples.
+    verbose : bool, str, int, or None
+        If not None, override default verbose level (see :func:`mne.verbose`
+        and :ref:`Logging documentation <tut_logging>` for more).
+    """
+
+    @fill_doc
+    def get_data_as_epoch(self, n_samples=1024, picks=None):
+        """Return last n_samples from current time.
+
+        Parameters
+        ----------
+        n_samples : int
+            Number of samples to fetch.
+        %(picks_all)s
+
+        Returns
+        -------
+        epoch : instance of Epochs
+            The samples fetched as an Epochs object.
+
+        See Also
+        --------
+        mne.Epochs.iter_evoked
+        """
+        # set up timeout in case LSL process hang. wait arb 5x expected time
+        wait_time = n_samples * 5. / self.info['sfreq']
+
+        # create an event at the start of the data collection
+        events = np.expand_dims(np.array([0, 1, 1]), axis=0)
+        samples, _ = self.client.pull_chunk(max_samples=n_samples,
+                                            timeout=wait_time)
+        data = np.vstack(samples).T
+
+        picks = _picks_to_idx(self.info, picks, 'all', exclude=())
+        info = pick_info(self.info, picks)
+        return EpochsArray(data[picks][np.newaxis], info, events)
+
+    def iter_raw_buffers(self):
+        """Return an iterator over raw buffers."""
+        pylsl = _check_pylsl_installed(strict=True)
+        inlet = pylsl.StreamInlet(self.client)
+
+        while True:
+            samples, _ = inlet.pull_chunk(max_samples=self.buffer_size)
+
+            yield np.vstack(samples).T
+
+    def _connect(self):
+        pylsl = _check_pylsl_installed(strict=True)
+        stream_info = pylsl.resolve_byprop('source_id', self.host,
+                                           timeout=self.wait_max)[0]
+        self.client = pylsl.StreamInlet(info=stream_info,
+                                        max_buflen=self.buffer_size)
+
+        return self
+
+    def _create_info(self):
+        montage = None
+        sfreq = self.client.info().nominal_srate()
+
+        lsl_info = self.client.info()
+        ch_info = lsl_info.desc().child("channels").child("channel")
+        ch_names = list()
+        ch_types = list()
+        ch_type = lsl_info.type().lower()
+        for k in range(1,  lsl_info.channel_count() + 1):
+            ch_names.append(ch_info.child_value("label") or
+                            '{} {:03d}'.format(ch_type.upper(), k))
+            ch_types.append(ch_info.child_value("type") or ch_type)
+            ch_info = ch_info.next_sibling()
+        if ch_type == "eeg":
+            try:
+                montage = 'standard_1005'
+                info = create_info(ch_names, sfreq, ch_types, montage=montage)
+            except ValueError:
+                info = create_info(ch_names, sfreq, ch_types)
+
+        return info
+
+    def _disconnect(self):
+        self.client.close_stream()
+
+        return self