ezmsg.blackrock.cereplex_impedance#

CerePlex impedance measurement pipeline.

The CerePlex headstage injects a 1 kHz, 1 nA sine wave for 100 ms per channel, cycling sequentially through all channels. Channels not under test read exactly zero (filters must be disabled). Impedance is extracted via single-bin DFT at 1 kHz: Z(kOhm) = V_peak(uV) / I_peak(nA).

Multiple headstages are tracked independently — each may be at a different point in its impedance sweep.

Functions

extract_impedance(data, fft_samples, fs, freq_lo, freq_hi, test_current_nA)[source]#

Extract impedance (kOhm) from the 1 kHz component via single-bin DFT.

Detrends the tail of the burst (skipping the settle transient), computes the FFT, and extracts the peak-to-peak amplitude in the freq_lo–freq_hi band. Impedance is V_p2p / I_peak.

Important

data must be in microvolts. Passing raw ADC counts will produce impedance values that are wrong by the ADC scale factor. When using CereLinkSource, set microvolts=True.

Parameters:

  • data (ndarray) – 1-D array of samples for a single channel burst, in microvolts.

  • fft_samples (int) – Number of samples (from the end of data) to use for the FFT.

  • fs (float) – Sampling rate in Hz.

  • freq_lo (float) – Lower bound of the extraction band (Hz).

  • freq_hi (float) – Upper bound of the extraction band (Hz).

  • test_current_nA (float) – Peak amplitude of the injected test current (nA).

Returns:

Impedance in kOhm, or None if data is too short or no energy is found in the target band.

Return type:

float | None

Classes

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

Bases: BaseTransformerUnit[CerePlexImpedanceSettings, AxisArray, AxisArray, CerePlexImpedanceProcessor]

Parameters:

settings (Settings | None)

SETTINGS#

alias of CerePlexImpedanceSettings

async on_settings(msg)[source]#

Apply new settings.

If only headstage_channel_offsets changed, rebuild the trackers in place — the accumulated impedance values for previously-measured channels remain valid. Any other change recreates the processor as usual (state is reset on the next message).

Parameters:

msg (CerePlexImpedanceSettings)

Return type:

None

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

Bases: BaseStatefulTransformer[CerePlexImpedanceSettings, AxisArray, AxisArray | None, CerePlexImpedanceState]

Stateful transformer that extracts per-channel impedance from a CerePlex sweep.

Expects a stream of AxisArray messages with dims ["time", "ch"] where the data is in microvolts. When using CereLinkSource, set microvolts=True; raw ADC counts will produce incorrect results.

The processor tracks one or more headstages independently (configured via CerePlexImpedanceSettings.headstage_channel_offsets). Each headstage’s impedance sweep cycles sequentially through its channels: exactly one channel is non-zero at a time while the others read zero.

On each impedance update the processor emits an AxisArray whose data is a (1, n_ch) array of impedance values in kOhm (NaN for channels not yet measured).

class CerePlexImpedanceSettings(headstage_channel_offsets: tuple[int, ...] = (0,), collect_duration_s: float = 0.1, fft_duration_s: float = 0.09227, freq_lo: float = 960.0, freq_hi: float = 1050.0, test_current_nA: float = 1.0)[source]#

Bases: Settings

Parameters:
headstage_channel_offsets: tuple[int, ...] = (0,)#

Starting channel index of each CerePlex headstage. Each headstage’s range extends from its offset to the next offset (or n_ch for the last). Example: two 128-ch headstages → (0, 128).

collect_duration_s: float = 0.1#

Maximum burst duration to buffer per channel (100 ms for CerePlex).

fft_duration_s: float = 0.09227#

Duration of data used for FFT, taken from the end of the burst. The preceding samples serve as settle time.

freq_lo: float = 960.0#

Lower bound of frequency range for peak extraction (Hz).

freq_hi: float = 1050.0#

Upper bound of frequency range for peak extraction (Hz).

test_current_nA: float = 1.0#

Injected test-current peak-to-peak amplitude (nA).

__init__(headstage_channel_offsets=(0,), collect_duration_s=0.1, fft_duration_s=0.09227, freq_lo=960.0, freq_hi=1050.0, test_current_nA=1.0)#
Parameters:
Return type:

None

class CerePlexImpedanceState[source]#

Bases: object

trackers: list | None = None#
max_buffer_samples: int = 0#
fft_samples: int = 0#
fs: float = 0.0#
impedance: ndarray | None = None#
ch_axis: Any = None#
class CerePlexImpedanceSettings(headstage_channel_offsets: tuple[int, ...] = (0,), collect_duration_s: float = 0.1, fft_duration_s: float = 0.09227, freq_lo: float = 960.0, freq_hi: float = 1050.0, test_current_nA: float = 1.0)[source]#

Bases: Settings

Parameters:
headstage_channel_offsets: tuple[int, ...] = (0,)#

Starting channel index of each CerePlex headstage. Each headstage’s range extends from its offset to the next offset (or n_ch for the last). Example: two 128-ch headstages → (0, 128).

collect_duration_s: float = 0.1#

Maximum burst duration to buffer per channel (100 ms for CerePlex).

fft_duration_s: float = 0.09227#

Duration of data used for FFT, taken from the end of the burst. The preceding samples serve as settle time.

freq_lo: float = 960.0#

Lower bound of frequency range for peak extraction (Hz).

freq_hi: float = 1050.0#

Upper bound of frequency range for peak extraction (Hz).

test_current_nA: float = 1.0#

Injected test-current peak-to-peak amplitude (nA).

__init__(headstage_channel_offsets=(0,), collect_duration_s=0.1, fft_duration_s=0.09227, freq_lo=960.0, freq_hi=1050.0, test_current_nA=1.0)#
Parameters:
Return type:

None

class CerePlexImpedanceState[source]#

Bases: object

trackers: list | None = None#
max_buffer_samples: int = 0#
fft_samples: int = 0#
fs: float = 0.0#
impedance: ndarray | None = None#
ch_axis: Any = None#
extract_impedance(data, fft_samples, fs, freq_lo, freq_hi, test_current_nA)[source]#

Extract impedance (kOhm) from the 1 kHz component via single-bin DFT.

Detrends the tail of the burst (skipping the settle transient), computes the FFT, and extracts the peak-to-peak amplitude in the freq_lo–freq_hi band. Impedance is V_p2p / I_peak.

Important

data must be in microvolts. Passing raw ADC counts will produce impedance values that are wrong by the ADC scale factor. When using CereLinkSource, set microvolts=True.

Parameters:

  • data (ndarray) – 1-D array of samples for a single channel burst, in microvolts.

  • fft_samples (int) – Number of samples (from the end of data) to use for the FFT.

  • fs (float) – Sampling rate in Hz.

  • freq_lo (float) – Lower bound of the extraction band (Hz).

  • freq_hi (float) – Upper bound of the extraction band (Hz).

  • test_current_nA (float) – Peak amplitude of the injected test current (nA).

Returns:

Impedance in kOhm, or None if data is too short or no energy is found in the target band.

Return type:

float | None

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

Bases: BaseStatefulTransformer[CerePlexImpedanceSettings, AxisArray, AxisArray | None, CerePlexImpedanceState]

Stateful transformer that extracts per-channel impedance from a CerePlex sweep.

Expects a stream of AxisArray messages with dims ["time", "ch"] where the data is in microvolts. When using CereLinkSource, set microvolts=True; raw ADC counts will produce incorrect results.

The processor tracks one or more headstages independently (configured via CerePlexImpedanceSettings.headstage_channel_offsets). Each headstage’s impedance sweep cycles sequentially through its channels: exactly one channel is non-zero at a time while the others read zero.

On each impedance update the processor emits an AxisArray whose data is a (1, n_ch) array of impedance values in kOhm (NaN for channels not yet measured).

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

Bases: BaseTransformerUnit[CerePlexImpedanceSettings, AxisArray, AxisArray, CerePlexImpedanceProcessor]

Parameters:

settings (Settings | None)

SETTINGS#

alias of CerePlexImpedanceSettings

async on_settings(msg)[source]#

Apply new settings.

If only headstage_channel_offsets changed, rebuild the trackers in place — the accumulated impedance values for previously-measured channels remain valid. Any other change recreates the processor as usual (state is reset on the next message).

Parameters:

msg (CerePlexImpedanceSettings)

Return type:

None