DataHub: Difference between revisions

From TSG Doc
Jump to navigation Jump to search
← Older edit
No edit summary
mNo edit summary
 
(8 intermediate revisions by 2 users not shown)
Line 1: Line 1:
{{Infobox software
{{Infobox software
| name                  = DataHub
| name                  = Lab Streaming Layer
| developer              = Christian Kothe; Chadwick Boulay.
| developer              = Christian Kothe; Chadwick Boulay.
| status                = -in development-
| status                = -in development-
Line 13: Line 13:
{{Infobox tsg
{{Infobox tsg
   | downloads          = {{bulleted list
   | downloads          = {{bulleted list
       | [https://surfdrive.surf.nl/files/index.php/s/zSpGD53mDQGuPKW viewer & streamer]
       | {{Surfdrive|https://surfdrive.surf.nl/s/GLSnbBaYsbEWjPX|Viewer & Streamer}}
     }}
     }}
   | manuals            = {{bulleted list
   | manuals            = {{bulleted list
Line 21: Line 21:
   }}
   }}
   | templates            = {{bulleted list
   | templates            = {{bulleted list
       | [https://surfdrive.surf.nl/files/index.php/s/qggfMMKsnUIDO0k example scripts (zip)]
       | {{Surfdrive|https://surfdrive.surf.nl/s/gPenWNAqq8AXa7x|Example Scripts}}
   }}
   }}
}}
}}
Line 39: Line 39:


===Versions===
===Versions===
The TSG uses the version 1.15.0. Open a command and type the following to find the version used:
Pylsl version 1.16.2 is installed on our [[Lab Computer]]s. Open a command and type the following to find the version used:


<code style="background-color:#000; color:#fff; padding:1px 3px;">c:>python</code><br/>
<code style="background-color:#000; color:#fff; padding:1px 3px;">c:>python</code><br/>
Line 53: Line 53:


import time
import time
import numpy as np
import random
import keyboard
import keyboard
import random
from pylsl import StreamInfo, StreamOutlet
from pylsl import StreamInfo, StreamOutlet, local_clock
from LSLsettings import MarkerStreamSettings, DynamicMultiChannelStreamSettings


print(f"Setting up streams...")
from pylsl import local_clock
marker_settings = MarkerStreamSettings()


# info marker StreamInfo, outlet
# --- Setup Marker Stream ---
print("Setting up marker stream...")
marker_info = StreamInfo(
marker_info = StreamInfo(
     name=marker_settings.stream_name,
     name="MarkerStream",
     type=marker_settings.stream_type,
     type="Markers",
     channel_count=marker_settings.channel_count,
     channel_count=1,
     nominal_srate=marker_settings.sample_rate,
     nominal_srate=0, # Irregular sampling
     channel_format=marker_settings.channel_format,
     channel_format="string",
     source_id=marker_settings.source_id
     source_id="MarkerSource"
)
)
marker_outlet = StreamOutlet(marker_info, chunk_size=marker_settings.push_chunk_size)
marker_outlet = StreamOutlet(marker_info, chunk_size=1)


data_settings = DynamicMultiChannelStreamSettings(
# --- Setup Data Stream ---
    n_channels=4,
print("Setting up data stream...")
     stream_name="TestStream",
data_info = StreamInfo(
     stream_type="TestData",
     name="TestStream",
     sample_rate=100,
     type="TestData",
     channel_names=["A", "B", "C", "D"]
     channel_count=4,
    nominal_srate=100,
     channel_format="float32",
    source_id="TestStream_Source"
)
)


# info data StreamInfo, outlet
channels = data_info.desc().append_child("channels")
data_info = StreamInfo(
for name in ["A", "B", "C", "D"]:
    name=data_settings.stream_name,
     chan = channels.append_child("channel")
    type=data_settings.stream_type,
     chan.append_child_value("name", name)
     channel_count=data_settings.channel_count,
     chan.append_child_value("unit", "unitless")
     nominal_srate=data_settings.sample_rate,
     chan.append_child_value("type", "TestData")
     channel_format=data_settings.channel_format,
     source_id=data_settings.source_id
)
data_outlet = StreamOutlet(data_info, chunk_size=data_settings.push_chunk_size)


running = True
data_outlet = StreamOutlet(data_info, chunk_size=1)
marker_outlet.push_sample(["Start"])
buffer_dtype = object if data_settings.channel_format == "string" else float
buffer_in = np.array(data_settings.default_buffer, dtype=buffer_dtype)


# --- Start Streaming ---
print("Start streaming... Press 'q' to stop.")
print("Start streaming... Press 'q' to stop.")
marker_outlet.push_sample(["Start"])


while not(keyboard.is_pressed('q')):
while not keyboard.is_pressed('q'):
 
     random_numbers = [random.randint(1, 2) for _ in range(4)]
     random_numbers = [random.randint(0, 1) for _ in range(4)]
    random_numbers = [num + 1 for num in random_numbers]
     data_outlet.push_sample(random_numbers)
     data_outlet.push_sample(random_numbers)
 
     time.sleep(0.01)
     time.sleep(data_settings.push_interval)


marker_outlet.push_sample(["Stop"])
marker_outlet.push_sample(["Stop"])
 
print("Streaming stopped.")
</syntaxhighlight>
 
the settings.py file for LSL:
<syntaxhighlight lang="python" line>
# settings.py
 
class BaseSettings:
    """Basisklasse met standaardinstellingen voor een LSL stream."""
    def __init__(self):
        self.stream_name = "BaseStream"
        self.stream_type = "Unknown"
        self.channel_count = 1
        self.sample_rate = 0
        self.channel_format = "float32"
        self.source_id = "BaseSource"
 
        self.channel_names = []
        self.push_chunk_size = 1
        self.push_interval = 0.001
 
        self.use_random_data = False
        self.dummy_string = "Test"
        self.random_string_options = ["A", "B", "C"]
 
        self.verbose = True
        self.quit_method = "psychopy"
        self.user_stop_message = "Press ENTER/RETURN to stop acquisition."
 
        self.default_buffer = ["Test"]
 
        # the scope is not implemented yet
        self.scope = "lan"
 
    def get_stream_options(self):
        if self.scope in ["local", "lan", "internet"]:
            return dict()
        else:
            raise ValueError(f"Unknown scope: {self.scope}")
 
 
class DynamicMultiChannelStreamSettings(BaseSettings):
    def __init__(
        self,
        n_channels=8,
        stream_name="DynamicStream",
        stream_type="EEG",
        sample_rate=256,
        channel_format="float32",
        channel_names=None
    ):
        super().__init__()
        self.stream_name = stream_name
        self.stream_type = stream_type
        self.channel_count = n_channels
        self.sample_rate = sample_rate
        self.channel_format = channel_format
        self.source_id = f"{stream_name}_Source"
 
        if channel_names is not None:
            if len(channel_names) != n_channels:
                raise ValueError(f"Number of channel names ({len(channel_names)}) does not match n_channels ({n_channels})")
            self.channel_names = channel_names
        else:
            self.channel_names = [f"Chan{i+1}" for i in range(n_channels)]
 
        self.default_buffer = [0.0 for _ in range(n_channels)]
 
class MarkerStreamSettings(BaseSettings):
    def __init__(self):
        super().__init__()
        self.stream_name = "MarkerStream"
        self.stream_type = "Markers"
        self.channel_count = 1
        self.sample_rate = 0  # 0 Hz = irregular sampling
        self.channel_format = "string"
        self.source_id = "MarkerSource"
 
        self.channel_names = []  # niet nodig voor markers
        self.default_buffer = ["TestMarker"]
</syntaxhighlight>
</syntaxhighlight>


Latest revision as of 15:24, 11 August 2026

Lab Streaming Layer
Developer(s)Christian Kothe; Chadwick Boulay.
Development status-in development-
Written inC, C++, Python, Java, C#, MATLAB
Operating systemWindows, Linux, MacOS, Android, iOS
TypeData collection
LicenseOpen source
WebsiteLSL webpage

Resources

DataHub
Downloads
Manuals
Templates

The DataHub Makes use of the lab streaming layer. The lab streaming layer (LSL) is a system for the unified collection of measurement time series in research experiments that handles both the networking, time-synchronization, (near-) real-time access as well as optionally the centralized collection, viewing and disk recording of the data.


Installation

Our support for LSL is mainly done in python. Download python here: Please choose a 64 bit version.. Run the installer and make sure to add Python to the file path (it's an option in the installer). Open a command prompt, start with upgrading the pip installer by typing:
c:>python -m pip install --upgrade pip
Then:
c:>pip install pylsl

more info: cross platform pylsl

Versions

Pylsl version 1.16.2 is installed on our Lab Computers. Open a command and type the following to find the version used:

c:>python
>>> import pylsl
>>> print(pylsl.__version__)

Usage

Python

Example demonstrating how to send LSL stream and marker data: 

#!/usr/bin/env python3.10
# -*- coding: utf-8 -*-

import time
import random
import keyboard
from pylsl import StreamInfo, StreamOutlet

from pylsl import local_clock

# --- Setup Marker Stream ---
print("Setting up marker stream...")
marker_info = StreamInfo(
    name="MarkerStream",
    type="Markers",
    channel_count=1,
    nominal_srate=0,  # Irregular sampling
    channel_format="string",
    source_id="MarkerSource"
)
marker_outlet = StreamOutlet(marker_info, chunk_size=1)

# --- Setup Data Stream ---
print("Setting up data stream...")
data_info = StreamInfo(
    name="TestStream",
    type="TestData",
    channel_count=4,
    nominal_srate=100,
    channel_format="float32",
    source_id="TestStream_Source"
)

channels = data_info.desc().append_child("channels")
for name in ["A", "B", "C", "D"]:
    chan = channels.append_child("channel")
    chan.append_child_value("name", name)
    chan.append_child_value("unit", "unitless")
    chan.append_child_value("type", "TestData")

data_outlet = StreamOutlet(data_info, chunk_size=1)

# --- Start Streaming ---
print("Start streaming... Press 'q' to stop.")
marker_outlet.push_sample(["Start"])

while not keyboard.is_pressed('q'):
    random_numbers = [random.randint(1, 2) for _ in range(4)]
    data_outlet.push_sample(random_numbers)
    time.sleep(0.01)

marker_outlet.push_sample(["Stop"])
print("Streaming stopped.")

Matlab

Please, read the instructions on the GitHub labstreaminglayer website (https://github.com/labstreaminglayer/liblsl-Matlab) on how to prepare Matlab to work with LSL. You can either use the latest release for your Matlab version, or if that doesn't workout well, build it from the source files. Make sure to add the liblsl-Matlab folder to your path recursively to make it available to your own scripts.

A short example for sending lsl streaming data: 

%% instantiate the library
disp('Loading library...');
lib = lsl_loadlib();

% make a new stream outlet
disp('Creating a new streaminfo...');
info = lsl_streaminfo(lib,'BioSemi','EEG',8,100,'cf_float32','sdfwerr32432');

disp('Opening an outlet...');
outlet = lsl_outlet(info);

% send data into the outlet, sample by sample
disp('Now transmitting data...');
while true
    outlet.push_sample(randn(8,1));
    pause(0.01);
end

A short example for receiving lsl streaming data: 

%% instantiate the library
disp('Loading the library...');
lib = lsl_loadlib();

% resolve a stream...
disp('Resolving an EEG stream...');
result = {};
while isempty(result)
    result = lsl_resolve_byprop(lib,'type','EEG'); end

% create a new inlet
disp('Opening an inlet...');
inlet = lsl_inlet(result{1});

disp('Now receiving data...');
while true
    % get data from the inlet
    [vec,ts] = inlet.pull_sample();
    % and display it
    fprintf('%.2f\t',vec);
    fprintf('%.5f\n',ts);
end
Retrieved from "https://tsgdoc.socsci.ru.nl/index.php?title=DataHub&oldid=6752"