neural_data_simulator.core.inputs

Inputs API and implementation.

neural_data_simulator.core.inputs.api

A collection of inputs that can be used by NDS.

class neural_data_simulator.core.inputs.api.Input[source]

Bases: ABC

Represents an input that can be used to consume data from.

This can be an interface for a joystick, a behavior data generator, a data streamer that loads data from disk, etc. Each read should return all newly available data since the last read call.

abstract read() → Samples[source]: Read available data.

abstract connect() → None[source]: Connect to input.

disconnect() → None[source]: Disconnect from input. The default implementation does nothing.

class neural_data_simulator.core.inputs.api.SpikeRateInput(*args, **kwargs)[source]

Bases: Protocol

An abstract input that can be used to read spike rates.

A python protocol (PEP-544) works in a similar way to an abstract class. The __init__() method of this protocol should never be called as protocols are not meant to be instantiated. An __init__() method may be defined in a concrete implementation of this protocol if needed.

property channel_count: int

Get the number of input channels.

Returns: The input channel count.

read() → Optional[ndarray][source]

Read spike rates, one per channel.

Note: above user can call Input.read() but just return the latest Sample

Returns: An array of spike rates with shape (n_units,) or None if no samples are available.

__init__(*args, **kwargs)

neural_data_simulator.core.inputs.lsl_input

Input implementation for LSL streams.

class neural_data_simulator.core.inputs.lsl_input.StreamInfo(lsl_stream_info: dataclasses.InitVar[StreamInfo])[source]

Bases: object

Selected advertised properties of an LSL stream.

name: str: Name of the LSL stream.

sample_rate: float: Advertised sample rate of the LSL stream.

channel_count: int: Number of channels in the LSL stream.

lsl_stream_info: dataclasses.InitVar[StreamInfo]: pylsl stream info object.

__init__(lsl_stream_info: dataclasses.InitVar[StreamInfo]) → None

class neural_data_simulator.core.inputs.lsl_input.LSLInput(stream_name: str, connection_timeout: float = 60.0, resolve_streams_wait_time: float = 1.0)[source]

Bases: Input

Represents an LSL Inlet stream for behavior data.

__init__(stream_name: str, connection_timeout: float = 60.0, resolve_streams_wait_time: float = 1.0)[source]

Initialize LSLInput class.

Parameters

stream_name – Name of the LSL stream to retrieve data from.
connection_timeout – Maximum time for attempting a connection to an LSL input stream.
resolve_streams_wait_time – Maximum waiting time to get the list of available streams. Should be bigger than 0.5 to ensure all streams are returned.

get_info() → StreamInfo[source]

Get information about the LSL stream.

If the stream is not connected, it will try to resolve the stream and return the information.

Returns: LSL stream properties.
Raises: ValueError – If the stream is not found.

read() → Samples[source]

Read available data from the inlet as a samples.

Returns: neural_data_simulator.samples.Samples dataclass with timestamps and data read from the LSL StreamInlet. If no data is available, an empty Samples is returned.
Raises: ValueError – LSL StreamInlet is not connected. connect should be called before read.

set_connection_timeout(timeout: float) → None[source]

Set the maximum time that the inlet search for the desired LSL stream.

Parameters: timeout – Maximum time to wait in seconds.
Raises: ValueError – if timeout equals or less than 0.

connect()[source]: Connect to the LSL Inlet stream.

disconnect()[source]: Disconnect from the LSL Inlet stream.

class neural_data_simulator.core.inputs.lsl_input.LSLSpikeRateInputAdapter(lsl_input: LSLInput)[source]

Bases: SpikeRateInput

Reads spike rates from an LSL input.

property channel_count: int

Get the LSL stream channel count.

Returns: The input channel count.

__init__(lsl_input: LSLInput)[source]

Create an adapter for a given LSL input.

Parameters: lsl_input – The LSL input to adapt.

connect()[source]: Connect to the LSL input stream.

read() → Optional[ndarray][source]

Connect to the LSL input stream.

Returns: An array of spike rates with shape (n_units,) or None if no samples are available.

neural_data_simulator.core.inputs.samples_input

Input implementation for neural_data_simulator.samples.Samples.

class neural_data_simulator.core.inputs.samples_input.SamplesInput(input_samples: Samples)[source]

Bases: Input

An input object based on a neural_data_simulator.samples.Samples.

The underlying samples dataclass will have its timestamps modified to be in reference to when the first read was made from this class, simulating the appearance of data being collected in real-time. Alternatively, the function set_reference_time_to_now can be called prior to the first read of the data to use that as a reference time.

A timer is synced between the reference time and the first timestamp in the input samples. Any calls to the read function will calculate the current time in reference to the synced timer and return the appropriate samples.

__init__(input_samples: Samples) → None[source]

Initialize the SamplesInput class.

Parameters: input_samples – Dataclass containing timestamps and behavior data.

set_reference_time_to_now()[source]: Set current time as starting time for data stream.

read() → Samples[source]

Get new samples from the time of last read.

If first call to read samples will be read since the call to set_reference_time_to_now. If set_reference_time_to_now was not previously called, it will be called.

Returns: neural_data_simulator.samples.Samples dataclass with timestamps and data available since last read call.

connect() → None[source]: No action required during connect for this class.

neural_data_simulator.core.inputs.testing_input

Example input classes for testing.

class neural_data_simulator.core.inputs.testing_input.SpikeRateTestingInput(n_channels: int, n_units: int)[source]

Bases: SpikeRateInput

A constant spike rate input that can be used for testing.

Generates spike rates so that spikes are more likely to happen on channels of a higher order and less likely on channels of a lower order. The spike rate for a channel is always constant.

__init__(n_channels: int, n_units: int)[source]

Create a testing spike rate input.

Parameters

n_channels – The number of input channels.
n_units – The total number of units, which should be a multiple of the number of channels.

property channel_count: int

Get the number of input channels.

Returns: The input channel count.

read() → Optional[ndarray][source]

Read spike rates, one per channel.

Returns

The array of testing spike rates with shape (n_units,). For example, if n_channels = 50 and n_units_per_channel = 1, the spike rates will be constant and equal to:

[ 0. 2. 4. 6. 8. 10. 12. 14. 16. 18. 20. 22. 24. 26. 28. 30. 32. 34. 36. 38. 40. 42. 44. 46. 48. 50. 52. 54. 56. 58. 60. 62. 64. 66. 68. 70. 72. 74. 76. 78. 80. 82. 84. 86. 88. 90. 92. 94. 96. 98.]