Player with annotations

Annotations from a Raw object can be streamed as an event stream by PlayerLSL. The stream will be irregularly sampled, numerical, and of type 'annotations'.

A Annotations contain 3 information:

• the onset of the annotation

• the duration of the annotation

• the description of the annotation

To stream all 3 information, it’s duration-hod encoded along the channels. For instance, consider a Raw object with 3 different Annotations description: 'event1', 'event2', and 'event3'. The event stream will have 3 channels, each corresponding to one of the 3 descriptions. When an annotation is streamed, it’s duration is encoded as the value on its channel while the other channels remain to zero.

import uuid

import matplotlib.patches as mpatches
import numpy as np
from matplotlib import pyplot as plt
from mne import Annotations, create_info
from mne.io import RawArray
from mne.viz import set_browser_backend

from mne_lsl.player import PlayerLSL
from mne_lsl.stream import StreamLSL

annotations = Annotations(
    onset=[1, 2, 3],
    duration=[0.1, 0.2, 0.3],
    description=["event1", "event2", "event3"],
)
annotations

<Annotations | 3 segments: event1 (1), event2 (1), event3 (1)>

With the 3 annotations above, the event stream will stream the following samples:

• at time 1, the annotation 'event1' is pushed. The sample push is array([[0.1, 0, 0]]), of shape (1, 3).

• at time 2, the annotation 'event2' is pushed. The sample push is array([[0, 0.2, 0]]), of shape (1, 3).

• at time 3, the annotation 'event3' is pushed. The sample push is array([[0, 0, 0.3]]), of shape (1, 3).

If more than one annotations are present in the chunk currently pushed, then a chunk is pushed. For instance, if the annotations at time 2 and 3 are pushed at the same time, then the chunk push is array([[0., 0.2., 0.], [0., 0., 0.3]]), of shape (2, 3).

%% Example on mock signal ———————-

Let’s create a mock Raw object with annotations and stream both the signal and the annotations.

data = np.zeros((1, 1000))  # 1 channel, 1000 samples
data[0, 100:200] = 1
data[0, 500:700] = 2
info = create_info(["signal"], 1000, "misc")
raw = RawArray(data, info)
annotations = Annotations(onset=[0.1, 0.5], duration=[0.1, 0.2], description=["1", "2"])
raw.set_annotations(annotations)
set_browser_backend("matplotlib")  # easier to plot with matplotlib in a documentation
raw.plot(scalings=dict(misc=2), show_scrollbars=False, show_scalebars=False)
plt.show()

Now that we have the Raw object, we can stream it with a PlayerLSL object.

Note

Note that forcing annotations=True is not necessary since the PlayerLSL will automatically stream annotations if they are present in the Raw object.

Note

A chunk_size of 1 is needed here or the timestamps ts from the signal and annotations streams are not reliable enough.

source_id = uuid.uuid4().hex
player = PlayerLSL(
    raw,
    chunk_size=1,
    name="tutorial-annots",
    source_id=source_id,
    annotations=True,
).start()

We can now acquire both streams with 2 StreamLSL objects.

stream = StreamLSL(2, name="tutorial-annots", source_id=source_id)
stream.connect(acquisition_delay=0.1, processing_flags="all")
stream.info

	General
	MNE object type	Info
	Measurement date	Unknown
	Participant	Unknown
	Experimenter	Unknown
	Acquisition
	Sampling frequency	1000.00 Hz
	Channels
	misc	1
	Head & sensor digitization	Not available
	Filters
	Highpass	0.00 Hz
	Lowpass	500.00 Hz

stream_annotations = StreamLSL(2, stype="annotations", source_id=source_id)
stream_annotations.connect(acquisition_delay=0.1, processing_flags="all")
stream_annotations.info

	General
	MNE object type	Info
	Measurement date	Unknown
	Participant	Unknown
	Experimenter	Unknown
	Acquisition
	Sampling frequency	0.00 Hz
	Channels
	misc	2
	Head & sensor digitization	Not available
	Filters
	Highpass	0.00 Hz
	Lowpass	0.00 Hz

We can now acquire new samples from both streams and create a matplotlib figure to plot the signal and the annotations in real-time.

if not plt.isinteractive():
    plt.ion()
fig, ax = plt.subplots()
# add legend
colors = ["lightcoral", "lightgreen"]
patches = [
    mpatches.Patch(color=colors[k], label=ch, alpha=0.5)
    for k, ch in enumerate(stream_annotations.ch_names)
]
ax.legend(handles=patches, loc="upper left")
plt.show()

n = 0  # number of annotations
while n <= 10:
    if stream.n_new_samples == 0:
        continue

    data, ts = stream.get_data(winsize=stream.n_new_samples / stream.info["sfreq"])
    ax.plot(ts, data.squeeze(), color="teal")

    if stream_annotations.n_new_samples != 0:
        data_annotations, ts_annotations = stream_annotations.get_data(
            winsize=stream_annotations.n_new_samples
        )
        for sample, time in zip(data_annotations.T, ts_annotations, strict=True):
            k = np.where(sample != 0)[0][0]  # find the annotation
            ax.axvspan(
                time,
                time + sample[k],
                label=stream_annotations.ch_names[k],
                color=colors[k],
                alpha=0.5,
            )
        n += 1

    fig.canvas.draw()
    fig.canvas.flush_events()

Free resources

When you are done with a PlayerLSL, a StreamLSL or a EpochsStream don’t forget to free the resources they use to continuously mock an LSL stream or receive new data from an LSL stream.

stream.disconnect()

<Stream: OFF | tutorial-annots (source: ffb2882f6edf49388b9f4ebaffff1a57)>

stream_annotations.disconnect()

<Stream: OFF | tutorial-annots-annotations (source: ffb2882f6edf49388b9f4ebaffff1a57)>

player.stop()

<Player: tutorial-annots | OFF>

Estimated memory usage: 190 MB