ezmsg.blackrock.sampling_delay_alignment

Align channels sampled at different instants by a sequential A/D.

The Gemini front-end samples channels in banks of bank_size (32), one every channel_sample_interval (~969.7 ns), so channel c’s sample n is the signal at t_n + tau_c, tau_c = (c % bank_size) * channel_sample_interval. For any cross-channel operation (CAR, whitening, beamforming) this misalignment smears the common-mode at high frequency: the phase spread across a bank is 2*pi*f*T_bank – negligible at 60 Hz (0.65 deg) but ~81 deg at 7.5 kHz, so e.g. CAR’s common-mode rejection collapses toward Nyquist.

This transformer removes that by delaying each channel by tau_c with a per-slot windowed-sinc fractional-delay filter, bringing every channel onto a common time grid (the bank start). A windowed-sinc is used rather than linear interpolation on purpose: linear interpolation is a delay-dependent low-pass that would impose a different high-frequency rolloff per channel – coloring the band exactly where the misalignment mattered. There are only bank_size distinct delays, so only that many distinct filters.

The within-bank slot defaults to acquisition order (c % bank_size). If the channel axis carries per-channel bank/elec metadata (e.g. attached by ChannelMapUnit), the slot is taken from elec (elec - 1) instead, so each channel’s delay is correct even when channels are reordered relative to hardware acquisition.

Cost / caveats:

• Latency: the causal FIR has a common bulk delay of (filter_len-1)//2 samples (the per-channel fractional delays ride on top). The output time axis offset is shifted to keep timestamps physically correct.

• It resamples the raw data – downstream sees interpolated samples. Fine for cross-channel cleaning; be deliberate if a step needs raw waveforms.

• Railing: clipped (rail) samples are corrupt and a fractional-delay filter would spread that corruption over its support. With rail_threshold set, railed samples are held at the last valid value before filtering (a basic mitigation). A production version should also emit a reliability mask so downstream can discount the ~``filter_len`` samples around each rail. FIR (used here) localizes the damage; an IIR all-pass (e.g. Thiran) would ring across it.

Array-API compatible: it detects the input’s namespace and runs on the working backend (numpy, MLX, torch, jax, cupy, …). The sinc taps are designed in numpy and moved to the backend; everything else – the FIR, concat/state handling, and the rail forward-fill – runs on the backend using only standard Array-API ops (the forward-fill’s cumulative max is built from maximum + shifts, since the standard lacks one). Only MLX’s concatenate-vs-concat spelling needs special-casing.

Backend-specific fast paths sit on top of that, because live acquisition delivers chunks of a few dozen samples where per-operation dispatch, not arithmetic, sets the wall time:

• on MLX the FIR runs as a single depthwise conv_general (one group per column, kernel cached at state reset) instead of filter_len multiply-add stages – ~2-3x less time per message, and roughly flat in chunk size;

• on numpy, if the optional numba dependency is installed, both the FIR and the rail forward-fill run as fused single-pass jitted kernels (see ezmsg.blackrock._numba_kernels): several times faster than the tap loop at live sizes and, with a threaded FIR for large buffers, ~20x faster for offline batch processing. That path also runs out of one reused [history, chunk] buffer rather than concatenating the history onto each message, so a steady stream allocates only its output;

• the forward-fill’s running max otherwise uses mx.cummax on MLX and the maximum ufunc’s accumulate on numpy/cupy, in place of the log scan.

All of these fall back to the portable formulation wherever the op or the optional dependency is missing.

Classes

class SamplingDelayAlignment(*args, settings=None, **kwargs)

Bases: BaseTransformerUnit[SamplingDelayAlignmentSettings, AxisArray, AxisArray, SamplingDelayAlignmentTransformer]

Parameters:

settings (Settings | None)

SETTINGS

alias of SamplingDelayAlignmentSettings

class SamplingDelayAlignmentSettings(bank_size=32, channel_sample_interval=9.696969696969698e-07, filter_len=13, rail_threshold=None)

Bases: Settings

Settings for SamplingDelayAlignmentTransformer.

Parameters:

• bank_size (int)

• channel_sample_interval (float)

• filter_len (int)

• rail_threshold (float | None)

bank_size: int = 32

Channels per simultaneously-started A/D bank. Used to derive each channel’s sweep slot (c % bank_size) only as a fallback, when the channel axis carries no bank/elec metadata.

channel_sample_interval: float = 9.696969696969698e-07

Seconds between successive channels within a bank.

filter_len: int = 13

Sinc FIR length (odd). Bulk delay is (filter_len-1)//2 samples; longer = flatter passband / better near Nyquist, at more latency and compute. Set to 0 to disable alignment entirely – the transformer becomes a pass-through that returns its input unchanged.

Worst case over all bank_size fractional delays, at 30 kHz (see tests/test_sampling_delay_alignment.py for the pinned numbers):

Passband	filter_len	Max phase error	Max mag error	Bulk delay
0-500 Hz	7	0.0009 deg	0.00001 dB	3 samples
0-3 kHz	9	0.0038 deg	0.0001 dB	4 samples
0-7.5 kHz	13	0.015 deg	0.0015 dB	6 samples
0-7.5 kHz	33	0.0028 deg	0.0004 dB	16 samples

The default 13 covers the full broadband/spike band with ~3 orders of magnitude of margin on the ~81 deg of skew it is correcting, at less than half the latency of the former 33-tap default. Use 7 in an LFP-only pipeline; 33 buys accuracy that is already far below the noise floor.

rail_threshold: float | None = None

If set, samples with abs(value) >= rail_threshold are treated as clipped and held at the last valid value before filtering. None skips rail handling. (For Blackrock int16 at 0.25 uV/count, the rail is ~8191 uV.)

__init__(bank_size=32, channel_sample_interval=9.696969696969698e-07, filter_len=13, rail_threshold=None)

Parameters:

• bank_size (int)

• channel_sample_interval (float)

• filter_len (int)

• rail_threshold (float | None)

Return type:

None

class SamplingDelayAlignmentState

Bases: object

State for SamplingDelayAlignmentTransformer.

axis: str = ''

The resolved stream dimension, fixed at reset so every later use agrees.

fir: NDArray | None = None

Per-channel sinc FIR taps, shape (filter_len, n_ch).

conv_w: Any | None = None

MLX depthwise-conv kernel, shape (n_cols, filter_len, 1) – the same taps as fir, reversed and laid out per flattened sample column. None on every other backend (and when the taps don’t broadcast over sample_shape), which selects the portable tap-sum instead.

nb_w: NDArray | None = None

numba FIR kernel weights, shape (filter_len, n_cols) – the same taps as fir, reversed and laid out per flattened sample column, in the data dtype. Set only on the numpy backend when numba is installed; None otherwise, which selects the portable tap-sum.

hist: NDArray | None = None

Carried input history, shape (filter_len-1, *sample_shape). On the numba path this is a view of the leading rows of scratch; elsewhere it owns its memory (see _own()).

scratch: NDArray | None = None

the reused (filter_len-1 + capacity, n_cols) work buffer holding the carried history followed by the current chunk, so the per-message concat and its full-chunk temporary don’t happen. Grown (never shrunk) to the largest chunk seen; None until the first message.

Type:

numba path only

bulk_delay: int = 0

Common bulk delay (filter_len-1)//2 samples (for the offset shift).

class SamplingDelayAlignmentTransformer(*args, **kwargs)

Bases: BaseStatefulTransformer[SamplingDelayAlignmentSettings, AxisArray, AxisArray, SamplingDelayAlignmentState]

Per-channel fractional-delay alignment (see module docstring).

NONRESET_SETTINGS_FIELDS: ClassVar[frozenset[str]] = frozenset({'rail_threshold'})

__init__(*args, **kwargs)

Return type:

None

class SamplingDelayAlignmentSettings(bank_size=32, channel_sample_interval=9.696969696969698e-07, filter_len=13, rail_threshold=None)

Bases: Settings

Settings for SamplingDelayAlignmentTransformer.

Parameters:

• bank_size (int)

• channel_sample_interval (float)

• filter_len (int)

• rail_threshold (float | None)

bank_size: int = 32

Channels per simultaneously-started A/D bank. Used to derive each channel’s sweep slot (c % bank_size) only as a fallback, when the channel axis carries no bank/elec metadata.

channel_sample_interval: float = 9.696969696969698e-07

Seconds between successive channels within a bank.

filter_len: int = 13

Sinc FIR length (odd). Bulk delay is (filter_len-1)//2 samples; longer = flatter passband / better near Nyquist, at more latency and compute. Set to 0 to disable alignment entirely – the transformer becomes a pass-through that returns its input unchanged.

Worst case over all bank_size fractional delays, at 30 kHz (see tests/test_sampling_delay_alignment.py for the pinned numbers):

Passband	filter_len	Max phase error	Max mag error	Bulk delay
0-500 Hz	7	0.0009 deg	0.00001 dB	3 samples
0-3 kHz	9	0.0038 deg	0.0001 dB	4 samples
0-7.5 kHz	13	0.015 deg	0.0015 dB	6 samples
0-7.5 kHz	33	0.0028 deg	0.0004 dB	16 samples

The default 13 covers the full broadband/spike band with ~3 orders of magnitude of margin on the ~81 deg of skew it is correcting, at less than half the latency of the former 33-tap default. Use 7 in an LFP-only pipeline; 33 buys accuracy that is already far below the noise floor.

rail_threshold: float | None = None

If set, samples with abs(value) >= rail_threshold are treated as clipped and held at the last valid value before filtering. None skips rail handling. (For Blackrock int16 at 0.25 uV/count, the rail is ~8191 uV.)

__init__(bank_size=32, channel_sample_interval=9.696969696969698e-07, filter_len=13, rail_threshold=None)

Parameters:

• bank_size (int)

• channel_sample_interval (float)

• filter_len (int)

• rail_threshold (float | None)

Return type:

None

class SamplingDelayAlignmentState

Bases: object

State for SamplingDelayAlignmentTransformer.

axis: str = ''

The resolved stream dimension, fixed at reset so every later use agrees.

fir: NDArray | None = None

Per-channel sinc FIR taps, shape (filter_len, n_ch).

conv_w: Any | None = None

MLX depthwise-conv kernel, shape (n_cols, filter_len, 1) – the same taps as fir, reversed and laid out per flattened sample column. None on every other backend (and when the taps don’t broadcast over sample_shape), which selects the portable tap-sum instead.

nb_w: NDArray | None = None

numba FIR kernel weights, shape (filter_len, n_cols) – the same taps as fir, reversed and laid out per flattened sample column, in the data dtype. Set only on the numpy backend when numba is installed; None otherwise, which selects the portable tap-sum.

hist: NDArray | None = None

Carried input history, shape (filter_len-1, *sample_shape). On the numba path this is a view of the leading rows of scratch; elsewhere it owns its memory (see _own()).

scratch: NDArray | None = None

the reused (filter_len-1 + capacity, n_cols) work buffer holding the carried history followed by the current chunk, so the per-message concat and its full-chunk temporary don’t happen. Grown (never shrunk) to the largest chunk seen; None until the first message.

Type:

numba path only

bulk_delay: int = 0

Common bulk delay (filter_len-1)//2 samples (for the offset shift).

class SamplingDelayAlignmentTransformer(*args, **kwargs)

Bases: BaseStatefulTransformer[SamplingDelayAlignmentSettings, AxisArray, AxisArray, SamplingDelayAlignmentState]

Per-channel fractional-delay alignment (see module docstring).

NONRESET_SETTINGS_FIELDS: ClassVar[frozenset[str]] = frozenset({'rail_threshold'})

__init__(*args, **kwargs)

Return type:

None

class SamplingDelayAlignment(*args, settings=None, **kwargs)

Bases: BaseTransformerUnit[SamplingDelayAlignmentSettings, AxisArray, AxisArray, SamplingDelayAlignmentTransformer]

Parameters:

settings (Settings | None)

SETTINGS

alias of SamplingDelayAlignmentSettings