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 tap-sum, 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 the MLX concatenate-vs-concat spelling is special-cased.

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=33, 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 = 33

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.

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=33, 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.

fir: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None

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

hist: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None

Carried input history, shape (filter_len-1, *sample_shape).

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'})

class SamplingDelayAlignmentSettings(bank_size=32, channel_sample_interval=9.696969696969698e-07, filter_len=33, 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 = 33

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.

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=33, 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.

fir: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None

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

hist: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None

Carried input history, shape (filter_len-1, *sample_shape).

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'})

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

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

Parameters:

settings (Settings | None)

SETTINGS

alias of SamplingDelayAlignmentSettings