Low-level LSL API

LSL is a library designed for streaming time series data across different platforms and programming languages. The core library is primarily written in C++, and bindings are accessible for Python, C#, Java, MATLAB, and Unity, among others. You can find a comprehensive list here.

MNE-LSL provides a reimplementation of the python binding, known as pylsl, within the mne_lsl.lsl module. It introduces additional functionalities to simplify the low-level interaction with LSL streams. Moreover, it enhances the detection of liblsl on your system and can retrieve a compatible version online if necessary. The differences between pylsl and mne_lsl.lsl are detailed here.

import time
import uuid

import numpy as np

from mne_lsl.lsl import (
    StreamInfo,
    StreamInlet,
    StreamOutlet,
    local_clock,
    resolve_streams,
)

Creating a stream

To create a stream, you must first define its properties. This is achieved by creating a StreamInfo object, which specifies the stream’s name, type, source and properties. Convenience methods are available to set the channel properties, including set_channel_info(), which uses a mne.Info object as source.

sinfo = StreamInfo(
    name="my-stream",
    stype="eeg",
    n_channels=3,
    sfreq=1024,
    dtype="float32",
    source_id=uuid.uuid4().hex,
)
sinfo.set_channel_names(["Fz", "Cz", "Oz"])
sinfo.set_channel_types("eeg")
sinfo.set_channel_units("microvolts")

Once the StreamInfo object is created, a StreamOutlet can be instantiated to create the stream.

outlet = StreamOutlet(sinfo)

Discover streams

At this point, the StreamOutlet is available on the network. The function resolve_streams() discovers all available streams on the network. This operation is commonly named the stream resolution.

Note

The stream resolution can be restricted by providing the name, stype, and source_id arguments.

streams = resolve_streams()
assert len(streams) == 1
streams[0]

< sInfo 'my-stream' >
  | Type: eeg
  | Sampling: 1024.0 Hz
  | Number of channels: 3
  | Data type: <class 'numpy.float32'>
  | Source: 4c4c66b90a6441d5890c59a1ab92c3c4

The resolution retrieves only the stream basic properties. The channel properties, stored in the stream description in an XML element tree, are absent from a StreamInfo returned by the resolution function.

assert streams[0].get_channel_names() is None

Connect to a Stream

To connect to a stream, a StreamInlet object must be created using the resolved StreamInfo. Once the stream is opened with open_stream(), the connection is established and both the properties and data become available.

inlet = StreamInlet(streams[0])
inlet.open_stream()
sinfo = inlet.get_sinfo()  # retrieve stream information with all properties

sinfo.get_channel_names()

['Fz', 'Cz', 'Oz']

sinfo.get_channel_types()

['eeg', 'eeg', 'eeg']

sinfo.get_channel_units()

['microvolts', 'microvolts', 'microvolts']

An mne.Info can be obtained directly with get_channel_info(). If the information contained in the XML element tree can not be parsed, default values are used. For instance, the channel names are replaced by the channel numbers similarly to mne.create_info().

sinfo.get_channel_info()

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

Push/Pull operations

For new data to be received, it first need to be pushed on the StreamOutlet. 2 methods are available:

• push_sample() to push an individual sample of shape (n_channels,)

• push_chunk() to push a chunk of samples of shape (n_samples, n_channels)

outlet.push_sample(np.array([1, 2, 3]))

Once pushed, samples become available at the client end. 2 methods are available to retrieve samples:

• pull_sample() to pull an individual sample of shape (n_channels,)

• pull_chunk() to pull a chunk of samples of shape (n_samples, n_channels)

# give a bit of time to the documentation build after the execution of the last cell
time.sleep(0.01)
assert inlet.samples_available == 1
data, ts = inlet.pull_sample()
assert inlet.samples_available == 0
data

array([1., 2., 3.], dtype=float32)

LSL clock

The local system timestamp is retrieved with local_clock(). This local timestamp can be compared with the LSL timestamp from acquired data.

now = local_clock()
print(f"Timestamp of the acquired data: {ts}")
print(f"Current time: {now}")
print(f"Delta: {now - ts} seconds")

Timestamp of the acquired data: 388.442667597
Current time: 388.645211508
Delta: 0.20254391099996383 seconds

Free resources

When you are done with a StreamInlet or StreamOutlet, don’t forget to free the resources they both use.

inlet.close_stream()
del inlet
del outlet

Estimated memory usage: 195 MB