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_to_peak(uV) / I_peak_to_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 Hann-windowed DFT.

Detrends the tail of the burst (skipping the settle transient), applies a Hann window, computes the FFT, and sums the band-limited energy to recover the peak amplitude of the injected test tone. Impedance is V_p2p / I_p2p.

Important

data must be in microvolts. Passing raw ADC counts will produce impedance values that are wrong by the ADC scale factor. When using CereLinkSignalSource, 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 frequency range for peak extraction (Hz).

  • freq_hi (float) – Upper bound of frequency range for peak extraction (Hz).

  • test_current_nA (float) – Peak-to-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

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 CereLinkSignalSource, 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. This relies on the device’s internal filtering being disabled — a filter produces small non-zero residuals on idle channels, which defeats the exact-zero exclusivity checks. Misconfiguration there is a sign that the recording chain needs fixing, not that this algorithm should accommodate it.

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).

NONRESET_SETTINGS_FIELDS: ClassVar[frozenset[str]] = frozenset({'freq_hi', 'freq_lo', 'headstage_channel_offsets', 'test_current_nA'})#
update_settings(new_settings)[source]#

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

Diffs new_settings against the current self.settings. If every changed field is listed in NONRESET_SETTINGS_FIELDS, self.settings is simply rebound and the processor keeps running. Otherwise, _request_reset() is called so the next message triggers a fresh _reset_state (for stateful subclasses). Override for finer-grained control than the class-level allow-list provides.

Parameters:

new_settings (CerePlexImpedanceSettings)

Return type:

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#
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 Hann-windowed DFT.

Detrends the tail of the burst (skipping the settle transient), applies a Hann window, computes the FFT, and sums the band-limited energy to recover the peak amplitude of the injected test tone. Impedance is V_p2p / I_p2p.

Important

data must be in microvolts. Passing raw ADC counts will produce impedance values that are wrong by the ADC scale factor. When using CereLinkSignalSource, 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 frequency range for peak extraction (Hz).

  • freq_hi (float) – Upper bound of frequency range for peak extraction (Hz).

  • test_current_nA (float) – Peak-to-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 CereLinkSignalSource, 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. This relies on the device’s internal filtering being disabled — a filter produces small non-zero residuals on idle channels, which defeats the exact-zero exclusivity checks. Misconfiguration there is a sign that the recording chain needs fixing, not that this algorithm should accommodate it.

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).

NONRESET_SETTINGS_FIELDS: ClassVar[frozenset[str]] = frozenset({'freq_hi', 'freq_lo', 'headstage_channel_offsets', 'test_current_nA'})#
update_settings(new_settings)[source]#

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

Diffs new_settings against the current self.settings. If every changed field is listed in NONRESET_SETTINGS_FIELDS, self.settings is simply rebound and the processor keeps running. Otherwise, _request_reset() is called so the next message triggers a fresh _reset_state (for stateful subclasses). Override for finer-grained control than the class-level allow-list provides.

Parameters:

new_settings (CerePlexImpedanceSettings)

Return type:

None

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

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

Parameters:

settings (Settings | None)

SETTINGS#

alias of CerePlexImpedanceSettings