ezmsg.lsl.inlet#

Classes

class LSLInfo(name: str = '', type: str = '', host: str = '', channel_count: int | None = None, nominal_srate: float = 0.0, channel_format: str | None = None)[source]#

Bases: object

Parameters:

  • name (str)

  • type (str)

  • host (str)

  • channel_count (int | None)

  • nominal_srate (float)

  • channel_format (str | None)

name: str = ''#
type: str = ''#
host: str = ''#
channel_count: int | None = None#
nominal_srate: float = 0.0#
channel_format: str | None = None#
__init__(name='', type='', host='', channel_count=None, nominal_srate=0.0, channel_format=None)#
Parameters:

  • name (str)

  • type (str)

  • host (str)

  • channel_count (int | None)

  • nominal_srate (float)

  • channel_format (str | None)

Return type:

None

class LSLInletGenerator(*args, **kwargs)[source]#

Bases: LSLInletProducer

Deprecated: use LSLInletProducer instead.

__init__(*args, **kwargs)[source]#
class LSLInletProducer(*args, settings=None, **kwargs)[source]#

Bases: BaseStatefulProducer[LSLInletSettings, AxisArray | None, LSLInletProducerState]

Parameters:

settings (SettingsType)

__init__(*args, settings=None, **kwargs)[source]#
Parameters:

settings (LSLInletSettings | None)

update_settings(new_settings)[source]#

Apply new settings to this producer, requesting a state reset if needed.

See BaseProcessor.update_settings(). The reset path matters most for stateful producers, where it reopens hardware or re-derives cached values in _reset_state.

Parameters:

new_settings (LSLInletSettings)

Return type:

None

shutdown()[source]#
Return type:

None

class LSLInletProducerState[source]#

Bases: object

__init__()[source]#
Return type:

None

resolver: ContinuousResolver | None = None#
inlet: StreamInlet | None = None#
clock_sync: ClockSync | None = None#
msg_template: AxisArray | None = None#
fetch_buffer: NDArray | None = None#
hash: int = -1#
class LSLInletSettings(info: LSLInfo = <factory>, local_buffer_dur: float = 1.0, use_arrival_time: bool = False, use_lsl_clock: bool = False, processing_flags: int = 15, pull_timeout: float = 0.1, max_pull_samples: int | None = None, reconnect_grace_dur: float = 5.0, distinct_key_per_connection: bool = False)[source]#

Bases: Settings

Parameters:

  • info (LSLInfo)

  • local_buffer_dur (float)

  • use_arrival_time (bool)

  • use_lsl_clock (bool)

  • processing_flags (int)

  • pull_timeout (float)

  • max_pull_samples (int | None)

  • reconnect_grace_dur (float)

  • distinct_key_per_connection (bool)

info: LSLInfo#
local_buffer_dur: float = 1.0#
use_arrival_time: bool = False#

Whether to ignore the LSL timestamps and use the time.time of the pull (True). If False (default), the LSL (send) timestamps are used. Send times may be converted from LSL clock to time.time clock. See use_lsl_clock.

use_lsl_clock: bool = False#

Whether the AxisArray.Axis.offset should use LSL’s clock (True) or time.time’s clock (False – default).

processing_flags: int = 15#

The processing flags option passed to pylsl.StreamInlet. Default is proc_ALL which includes all flags. Many users will want to set this to pylsl.proc_clocksync to disable dejittering.

pull_timeout: float = 0.1#

Maximum seconds to wait for the first sample during each steady-state pull. Once a sample arrives, pull_chunk(min_samples=1) immediately drains any other available samples up to max_pull_samples without waiting for the buffer to fill. The 0.1 second default therefore reduces idle wakeups without adding batching latency.

Pulling runs on a worker thread via asyncio.to_thread; the liblsl C call releases the GIL, so the event loop stays free while the inlet waits.

max_pull_samples: int | None = None#

Total cap on samples returned per pull_chunk. None (default) uses the full local fetch buffer. The inlet waits only for the first sample and then drains immediately available data up to this cap; it does not wait for the cap to be reached. A smaller value limits message size when clearing a large backlog, but may split that backlog across successive messages.

reconnect_grace_dur: float = 5.0#

Seconds after losing a stream during which only the same stream (matching source_id) is accepted. Afterwards any stream matching info is taken.

This reproduces liblsl’s own recovery, which re-acquires by source_id alone, but bounds it: liblsl retries indefinitely, so an upstream that comes back with a different shape (and therefore a different source_id) would never be found. The grace period prefers the original stream while it might still be coming back, then falls back to the resolver criteria so a restarted, reconfigured upstream is picked up. Set to 0 to always take the first match; the host criterion is honoured either way, which liblsl’s recovery does not do.

distinct_key_per_connection: bool = False#

Whether key gains a #<n> suffix that increments each time the inlet attaches to a different outlet instance (a changed StreamInfo uid).

Downstream processors key their state on (shape, rate, key), so a restarted upstream that keeps the same name and shape is otherwise invisible to them: filter state and partial windows carry across the discontinuity as if no gap occurred. Enabling this forces those resets.

Off by default because it changes the identity that key denotes: NWB writers name containers by it and pipelines route on it, so a reconnect would fork the recording. A dropped socket that re-attaches to the same outlet instance never bumps the suffix, so a brief blip preserves state either way.

__init__(info=<factory>, local_buffer_dur=1.0, use_arrival_time=False, use_lsl_clock=False, processing_flags=15, pull_timeout=0.1, max_pull_samples=None, reconnect_grace_dur=5.0, distinct_key_per_connection=False)#
Parameters:

  • info (LSLInfo)

  • local_buffer_dur (float)

  • use_arrival_time (bool)

  • use_lsl_clock (bool)

  • processing_flags (int)

  • pull_timeout (float)

  • max_pull_samples (int | None)

  • reconnect_grace_dur (float)

  • distinct_key_per_connection (bool)

Return type:

None

class LSLInletUnit(*args, settings=None, **kwargs)[source]#

Bases: BaseProducerUnit[LSLInletSettings, AxisArray | None, LSLInletProducer]

Represents a node in a graph that creates an LSL inlet and forwards the pulled data to the unit’s output.

Parameters:

  • stream_name – The name of the created LSL outlet.

  • stream_type – The type of the created LSL outlet.

  • settings (Settings | None)

SETTINGS#

alias of LSLInletSettings

create_producer()[source]#

Create the producer instance from settings.

Closes the previous producer first (if any), so resources held by it — sockets, file handles, hardware sessions — are released deterministically rather than being left to garbage collection.

Return type:

None

shutdown()[source]#

Runs when the Unit terminates.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should clean up resources and perform any necessary shutdown procedures.

Return type:

None

class LSLInfo(name: str = '', type: str = '', host: str = '', channel_count: int | None = None, nominal_srate: float = 0.0, channel_format: str | None = None)[source]#

Bases: object

Parameters:

  • name (str)

  • type (str)

  • host (str)

  • channel_count (int | None)

  • nominal_srate (float)

  • channel_format (str | None)

name: str = ''#
type: str = ''#
host: str = ''#
channel_count: int | None = None#
nominal_srate: float = 0.0#
channel_format: str | None = None#
__init__(name='', type='', host='', channel_count=None, nominal_srate=0.0, channel_format=None)#
Parameters:

  • name (str)

  • type (str)

  • host (str)

  • channel_count (int | None)

  • nominal_srate (float)

  • channel_format (str | None)

Return type:

None

class LSLInletSettings(info: LSLInfo = <factory>, local_buffer_dur: float = 1.0, use_arrival_time: bool = False, use_lsl_clock: bool = False, processing_flags: int = 15, pull_timeout: float = 0.1, max_pull_samples: int | None = None, reconnect_grace_dur: float = 5.0, distinct_key_per_connection: bool = False)[source]#

Bases: Settings

Parameters:

  • info (LSLInfo)

  • local_buffer_dur (float)

  • use_arrival_time (bool)

  • use_lsl_clock (bool)

  • processing_flags (int)

  • pull_timeout (float)

  • max_pull_samples (int | None)

  • reconnect_grace_dur (float)

  • distinct_key_per_connection (bool)

info: LSLInfo#
local_buffer_dur: float = 1.0#
use_arrival_time: bool = False#

Whether to ignore the LSL timestamps and use the time.time of the pull (True). If False (default), the LSL (send) timestamps are used. Send times may be converted from LSL clock to time.time clock. See use_lsl_clock.

use_lsl_clock: bool = False#

Whether the AxisArray.Axis.offset should use LSL’s clock (True) or time.time’s clock (False – default).

processing_flags: int = 15#

The processing flags option passed to pylsl.StreamInlet. Default is proc_ALL which includes all flags. Many users will want to set this to pylsl.proc_clocksync to disable dejittering.

pull_timeout: float = 0.1#

Maximum seconds to wait for the first sample during each steady-state pull. Once a sample arrives, pull_chunk(min_samples=1) immediately drains any other available samples up to max_pull_samples without waiting for the buffer to fill. The 0.1 second default therefore reduces idle wakeups without adding batching latency.

Pulling runs on a worker thread via asyncio.to_thread; the liblsl C call releases the GIL, so the event loop stays free while the inlet waits.

max_pull_samples: int | None = None#

Total cap on samples returned per pull_chunk. None (default) uses the full local fetch buffer. The inlet waits only for the first sample and then drains immediately available data up to this cap; it does not wait for the cap to be reached. A smaller value limits message size when clearing a large backlog, but may split that backlog across successive messages.

reconnect_grace_dur: float = 5.0#

Seconds after losing a stream during which only the same stream (matching source_id) is accepted. Afterwards any stream matching info is taken.

This reproduces liblsl’s own recovery, which re-acquires by source_id alone, but bounds it: liblsl retries indefinitely, so an upstream that comes back with a different shape (and therefore a different source_id) would never be found. The grace period prefers the original stream while it might still be coming back, then falls back to the resolver criteria so a restarted, reconfigured upstream is picked up. Set to 0 to always take the first match; the host criterion is honoured either way, which liblsl’s recovery does not do.

distinct_key_per_connection: bool = False#

Whether key gains a #<n> suffix that increments each time the inlet attaches to a different outlet instance (a changed StreamInfo uid).

Downstream processors key their state on (shape, rate, key), so a restarted upstream that keeps the same name and shape is otherwise invisible to them: filter state and partial windows carry across the discontinuity as if no gap occurred. Enabling this forces those resets.

Off by default because it changes the identity that key denotes: NWB writers name containers by it and pipelines route on it, so a reconnect would fork the recording. A dropped socket that re-attaches to the same outlet instance never bumps the suffix, so a brief blip preserves state either way.

__init__(info=<factory>, local_buffer_dur=1.0, use_arrival_time=False, use_lsl_clock=False, processing_flags=15, pull_timeout=0.1, max_pull_samples=None, reconnect_grace_dur=5.0, distinct_key_per_connection=False)#
Parameters:

  • info (LSLInfo)

  • local_buffer_dur (float)

  • use_arrival_time (bool)

  • use_lsl_clock (bool)

  • processing_flags (int)

  • pull_timeout (float)

  • max_pull_samples (int | None)

  • reconnect_grace_dur (float)

  • distinct_key_per_connection (bool)

Return type:

None

class LSLInletProducerState[source]#

Bases: object

__init__()[source]#
Return type:

None

resolver: ContinuousResolver | None = None#
inlet: StreamInlet | None = None#
clock_sync: ClockSync | None = None#
msg_template: AxisArray | None = None#
fetch_buffer: NDArray | None = None#
hash: int = -1#
class LSLInletProducer(*args, settings=None, **kwargs)[source]#

Bases: BaseStatefulProducer[LSLInletSettings, AxisArray | None, LSLInletProducerState]

Parameters:

settings (SettingsType)

__init__(*args, settings=None, **kwargs)[source]#
Parameters:

settings (LSLInletSettings | None)

update_settings(new_settings)[source]#

Apply new settings to this producer, requesting a state reset if needed.

See BaseProcessor.update_settings(). The reset path matters most for stateful producers, where it reopens hardware or re-derives cached values in _reset_state.

Parameters:

new_settings (LSLInletSettings)

Return type:

None

shutdown()[source]#
Return type:

None

class LSLInletGenerator(*args, **kwargs)[source]#

Bases: LSLInletProducer

Deprecated: use LSLInletProducer instead.

__init__(*args, **kwargs)[source]#
class LSLInletUnit(*args, settings=None, **kwargs)[source]#

Bases: BaseProducerUnit[LSLInletSettings, AxisArray | None, LSLInletProducer]

Represents a node in a graph that creates an LSL inlet and forwards the pulled data to the unit’s output.

Parameters:

  • stream_name – The name of the created LSL outlet.

  • stream_type – The type of the created LSL outlet.

  • settings (Settings | None)

SETTINGS#

alias of LSLInletSettings

create_producer()[source]#

Create the producer instance from settings.

Closes the previous producer first (if any), so resources held by it — sockets, file handles, hardware sessions — are released deterministically rather than being left to garbage collection.

Return type:

None

shutdown()[source]#

Runs when the Unit terminates.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should clean up resources and perform any necessary shutdown procedures.

Return type:

None

On this page
Show Source