neural_data_simulator.streamer

A utility to read data from a file and output to an LSL stream.

A streamer is useful for replaying pre-recorded data in real-time.

Currently, this streamer package hosts the streamers.LSLStreamer class, which can be used to output neural_data_simulator.core.samples.Samples to an LSL outlet.

neural_data_simulator.streamer.run_streamer

Script that starts the streamer.

The streamer default configuration is located in NDS_HOME/settings_streamer.yaml (see neural_data_simulator.scripts.post_install_config). The script can use different config file specified via the --settings-path argument.

Upon start, the streamer expects to read data from a file and output to an LSL outlet. By default, a sample behavior data file will be downloaded by the neural_data_simulator.scripts.post_install_config script, so the streamer should be able to run without any additional configuration. If the input file cannot be found, the streamer will not be able to start.

class neural_data_simulator.streamer.run_streamer.StreamGroup(streams_configs: List[StreamConfig])[source]

Bases: object

Utility class for managing a list of streams.

__init__(streams_configs: List[StreamConfig])[source]: Create a new instance.

connect()[source]: Connect all streams.

disconnect()[source]: Disconnect all streams.

open_connection() → Iterator[None][source]

Open a managed connection.

The connection is released after it is consumed.

neural_data_simulator.streamer.run_streamer.load_blackrock_file(filepath: Path, output_settings: LSLSimplifiedOutputModel) → Tuple[List[StreamConfig], List[Samples]][source]: Parse streams from a Blackrock Neurotech file.

neural_data_simulator.streamer.run_streamer.run()[source]: Load the configuration and start the streamer.

neural_data_simulator.streamer.settings

Settings schema for the streamer.

class neural_data_simulator.streamer.settings.StreamerInputType(value)[source]

Bases: str, Enum

Possible types for the streamer input.

NPZ = 'npz'

Blackrock = 'blackrock'

class neural_data_simulator.streamer.settings.LSLSimplifiedOutputModel(*, channel_format: LSLChannelFormatType, instrument: _Instrument)[source]

Bases: BaseModel

Settings for all LSL outlets.

channel_format: LSLChannelFormatType

instrument: _Instrument

class neural_data_simulator.streamer.settings.Streamer(*, blackrock: Optional[Blackrock] = None, npz: Optional[NPZ] = None, input_type: StreamerInputType, lsl_chunk_frequency: float, stream_indefinitely: bool)[source]

Bases: BaseModel

Settings specific to the streamer.

class NPZ(*, output: Output, input: Input)[source]

Bases: BaseModel

Settings for streaming from a numpy archive file (.npz).

class Output(*, sampling_rate: float, n_channels: int, lsl: LSLOutputModel)[source]

Bases: BaseModel

Settings for outputting to LSL.

sampling_rate: float

n_channels: int

lsl: LSLOutputModel

class Input(*, file: Path, timestamps_array_name: str, data_array_name: str)[source]

Bases: BaseModel

Settings for reading in from a .npz file.

file: Path

timestamps_array_name: str

data_array_name: str

output: Output

input: Input

class Blackrock(*, output: Output, input: Input)[source]

Bases: BaseModel

Settings for streaming from Blackrock Neurotech files.

class Output(*, lsl: LSLSimplifiedOutputModel)[source]

Bases: BaseModel

Settings for outputting to LSL.

lsl: LSLSimplifiedOutputModel

class Input(*, file: Path)[source]

Bases: BaseModel

Settings for reading in from a Blackrock Neurotech file.

file: Path

output: Output

input: Input

blackrock: Optional[Blackrock]

npz: Optional[NPZ]

input_type: StreamerInputType

lsl_chunk_frequency: float

stream_indefinitely: bool

neural_data_simulator.streamer.streamers

Stream data over LSL.

class neural_data_simulator.streamer.streamers.Stream(output: LSLOutputDevice, samples: Samples, is_finished: bool = False, read_cursor: int = 0)[source]

Bases: object

Stream information.

output: LSLOutputDevice: The output device to stream data to.

samples: Samples: The data with timestamps to stream.

is_finished: bool = False: Flag to indicate if the stream has finished.

read_cursor: int = 0: The current read cursor position.

property length

The length of the data to stream.

Returns: The number of data samples to stream.

__init__(output: LSLOutputDevice, samples: Samples, is_finished: bool = False, read_cursor: int = 0) → None

class neural_data_simulator.streamer.streamers.LSLStreamer(outputs: List[LSLOutputDevice], samples: List[Samples], lsl_chunk_frequency: float, stream_indefinitely: bool)[source]

Bases: object

Streamer class that can be used to send samples through an LSL stream.

A streamer class that takes a neural_data_simulator.core.samples.Samples dataclass with timestamps and behavior data and stream it through LSL. Following the timestamps provided, the data is streamed to simulate a real-time data acquisition.

__init__(outputs: List[LSLOutputDevice], samples: List[Samples], lsl_chunk_frequency: float, stream_indefinitely: bool) → None[source]

Initialize an LSLStreamer class.

Parameters

outputs – A list of LSL outputs to stream data out.
samples – A list of Samples representing the data to stream.
lsl_chunk_frequency – How often to send data to LSL outlets.
stream_indefinitely – Set to True to continue streaming from the beginning after reaching the end of data.

stream()[source]

Stream samples.

Stream all samples to all the outputs one output at a time. Irregular streams (sample rate is 0) are streamed as samples to LSL while regular streams are streamed as chunk.