dspk.ops.filter
Filters on the device, one module per family (mirroring dspk_ops::filter).
- IIR designs and filtering with
scipy.signalsemantics:sos,iir,notch,sosfilt,settling_samples. - FIR filtering, design and Gaussian smoothing:
fir,firwin,gaussian_smooth. - Non-linear:
median,teager_kaiser. - Template subtraction:
subtract_template.
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. |
Functions
Section titled “Functions”fir( data, taps, *, centered: bool = False, edge: str = FIR_DEFAULT_EDGE, runtime: str | None = None,) -> np.ndarrayFilters 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, defaultFalse
False: causal,y[t] = sum(taps[k] * x[t - k]), asscipy.signal.lfilter(taps, 1, x)withedge="zeros".True: the taps (an odd number) centred on each sample, zero phase for symmetric taps.edge:{"zeros", "odd", "reflect", "nearest"}, defaultFIR_DEFAULT_EDGE(‘zeros’)
What the samples past each end are taken to be.runtime:str, defaultNone
A runtime fromruntime.available("cuda","hip","wgpu","cpu") for this call. By default: the one chosen withruntime.set, elseDSPK_RUNTIME, else the first available (GPUs first).
Returns
np.ndarray— The filtered signal(channels, samples), float32.
Raises
ValueError: For emptytaps, an even number of centred taps, an unknownedge, ordatanot 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
Section titled “firwin”firwin( num_taps: int, cutoff: float, *, sample_rate: float | None = None, window: str = 'hamming', beta: float | None = None,) -> np.ndarrayLow-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 whensample_rateis given, else a fraction of Nyquist in(0, 1].sample_rate:float, defaultNone
Sample rate in Hz.window:{"hamming", "kaiser"}, default'hamming'
The window shaping the taps.beta:float, defaultNone
Kaiser shape parameter;"kaiser"needs it,"hamming"takes none.
Returns
np.ndarray— Taps(num_taps,).
Raises
ValueError: Fornum_tapsbelow 1, a cutoff outside(0, Nyquist], a non-positivesample_rate, an unknownwindow, orbetagiven to (or missing from) a window.
Examples
>>> import dspk>>> taps = dspk.ops.filter.firwin(101, 250.0, sample_rate=30_000.0)gaussian_smooth
Section titled “gaussian_smooth”gaussian_smooth( data, sigma_samples: float, *, edge: str = GAUSSIAN_DEFAULT_EDGE, runtime: str | None = None,) -> np.ndarrayZero-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"}, defaultGAUSSIAN_DEFAULT_EDGE(‘reflect’)
What the samples past each end are taken to be (scipy’smode).runtime:str, defaultNone
A runtime fromruntime.available("cuda","hip","wgpu","cpu") for this call. By default: the one chosen withruntime.set, elseDSPK_RUNTIME, else the first available (GPUs first).
Returns
np.ndarray— The filtered signal(channels, samples), float32.
Raises
ValueError: For a non-positivesigma_samples, an unknownedge, ordatanot 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.ndarrayDesigns 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 ofdatain 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, defaultDEFAULT_ORDER(5)
Filter order (a zero-phase run applies it twice).design:{"butterworth", "chebyshev1", "bessel"}, default'butterworth'
The filter family.ripple_db:float, defaultNone
Passband ripple in dB;"chebyshev1"needs it.mode:{"forward_backward", "forward"}, default'forward_backward'
"forward_backward": zero phase, asscipy.signal.sosfiltfilt(the magnitude response is applied twice)."forward": causal, one pass, asscipy.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 (sosfiltwithoutzi), 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, defaultNone
A runtime fromruntime.available("cuda","hip","wgpu","cpu") for this call. By default: the one chosen withruntime.set, elseDSPK_RUNTIME, else the first available (GPUs first).
Returns
np.ndarray— The filtered signal(channels, samples), float32.
Raises
ValueError: For an invalid design (seesos),modeorstart, ordatanot 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
Section titled “median”median( data, width: int, *, edge: str = MEDIAN_DEFAULT_EDGE, runtime: str | None = None,) -> np.ndarrayRunning 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 mostMAX_MEDIAN_WIDTH.edge:{"zeros", "odd", "reflect", "nearest"}, defaultMEDIAN_DEFAULT_EDGE(‘zeros’)
What the samples past each end are taken to be.runtime:str, defaultNone
A runtime fromruntime.available("cuda","hip","wgpu","cpu") for this call. By default: the one chosen withruntime.set, elseDSPK_RUNTIME, else the first available (GPUs first).
Returns
np.ndarray— The filtered signal(channels, samples), float32.
Raises
ValueError: For an even or too largewidth, an unknownedge, ordatanot 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.ndarrayRemoves 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 ofdatain Hz.freq:float
Frequency to remove, in Hz (mains hum: 50 or 60).q:float
Quality factor: the notch isfreq / qHz wide at -3 dB; larger is narrower.mode:{"forward_backward", "forward"}, default'forward_backward'
"forward_backward": zero phase, asscipy.signal.sosfiltfilt(the magnitude response is applied twice)."forward": causal, one pass, asscipy.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 (sosfiltwithoutzi), 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, defaultNone
A runtime fromruntime.available("cuda","hip","wgpu","cpu") for this call. By default: the one chosen withruntime.set, elseDSPK_RUNTIME, else the first available (GPUs first).
Returns
np.ndarray— The filtered signal(channels, samples), float32.
Raises
ValueError: Whendatais 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
Section titled “settling_samples”settling_samples(sos: np.ndarray) -> intSamples 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.ndarraySecond-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, defaultDEFAULT_ORDER(5)
Filter order.design:{"butterworth", "chebyshev1", "bessel"}, default'butterworth'
The filter family.ripple_db:float, defaultNone
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]asscipy.signal.butter(..., output="sos")gives.
Raises
ValueError: For an unknownbandordesign, the wrong number of cutoffs, orripple_dbgiven 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
Section titled “sosfilt”sosfilt( data, sos: np.ndarray, *, mode: str = 'forward_backward', start: str = 'rest', runtime: str | None = None,) -> np.ndarrayApplies 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, fromsosorscipy.signal.mode:{"forward_backward", "forward"}, default'forward_backward'
"forward_backward": zero phase, asscipy.signal.sosfiltfilt(the magnitude response is applied twice)."forward": causal, one pass, asscipy.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 (sosfiltwithoutzi), 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, defaultNone
A runtime fromruntime.available("cuda","hip","wgpu","cpu") for this call. By default: the one chosen withruntime.set, elseDSPK_RUNTIME, else the first available (GPUs first).
Returns
np.ndarray— The filtered signal(channels, samples), float32.
Raises
ValueError: For an invalidmodeorstart, ordatanot 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
Section titled “subtract_template”subtract_template( data, template, events, center_offset: int, *, max_lag: int = TEMPLATE_DEFAULT_MAX_LAG, dynamic_scaling: bool = True, runtime: str | None = None,) -> np.ndarraySubtracts 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, defaultTEMPLATE_DEFAULT_MAX_LAG(8)
How far, in samples, an event may shift to align.dynamic_scaling:bool, defaultTrue
Scale the template to each event (least squares);Falsesubtracts it as it is.runtime:str, defaultNone
A runtime fromruntime.available("cuda","hip","wgpu","cpu") for this call. By default: the one chosen withruntime.set, elseDSPK_RUNTIME, else the first available (GPUs first).
Returns
np.ndarray—datawith the events removed(channels, samples), float32.
Raises
ValueError: Whendatais 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
Section titled “teager_kaiser”teager_kaiser( data, *, edge: str = TEAGER_KAISER_DEFAULT_EDGE, runtime: str | None = None,) -> np.ndarrayTeager–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"}, defaultTEAGER_KAISER_DEFAULT_EDGE(‘reflect’)
What the neighbours of the first and last samples are taken to be.runtime:str, defaultNone
A runtime fromruntime.available("cuda","hip","wgpu","cpu") for this call. By default: the one chosen withruntime.set, elseDSPK_RUNTIME, else the first available (GPUs first).
Returns
np.ndarray— The energy(channels, samples), float32.
Raises
ValueError: For an unknownedgeordatanot 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)