[MRG] RealTime client refactor (mne-tools#6141)

* Add an example using the LSLClient

n_chan --> n_channels

* Refactor FTClient; Add MockLSLStream, refactor test to use mock stream

* update reference and whats new

* fixing some errors

* update style

* temp

* improvements to the realtime module

currently the test is breaking when it comes to using the RtEpochs object.

* minor fix

* move the RtEpochs testing to separate PR

* cleanup

* fix the way super is called

* updated the MockLSLStream to take raw instance

* add time dilation factor, cleanup

* add more info on lsl identifier

* address ci

* skip running test with multiprocessing on windows

Windows runs into a problem with multiprocessing: 'https://stackoverflow.com/questions/50079165/'

* cleaned up the windows check

* update the pylsl requirement to 1.12

this is compatible across platforms

.circleci/config.yml

@@ -59,6 +59,7 @@ jobs:
             command: |
               pip install --user -q --upgrade pip numpy vtk
               pip install --user --progress-bar off -r requirements.txt
+              pip install --user pylsl
               pip install --user --progress-bar off ipython sphinx_fontawesome sphinx_bootstrap_theme "https://api.github.com/repos/sphinx-gallery/sphinx-gallery/zipball/master" memory_profiler "https://api.github.com/repos/nipy/PySurfer/zipball/master"
 
         - save_cache:

README.rst

@@ -93,7 +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
+- PyLSL >= 1.12
 
 Contributing to MNE-Python
 ^^^^^^^^^^^^^^^^^^^^^^^^^^

doc/python_reference.rst

@@ -954,6 +954,7 @@ Realtime
 
    FieldTripClient
    LSLClient
+   MockLSLStream
    MockRtClient
    RtEpochs
    RtClient

doc/whats_new.rst

@@ -99,6 +99,8 @@ Changelog
 
 - Add option ``ids = None`` in :func:`mne.event.shift_time_events` for considering all events by `Nikolas Chalas`_ and `Joan Massich`_
 
+- Add :class:`mne.realtime.MockLSLStream` to simulate an LSL stream for testing and examples by `Teon Brooks`_
+
 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,4 +45,4 @@ dependencies:
   - pydocstyle
   - codespell
   - python-picard
-#  - pylsl>=1.13.1  Needs https://github.com/labstreaminglayer/liblsl-Python/issues/6
+  - pylsl>=1.12

examples/realtime/plot_lslclient_rt.py

@@ -0,0 +1,55 @@
+"""
+==============================================================
+Plot real-time epoch data with LSL client
+==============================================================
+
+This example demonstrates how to use the LSL client to plot real-time
+collection of event data from an LSL stream.
+For the purposes of demo, a mock LSL stream is constructed. You can
+replace this with the stream of your choice by changing the host id to
+the desired stream.
+
+"""
+# Author: Teon Brooks <[email protected]>
+#
+# License: BSD (3-clause)
+import matplotlib.pyplot as plt
+
+from mne.realtime import LSLClient, MockLSLStream
+from mne.datasets import sample
+from mne.io import read_raw_fif
+
+print(__doc__)
+
+# this is the host id that identifies your stream on LSL
+host = 'mne_stream'
+# this is the max wait time in seconds until client connection
+wait_max = 5
+
+
+# Load a file to stream raw data
+data_path = sample.data_path()
+raw_fname = data_path + '/MEG/sample/sample_audvis_filt-0-40_raw.fif'
+raw = read_raw_fif(raw_fname, preload=True).pick('eeg')
+
+# For this example, let's use the mock LSL stream.
+stream = MockLSLStream(host, raw, 'eeg')
+stream.start()
+
+# Let's observe it
+plt.ion()  # make plot interactive
+_, ax = plt.subplots(1)
+with LSLClient(info=raw.info, host=host, wait_max=wait_max) as client:
+    client_info = client.get_measurement_info()
+    sfreq = int(client_info['sfreq'])
+    print(client_info)
+
+    # let's observe ten seconds of data
+    for ii in range(10):
+        plt.cla()
+        epoch = client.get_data_as_epoch(n_samples=sfreq)
+        epoch.average().plot(axes=ax)
+        plt.pause(1)
+plt.draw()
+# Let's terminate the mock LSL stream
+stream.stop()

mne/realtime/__init__.py

@@ -2,14 +2,16 @@
 
 # Authors: Christoph Dinh <[email protected]>
 #          Martin Luessi <[email protected]>
-#          Mainak Jas <[email protected]>
+#          Mainak Jas <[email protected]>
 #          Matti Hamalainen <[email protected]>
+#          Teon Brooks <[email protected]>
 #
 # License: BSD (3-clause)
 
 from .client import RtClient
 from .epochs import RtEpochs
 from .lsl_client import LSLClient
+from .mock_lsl_stream import MockLSLStream
 from .mockclient import MockRtClient
 from .fieldtrip_client import FieldTripClient
 from .stim_server_client import StimServer, StimClient

mne/realtime/base_client.py

@@ -21,6 +21,7 @@ def _buffer_recv_worker(client):
         print('Buffer receive thread stopped: %s' % err)
 
 
+@fill_doc
 class _BaseClient(object):
     """Base Realtime Client.
 
@@ -42,9 +43,7 @@ class _BaseClient(object):
         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).
+    %(verbose)s
     """
 
     def __init__(self, info=None, host='localhost', port=None,
@@ -58,6 +57,8 @@ def __init__(self, info=None, host='localhost', port=None,
         self.tmax = tmax
         self.buffer_size = buffer_size
         self.verbose = verbose
+        self._recv_thread = None
+        self._recv_callbacks = list()
 
     def __enter__(self):  # noqa: D105
 
@@ -132,6 +133,12 @@ def register_receive_callback(self, callback):
         if callback not in self._recv_callbacks:
             self._recv_callbacks.append(callback)
 
+    def start(self):
+        """Start the client."""
+        self.__enter__()
+
+        return self
+
     def start_receive_thread(self, nchan):
         """Start the receive thread.
 
@@ -149,6 +156,12 @@ def start_receive_thread(self, nchan):
             self._recv_thread.daemon = True
             self._recv_thread.start()
 
+    def stop(self):
+        """Stop the client."""
+        self._disconnect()
+
+        return self
+
     def stop_receive_thread(self, stop_measurement=False):
         """Stop the receive thread.