Skip to content

dspk.ops.filter

Filters on the device, one module per family (mirroring dspk_ops::filter).

The implementation files (_iir.py, _fir.py, …) are private; use the names here.

fir Filters with FIR taps on the device.
firwin Low-pass FIR taps with unit gain at DC, as scipy.signal.firwin with one cutoff.
gaussian_smooth Zero-phase Gaussian smoothing, as scipy.ndimage.gaussian_filter1d along time.
iir Designs an IIR filter and applies it on the device.
median Running median along time, as scipy.signal.medfilt (with edge="zeros").
notch Removes one frequency (a second-order notch), as scipy.signal.iirnotch.
settling_samples Samples after which the impulse response of sos has settled.
sos Second-order sections of an IIR filter design, in scipy’s layout.
sosfilt Applies second-order sections on the device.
subtract_template Subtracts a template from data at every event, each aligned and scaled.
teager_kaiser Teager–Kaiser energy along time.
fir(
data,
taps,
*,
centered: bool = False,
edge: str = FIR_DEFAULT_EDGE,
runtime: str | None = None,
) -> np.ndarray

Filters with FIR taps on the device.

Parameters

  • data : array_like
    Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
  • taps : array_like
    Filter taps (1-D).
  • centered : bool, default False
    False: causal, y[t] = sum(taps[k] * x[t - k]), as scipy.signal.lfilter(taps, 1, x) with edge="zeros". True: the taps (an odd number) centred on each sample, zero phase for symmetric taps.
  • edge : {"zeros", "odd", "reflect", "nearest"}, default FIR_DEFAULT_EDGE (‘zeros’)
    What the samples past each end are taken to be.
  • runtime : str, default None
    A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

  • np.ndarray — The filtered signal (channels, samples), float32.

Raises

  • ValueError: For empty taps, an even number of centred taps, an unknown edge, or data not 2-D.
  • RuntimeError: When the runtime has no device.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> taps = dspk.ops.filter.firwin(101, 250.0, sample_rate=30_000.0)
>>> lfp = dspk.ops.filter.fir(x, taps, centered=True)
firwin(
num_taps: int,
cutoff: float,
*,
sample_rate: float | None = None,
window: str = 'hamming',
beta: float | None = None,
) -> np.ndarray

Low-pass FIR taps with unit gain at DC, as scipy.signal.firwin with one cutoff.

Parameters

  • num_taps : int
    Number of taps (odd for a centred, zero-phase filter).
  • cutoff : float
    Cutoff in Hz when sample_rate is given, else a fraction of Nyquist in (0, 1].
  • sample_rate : float, default None
    Sample rate in Hz.
  • window : {"hamming", "kaiser"}, default 'hamming'
    The window shaping the taps.
  • beta : float, default None
    Kaiser shape parameter; "kaiser" needs it, "hamming" takes none.

Returns

  • np.ndarray — Taps (num_taps,).

Raises

  • ValueError: For num_taps below 1, a cutoff outside (0, Nyquist], a non-positive sample_rate, an unknown window, or beta given to (or missing from) a window.

Examples

>>> import dspk
>>> taps = dspk.ops.filter.firwin(101, 250.0, sample_rate=30_000.0)
gaussian_smooth(
data,
sigma_samples: float,
*,
edge: str = GAUSSIAN_DEFAULT_EDGE,
runtime: str | None = None,
) -> np.ndarray

Zero-phase Gaussian smoothing, as scipy.ndimage.gaussian_filter1d along time.

The kernel is cut at GAUSSIAN_TRUNCATE standard deviations.

Parameters

  • data : array_like
    Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
  • sigma_samples : float
    Standard deviation of the Gaussian, in samples.
  • edge : {"zeros", "odd", "reflect", "nearest"}, default GAUSSIAN_DEFAULT_EDGE (‘reflect’)
    What the samples past each end are taken to be (scipy’s mode).
  • runtime : str, default None
    A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

  • np.ndarray — The filtered signal (channels, samples), float32.

Raises

  • ValueError: For a non-positive sigma_samples, an unknown edge, or data not 2-D.
  • RuntimeError: When the runtime has no device.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> smooth = dspk.ops.filter.gaussian_smooth(np.abs(x), 30.0)
iir(
data,
sample_rate: float,
band: str,
cutoff: float | Sequence[float],
*,
order: int = DEFAULT_ORDER,
design: str = 'butterworth',
ripple_db: float | None = None,
mode: str = 'forward_backward',
start: str = 'rest',
runtime: str | None = None,
) -> np.ndarray

Designs an IIR filter and applies it on the device.

The design is sos’s; filtering is sosfilt’s.

Parameters

  • data : array_like
    Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
  • sample_rate : float
    Sample rate of data in Hz.
  • band : {"lowpass", "highpass", "bandpass", "bandstop"}
    The band the filter passes (or stops).
  • cutoff : float or (float, float)
    Cutoff in Hz: one for "lowpass" and "highpass", (low, high) for "bandpass" and "bandstop".
  • order : int, default DEFAULT_ORDER (5)
    Filter order (a zero-phase run applies it twice).
  • design : {"butterworth", "chebyshev1", "bessel"}, default 'butterworth'
    The filter family.
  • ripple_db : float, default None
    Passband ripple in dB; "chebyshev1" needs it.
  • mode : {"forward_backward", "forward"}, default 'forward_backward'
    "forward_backward": zero phase, as scipy.signal.sosfiltfilt (the magnitude response is applied twice). "forward": causal, one pass, as scipy.signal.sosfilt.
  • start : {"rest", "steady_state"}, default 'rest'
    Where a "forward" pass starts: at rest, as if the signal were zero before the first sample (sosfilt without zi), or from the steady state of the first sample (zi = sosfilt_zi(sos) * x[0]: no step at the start). Forward-backward passes always start from the steady state.
  • runtime : str, default None
    A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

  • np.ndarray — The filtered signal (channels, samples), float32.

Raises

  • ValueError: For an invalid design (see sos), mode or start, or data not 2-D.
  • RuntimeError: When the runtime has no device.

Notes

Zero-phase runs pad each end by settling_samples (odd reflection), where scipy.signal.sosfiltfilt pads 3 * (2 * len(sos) + 1) samples by default: the interior is the same, the first and last samples can differ. sosfiltfilt(sos, x, padlen=settling_samples(sos)) matches to float32 precision.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> spikes = dspk.ops.filter.iir(x, 30_000.0, "highpass", 300.0)
>>> spikes.shape
(4, 30000)
median(
data,
width: int,
*,
edge: str = MEDIAN_DEFAULT_EDGE,
runtime: str | None = None,
) -> np.ndarray

Running median along time, as scipy.signal.medfilt (with edge="zeros").

Parameters

  • data : array_like
    Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
  • width : int
    Window length in samples, odd, at most MAX_MEDIAN_WIDTH.
  • edge : {"zeros", "odd", "reflect", "nearest"}, default MEDIAN_DEFAULT_EDGE (‘zeros’)
    What the samples past each end are taken to be.
  • runtime : str, default None
    A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

  • np.ndarray — The filtered signal (channels, samples), float32.

Raises

  • ValueError: For an even or too large width, an unknown edge, or data not 2-D.
  • RuntimeError: When the runtime has no device.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> baseline = dspk.ops.filter.median(x, 31)
notch(
data,
sample_rate: float,
freq: float,
q: float,
*,
mode: str = 'forward_backward',
start: str = 'rest',
runtime: str | None = None,
) -> np.ndarray

Removes one frequency (a second-order notch), as scipy.signal.iirnotch.

Parameters

  • data : array_like
    Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
  • sample_rate : float
    Sample rate of data in Hz.
  • freq : float
    Frequency to remove, in Hz (mains hum: 50 or 60).
  • q : float
    Quality factor: the notch is freq / q Hz wide at -3 dB; larger is narrower.
  • mode : {"forward_backward", "forward"}, default 'forward_backward'
    "forward_backward": zero phase, as scipy.signal.sosfiltfilt (the magnitude response is applied twice). "forward": causal, one pass, as scipy.signal.sosfilt.
  • start : {"rest", "steady_state"}, default 'rest'
    Where a "forward" pass starts: at rest, as if the signal were zero before the first sample (sosfilt without zi), or from the steady state of the first sample (zi = sosfilt_zi(sos) * x[0]: no step at the start). Forward-backward passes always start from the steady state.
  • runtime : str, default None
    A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

  • np.ndarray — The filtered signal (channels, samples), float32.

Raises

  • ValueError: When data is not 2-D.
  • RuntimeError: When the runtime has no device.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> clean = dspk.ops.filter.notch(x, 30_000.0, 60.0, 30.0)
settling_samples(sos: np.ndarray) -> int

Samples after which the impulse response of sos has settled.

The first sample t from which the rest of the response holds at most SETTLING_TOLERANCE of its total magnitude (sum(|h[t:]|) <= tolerance * sum(|h|)). It is how far a filter “remembers”: the padding zero-phase runs use and the halo windowed runs need.

Parameters

  • sos : numpy.ndarray
    Sections (sections, 6) in scipy’s layout.

Returns

  • int — A number of samples.

Examples

>>> import dspk
>>> n = dspk.ops.filter.settling_samples(dspk.ops.filter.sos("highpass", 300.0, 30_000.0))
sos(
band: str,
cutoff: float | Sequence[float],
sample_rate: float,
*,
order: int = DEFAULT_ORDER,
design: str = 'butterworth',
ripple_db: float | None = None,
) -> np.ndarray

Second-order sections of an IIR filter design, in scipy’s layout.

Parameters

  • band : {"lowpass", "highpass", "bandpass", "bandstop"}
    The band the filter passes (or stops).
  • cutoff : float or (float, float)
    Cutoff in Hz: one for "lowpass" and "highpass", (low, high) for "bandpass" and "bandstop".
  • sample_rate : float
    Sample rate in Hz.
  • order : int, default DEFAULT_ORDER (5)
    Filter order.
  • design : {"butterworth", "chebyshev1", "bessel"}, default 'butterworth'
    The filter family.
  • ripple_db : float, default None
    Passband ripple in dB; "chebyshev1" needs it, the other designs take none.

Returns

  • np.ndarray — Sections (sections, 6), each row [b0, b1, b2, a0, a1, a2] as scipy.signal.butter(..., output="sos") gives.

Raises

  • ValueError: For an unknown band or design, the wrong number of cutoffs, or ripple_db given to (or missing from) a design.

Examples

>>> import dspk
>>> sections = dspk.ops.filter.sos("bandpass", (300.0, 6000.0), 30_000.0, order=4)
sosfilt(
data,
sos: np.ndarray,
*,
mode: str = 'forward_backward',
start: str = 'rest',
runtime: str | None = None,
) -> np.ndarray

Applies second-order sections on the device.

Parameters

  • data : array_like
    Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
  • sos : numpy.ndarray
    Sections (sections, 6) in scipy’s layout, from sos or scipy.signal.
  • mode : {"forward_backward", "forward"}, default 'forward_backward'
    "forward_backward": zero phase, as scipy.signal.sosfiltfilt (the magnitude response is applied twice). "forward": causal, one pass, as scipy.signal.sosfilt.
  • start : {"rest", "steady_state"}, default 'rest'
    Where a "forward" pass starts: at rest, as if the signal were zero before the first sample (sosfilt without zi), or from the steady state of the first sample (zi = sosfilt_zi(sos) * x[0]: no step at the start). Forward-backward passes always start from the steady state.
  • runtime : str, default None
    A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

  • np.ndarray — The filtered signal (channels, samples), float32.

Raises

  • ValueError: For an invalid mode or start, or data not 2-D.
  • RuntimeError: When the runtime has no device.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> sections = dspk.ops.filter.sos("lowpass", 250.0, 30_000.0)
>>> lfp = dspk.ops.filter.sosfilt(x, sections)
subtract_template(
data,
template,
events,
center_offset: int,
*,
max_lag: int = TEMPLATE_DEFAULT_MAX_LAG,
dynamic_scaling: bool = True,
runtime: str | None = None,
) -> np.ndarray

Subtracts a template from data at every event, each aligned and scaled.

The template’s sample center_offset is placed on each event, then shifted by up to max_lag samples to where it best matches (cross-correlation) and, with dynamic_scaling, scaled by least squares before it is subtracted.

Parameters

  • data : array_like
    Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
  • template : array_like
    The waveform (channels, template_samples), or 1-D for one channel.
  • events : array_like
    Sample indices of the events (1-D).
  • center_offset : int
    The template sample that sits on each event.
  • max_lag : int, default TEMPLATE_DEFAULT_MAX_LAG (8)
    How far, in samples, an event may shift to align.
  • dynamic_scaling : bool, default True
    Scale the template to each event (least squares); False subtracts it as it is.
  • runtime : str, default None
    A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

  • np.ndarray — data with the events removed (channels, samples), float32.

Raises

  • ValueError: When data is not 2-D.
  • RuntimeError: When the runtime has no device.

Notes

Overlapping events are subtracted in order, each aligned on what the earlier ones left, so where waveforms overlap a remainder can stay (about 10 % for two events 5 samples apart in the tests). Isolated events are removed to well under 1 %.

teager_kaiser(
data,
*,
edge: str = TEAGER_KAISER_DEFAULT_EDGE,
runtime: str | None = None,
) -> np.ndarray

Teager–Kaiser energy along time: x[t]**2 - x[t-1] * x[t+1].

It is large where the signal is both large and fast, so spikes stand out from slow waves.

Parameters

  • data : array_like
    Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
  • edge : {"zeros", "odd", "reflect", "nearest"}, default TEAGER_KAISER_DEFAULT_EDGE (‘reflect’)
    What the neighbours of the first and last samples are taken to be.
  • runtime : str, default None
    A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

  • np.ndarray — The energy (channels, samples), float32.

Raises

  • ValueError: For an unknown edge or data not 2-D.
  • RuntimeError: When the runtime has no device.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> energy = dspk.ops.filter.teager_kaiser(x)
Name Type Value Description
DEFAULT_ORDER int 5
FIR_DEFAULT_EDGE str 'zeros'
GAUSSIAN_DEFAULT_EDGE str 'reflect'
GAUSSIAN_TRUNCATE float 4.0
MAX_MEDIAN_WIDTH int 31
MEDIAN_DEFAULT_EDGE str 'zeros'
SETTLING_TOLERANCE float 0.001
TEAGER_KAISER_DEFAULT_EDGE str 'reflect'
TEMPLATE_DEFAULT_MAX_LAG int 8