ezmsg.event.peak#

Detects peaks in a signal.

Note

This module supports the Array API standard, enabling use with NumPy, CuPy, PyTorch, and other compatible array libraries. Signal data operations are array-API compliant. Event detection and sparse output use NumPy regardless of input backend. Output is always sparse.COO.

Classes

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

Bases: BaseTransformerUnit[ThresholdSettings, AxisArray, AxisArray, ThresholdCrossingTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of ThresholdSettings

class ThresholdCrossingState[source]#

Bases: object

State for ThresholdCrossingTransformer.

max_width: int = 0#
min_width: int = 1#
refrac_width: int = 0#
scaler: AdaptiveStandardScalerTransformer | None = None#

Object performing adaptive z-scoring.

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

Trailing buffer in case peak spans sample chunks. Only used if align_on_peak or return_peak_val.

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

Keep track of the raw data so we can return_peak_val. Only needed if using the scaler.

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

Track number of samples since last event for each feature. Used especially for refractory period.

class ThresholdCrossingTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[ThresholdSettings, AxisArray, AxisArray, ThresholdCrossingState]

Transformer that detects threshold crossing events.

class ThresholdSettings(threshold: float = -3.5, max_peak_dur: float = 0.002, min_peak_dur: float = 0.0, refrac_dur: float = 0.001, align_on_peak: bool = False, return_peak_val: bool = False, auto_scale_tau: float = 0.0)[source]#

Bases: Settings

Parameters:
threshold: float = -3.5#

the value the signal must cross before the peak is found.

max_peak_dur: float = 0.002#

The maximum duration of a peak in seconds.

min_peak_dur: float = 0.0#

The minimum duration of a peak in seconds. If 0 (default), no minimum duration is enforced.

refrac_dur: float = 0.001#

The minimum duration between peaks in seconds. If 0 (default), no refractory period is enforced.

align_on_peak: bool = False#

If False (default), the returned sample index indicates the first sample across threshold. If True, the sample index indicates the sample with the largest deviation after threshold crossing.

return_peak_val: bool = False#

If True then the peak value is included in the EventMessage or sparse matrix payload.

auto_scale_tau: float = 0.0#

If > 0, the data will be passed through a standard scaler prior to thresholding.

__init__(threshold=-3.5, max_peak_dur=0.002, min_peak_dur=0.0, refrac_dur=0.001, align_on_peak=False, return_peak_val=False, auto_scale_tau=0.0)#
Parameters:
Return type:

None