ezmsg.blackrock.cerelink#

CereLink-based source for ezmsg — streams continuous and spike data from Blackrock devices.

Module Attributes

DeviceConfig

Device-configuration mode for one source.

Classes

class CcfConfig(path)[source]#

Bases: object

Load a CCF file. Device-wide configuration — at most one source per graph should carry this. Other sources targeting the same device set configure=None (pure subscriber).

Parameters:: path (str)

path: str#

__init__(path)#

Parameters:: path (str)
Return type:: None

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

Bases: _CereLinkBaseProducer[CereLinkSignalSettings, CereLinkSignalProducerState]

Streams one continuous sample-group as AxisArray.

class CereLinkSignalProducerState[source]#

Bases: _CereLinkSharedState

Signal-producer ring buffer + emission template.

buffer_data: ndarray | None = None#

buffer_timestamps: ndarray | None = None#

write_idx: int = 0#

read_idx: int = 0#

n_channels: int = 0#

template: AxisArray | None = None#

scale_factors: ndarray | None = None#

data_event: Event | None = None#

class CereLinkSignalSettings(device_type=None, subscribe_rate=SampleRate.SR_RAW, configure=None, cbtime=False, microvolts=True, cont_buffer_dur=0.5, cmp_configs=())[source]#

Bases: Settings

Settings for CereLinkSignalSource — emits one continuous sample-group as AxisArray.

Parameters:

device_type (DeviceType | None)
subscribe_rate (SampleRate)
configure (CcfConfig | SliceConfig | None)
cbtime (bool)
microvolts (bool)
cont_buffer_dur (float)
cmp_configs (tuple[ChannelMapSettings, ...])

device_type: DeviceType | None = None#: Device to connect to. None = idle (no Session opened).

subscribe_rate: SampleRate = 6#: The sample-group rate this source streams. Defaults to SR_RAW. Explicit SampleRate.NONE is rejected — a Source must subscribe to something.

configure: CcfConfig | SliceConfig | None = None#: Device configuration this source applies on open.

cbtime: bool = False#: True = raw device nanoseconds/1e9; False = time.monotonic() via clock sync.

microvolts: bool = True#: Convert int16 → µV using channel scale factors.

cont_buffer_dur: float = 0.5#: Ring buffer duration in seconds.

cmp_configs: tuple[ChannelMapSettings, ...] = ()#: One ChannelMapSettings per headstage applied after connection.

__init__(device_type=None, subscribe_rate=SampleRate.SR_RAW, configure=None, cbtime=False, microvolts=True, cont_buffer_dur=0.5, cmp_configs=())#

Parameters:

device_type (DeviceType | None)
subscribe_rate (SampleRate)
configure (CcfConfig | SliceConfig | None)
cbtime (bool)
microvolts (bool)
cont_buffer_dur (float)
cmp_configs (tuple[ChannelMapSettings, ...])

Return type:

None

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

Bases: BaseProducerUnit[CereLinkSignalSettings, AxisArray, CereLinkSignalProducer]

ezmsg Unit that streams one continuous sample-group from a Blackrock device.

SETTINGS#: alias of CereLinkSignalSettings

OUTPUT_DEVICE_STATUS = OutputStream:unlocated[DeviceStatus](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#

__init__(*args, **kwargs)[source]#

Initialize an Addressable object.

The name and location are initially None and must be set before the object can be properly addressed. This is achieved through the _set_name() and _set_location() methods.

Return type:: None

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

async device_status()[source]#

Return type:: AsyncGenerator

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

Bases: _CereLinkBaseProducer[CereLinkSpikeSettings, CereLinkSpikeProducerState]

Streams spike events as AxisArray of shape [time, ch, unit=7].

Time axis is the device’s 30 kHz spike clock; unit axis indexes the device convention (0=unsorted, 1..5=sorted, 6=noise — values >5 collapse into the noise bucket because the unit axis has fixed length 7).

Emission cadence is regular: every spike_buffer_dur seconds, one window of shape [N_t, n_ch, 7] is emitted. Empty windows are emitted too (downstream wants a steady time axis). The window before the first spike is suppressed — there’s no device-time anchor until then.

Spikes whose timestamp lands beyond the current window’s range are dropped (a downstream-backpressure failure mode); in normal operation _produce runs at the buffer cadence and the buffer is large enough to hold one window of spikes.

__init__(*args, **kwargs)[source]#

Return type:: None

class CereLinkSpikeProducerState[source]#

Bases: _CereLinkSharedState

Spike-producer rolling buffer + emission template.

buffer: ndarray | None = None#

n_channels: int = 0#

n_t: int = 0#

template: AxisArray | None = None#

data_event: Event | None = None#

chid_to_buffer_idx: dict | None = None#

window_origin_ns: int = -1#

class CereLinkSpikeSettings(device_type=None, configure=None, cbtime=False, microvolts=True, spike_buffer_dur=0.5, cmp_configs=())[source]#

Bases: Settings

Settings for CereLinkSpikeSource — emits sparse spike events as AxisArray of shape [time, ch, unit=7] at the 30 kHz spike clock. Unit indices follow the device convention: 0=unsorted, 1..5=sorted, 6=noise (header.type > 5).

Parameters:

device_type (DeviceType | None)
configure (CcfConfig | SliceConfig | None)
cbtime (bool)
microvolts (bool)
spike_buffer_dur (float)
cmp_configs (tuple[ChannelMapSettings, ...])

device_type: DeviceType | None = None#: Device to connect to. None = idle.

configure: CcfConfig | SliceConfig | None = None#: Device configuration this source applies on open.

cbtime: bool = False#: True = raw device nanoseconds/1e9; False = time.monotonic() via clock sync.

microvolts: bool = True#: Reserved for future spike-waveform emission — int16 → µV scaling.

spike_buffer_dur: float = 0.5#: Ring buffer duration in seconds (at the 30 kHz spike clock).

cmp_configs: tuple[ChannelMapSettings, ...] = ()#: One ChannelMapSettings per headstage applied after connection.

__init__(device_type=None, configure=None, cbtime=False, microvolts=True, spike_buffer_dur=0.5, cmp_configs=())#

Parameters:

device_type (DeviceType | None)
configure (CcfConfig | SliceConfig | None)
cbtime (bool)
microvolts (bool)
spike_buffer_dur (float)
cmp_configs (tuple[ChannelMapSettings, ...])

Return type:

None

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

Bases: BaseProducerUnit[CereLinkSpikeSettings, AxisArray, CereLinkSpikeProducer]

ezmsg Unit that streams spike events as AxisArray[time, ch, unit=7].

SETTINGS#: alias of CereLinkSpikeSettings

OUTPUT_DEVICE_STATUS = OutputStream:unlocated[DeviceStatus](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#

__init__(*args, **kwargs)[source]#

Initialize an Addressable object.

The name and location are initially None and must be set before the object can be properly addressed. This is achieved through the _set_name() and _set_location() methods.

Return type:: None

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

async device_status()[source]#

Return type:: AsyncGenerator

class ChannelSelection(*values)[source]#

Bases: Enum

Sentinel selections for SliceConfig.channels (the cases an explicit channel-ID list can’t express).

ALL = 'all'#: Every channel matching channel_type; the others are disabled.

ENABLED = 'enabled'#: Only the channels the device already has enabled for this stream; the enabled set is left unchanged. What counts as “enabled” is stream-specific: continuous sample-group membership for the signal source (channels are retuned to its rate/coupling but never enabled/disabled), and spike-extraction state for the spike source (extraction is left exactly as-is, and the source subscribes to whatever already has it on).

class DeviceStatus(device_type, success, error='')[source]#

Bases: object

Result of a settings-driven device switch.

Emitted on CereLinkSignalSource.OUTPUT_DEVICE_STATUS (and the corresponding spike-source output) after each _areset_state attempt. The GUI consumes this to confirm or revert its device selection — the snapshot’s settings_changed event fires before the open completes and so cannot distinguish success from failure.

Parameters:

device_type (DeviceType | None)
success (bool)
error (str)

device_type: DeviceType | None#

success: bool#

error: str = ''#

__init__(device_type, success, error='')#

Parameters:

device_type (DeviceType | None)
success (bool)
error (str)

Return type:

None

class SliceConfig(channels=ChannelSelection.ALL, channel_type=ChannelType.FRONTEND, ac_input_coupling=False, enable_spiking=False)[source]#

Bases: object

Programmatic per-slice device configuration owned by this source.

The source applies the device state needed to make its subscribed stream produce data on channels: set_sample_group for signal sources, set_spike_extraction for spike sources, plus AC coupling.

Multiple sources can carry disjoint slices for the same device; pycbsdk handles the merge. Overlap with incompatible settings is the user’s responsibility.

Parameters:

channels (list[int] | ChannelSelection)
channel_type (ChannelType)
ac_input_coupling (bool)
enable_spiking (bool)

channels: list[int] | ChannelSelection = 'all'#

Which channels this slice targets — one field, three intents:

list[int] — enable exactly these 1-based channel IDs; the other channels of channel_type are disabled. A provided list is always respected.
ChannelSelection.ALL (default) — enable every channel matching channel_type; others disabled.
ChannelSelection.ENABLED — leave the device’s enabled set as-is and only consume it. Signal source: retune the already-streaming channels (disable_others=False, so an unused front-end bank stays off). Spike source: leave spike extraction untouched and subscribe to whatever already has it on (enable_spiking is ignored in this mode). An empty enabled set yields nothing and warns.

channel_type: ChannelType = 0#

ac_input_coupling: bool = False#: Enable (True) or disable (False) AC coupling (highpass filter) on this slice. Note: This is applied unconditionally to CereLinkSignalProducer.

enable_spiking: bool = False#: Enable spike extraction on the selected channels (FRONTEND only). Honored by CereLinkSpikeSource; ignored by signal sources, and ignored when channels is ChannelSelection.ENABLED (which leaves extraction exactly as the device has it).

__init__(channels=ChannelSelection.ALL, channel_type=ChannelType.FRONTEND, ac_input_coupling=False, enable_spiking=False)#

Parameters:

channels (list[int] | ChannelSelection)
channel_type (ChannelType)
ac_input_coupling (bool)
enable_spiking (bool)

Return type:

None

class DeviceStatus(device_type, success, error='')[source]#

Bases: object

Result of a settings-driven device switch.

Parameters:

device_type (DeviceType | None)
success (bool)
error (str)

device_type: DeviceType | None#

success: bool#

error: str = ''#

__init__(device_type, success, error='')#

Parameters:

device_type (DeviceType | None)
success (bool)
error (str)

Return type:

None

class CcfConfig(path)[source]#

Bases: object

Load a CCF file. Device-wide configuration — at most one source per graph should carry this. Other sources targeting the same device set configure=None (pure subscriber).

Parameters:: path (str)

path: str#

__init__(path)#

Parameters:: path (str)
Return type:: None

class ChannelSelection(*values)[source]#

Bases: Enum

Sentinel selections for SliceConfig.channels (the cases an explicit channel-ID list can’t express).

ALL = 'all'#: Every channel matching channel_type; the others are disabled.

ENABLED = 'enabled'#: Only the channels the device already has enabled for this stream; the enabled set is left unchanged. What counts as “enabled” is stream-specific: continuous sample-group membership for the signal source (channels are retuned to its rate/coupling but never enabled/disabled), and spike-extraction state for the spike source (extraction is left exactly as-is, and the source subscribes to whatever already has it on).

class SliceConfig(channels=ChannelSelection.ALL, channel_type=ChannelType.FRONTEND, ac_input_coupling=False, enable_spiking=False)[source]#

Bases: object

Programmatic per-slice device configuration owned by this source.

The source applies the device state needed to make its subscribed stream produce data on channels: set_sample_group for signal sources, set_spike_extraction for spike sources, plus AC coupling.

Multiple sources can carry disjoint slices for the same device; pycbsdk handles the merge. Overlap with incompatible settings is the user’s responsibility.

Parameters:

channels (list[int] | ChannelSelection)
channel_type (ChannelType)
ac_input_coupling (bool)
enable_spiking (bool)

channels: list[int] | ChannelSelection = 'all'#

Which channels this slice targets — one field, three intents:

list[int] — enable exactly these 1-based channel IDs; the other channels of channel_type are disabled. A provided list is always respected.
ChannelSelection.ALL (default) — enable every channel matching channel_type; others disabled.
ChannelSelection.ENABLED — leave the device’s enabled set as-is and only consume it. Signal source: retune the already-streaming channels (disable_others=False, so an unused front-end bank stays off). Spike source: leave spike extraction untouched and subscribe to whatever already has it on (enable_spiking is ignored in this mode). An empty enabled set yields nothing and warns.

channel_type: ChannelType = 0#

ac_input_coupling: bool = False#: Enable (True) or disable (False) AC coupling (highpass filter) on this slice. Note: This is applied unconditionally to CereLinkSignalProducer.

enable_spiking: bool = False#: Enable spike extraction on the selected channels (FRONTEND only). Honored by CereLinkSpikeSource; ignored by signal sources, and ignored when channels is ChannelSelection.ENABLED (which leaves extraction exactly as the device has it).

__init__(channels=ChannelSelection.ALL, channel_type=ChannelType.FRONTEND, ac_input_coupling=False, enable_spiking=False)#

Parameters:

channels (list[int] | ChannelSelection)
channel_type (ChannelType)
ac_input_coupling (bool)
enable_spiking (bool)

Return type:

None

DeviceConfig = ezmsg.blackrock.cerelink.CcfConfig | ezmsg.blackrock.cerelink.SliceConfig | None#: Device-configuration mode for one source. None means another source or component owns the device config; this source only subscribes.

class CereLinkSignalSettings(device_type=None, subscribe_rate=SampleRate.SR_RAW, configure=None, cbtime=False, microvolts=True, cont_buffer_dur=0.5, cmp_configs=())[source]#

Bases: Settings

Settings for CereLinkSignalSource — emits one continuous sample-group as AxisArray.

Parameters:

device_type (DeviceType | None)
subscribe_rate (SampleRate)
configure (CcfConfig | SliceConfig | None)
cbtime (bool)
microvolts (bool)
cont_buffer_dur (float)
cmp_configs (tuple[ChannelMapSettings, ...])

device_type: DeviceType | None = None#: Device to connect to. None = idle (no Session opened).

subscribe_rate: SampleRate = 6#: The sample-group rate this source streams. Defaults to SR_RAW. Explicit SampleRate.NONE is rejected — a Source must subscribe to something.

configure: CcfConfig | SliceConfig | None = None#: Device configuration this source applies on open.

cbtime: bool = False#: True = raw device nanoseconds/1e9; False = time.monotonic() via clock sync.

microvolts: bool = True#: Convert int16 → µV using channel scale factors.

cont_buffer_dur: float = 0.5#: Ring buffer duration in seconds.

cmp_configs: tuple[ChannelMapSettings, ...] = ()#: One ChannelMapSettings per headstage applied after connection.

__init__(device_type=None, subscribe_rate=SampleRate.SR_RAW, configure=None, cbtime=False, microvolts=True, cont_buffer_dur=0.5, cmp_configs=())#

Parameters:

device_type (DeviceType | None)
subscribe_rate (SampleRate)
configure (CcfConfig | SliceConfig | None)
cbtime (bool)
microvolts (bool)
cont_buffer_dur (float)
cmp_configs (tuple[ChannelMapSettings, ...])

Return type:

None

class CereLinkSpikeSettings(device_type=None, configure=None, cbtime=False, microvolts=True, spike_buffer_dur=0.5, cmp_configs=())[source]#

Bases: Settings

Parameters:

device_type (DeviceType | None)
configure (CcfConfig | SliceConfig | None)
cbtime (bool)
microvolts (bool)
spike_buffer_dur (float)
cmp_configs (tuple[ChannelMapSettings, ...])

device_type: DeviceType | None = None#: Device to connect to. None = idle.

configure: CcfConfig | SliceConfig | None = None#: Device configuration this source applies on open.

cbtime: bool = False#: True = raw device nanoseconds/1e9; False = time.monotonic() via clock sync.

microvolts: bool = True#: Reserved for future spike-waveform emission — int16 → µV scaling.

spike_buffer_dur: float = 0.5#: Ring buffer duration in seconds (at the 30 kHz spike clock).

cmp_configs: tuple[ChannelMapSettings, ...] = ()#: One ChannelMapSettings per headstage applied after connection.

__init__(device_type=None, configure=None, cbtime=False, microvolts=True, spike_buffer_dur=0.5, cmp_configs=())#

Parameters:

device_type (DeviceType | None)
configure (CcfConfig | SliceConfig | None)
cbtime (bool)
microvolts (bool)
spike_buffer_dur (float)
cmp_configs (tuple[ChannelMapSettings, ...])

Return type:

None

class CereLinkSignalProducerState[source]#

Bases: _CereLinkSharedState

Signal-producer ring buffer + emission template.

buffer_data: ndarray | None = None#

buffer_timestamps: ndarray | None = None#

write_idx: int = 0#

read_idx: int = 0#

n_channels: int = 0#

template: AxisArray | None = None#

scale_factors: ndarray | None = None#

data_event: Event | None = None#

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

Bases: _CereLinkBaseProducer[CereLinkSignalSettings, CereLinkSignalProducerState]

Streams one continuous sample-group as AxisArray.

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

Bases: BaseProducerUnit[CereLinkSignalSettings, AxisArray, CereLinkSignalProducer]

ezmsg Unit that streams one continuous sample-group from a Blackrock device.

SETTINGS#: alias of CereLinkSignalSettings

OUTPUT_DEVICE_STATUS = OutputStream:unlocated[DeviceStatus](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#

__init__(*args, **kwargs)[source]#

Initialize an Addressable object.

The name and location are initially None and must be set before the object can be properly addressed. This is achieved through the _set_name() and _set_location() methods.

Return type:: None

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

async device_status()[source]#

Return type:: AsyncGenerator

class CereLinkSpikeProducerState[source]#

Bases: _CereLinkSharedState

Spike-producer rolling buffer + emission template.

buffer: ndarray | None = None#

n_channels: int = 0#

n_t: int = 0#

template: AxisArray | None = None#

data_event: Event | None = None#

chid_to_buffer_idx: dict | None = None#

window_origin_ns: int = -1#

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

Bases: _CereLinkBaseProducer[CereLinkSpikeSettings, CereLinkSpikeProducerState]

Streams spike events as AxisArray of shape [time, ch, unit=7].

__init__(*args, **kwargs)[source]#

Return type:: None

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

Bases: BaseProducerUnit[CereLinkSpikeSettings, AxisArray, CereLinkSpikeProducer]

ezmsg Unit that streams spike events as AxisArray[time, ch, unit=7].

SETTINGS#: alias of CereLinkSpikeSettings

OUTPUT_DEVICE_STATUS = OutputStream:unlocated[DeviceStatus](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#

__init__(*args, **kwargs)[source]#

Initialize an Addressable object.

The name and location are initially None and must be set before the object can be properly addressed. This is achieved through the _set_name() and _set_location() methods.

Return type:: None

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

async device_status()[source]#

Return type:: AsyncGenerator