ezmsg.event.peak#

Detects peaks in a signal.

Functions

threshold_crossing(threshold=-3.5, max_peak_dur=0.002, refrac_dur=0.001, align_on_peak=False, return_peak_val=False, auto_scale_tau=0.0)[source]#

Detect threshold crossing events.

Parameters:

  • threshold (float) – the value the signal must cross before the peak is found.

  • max_peak_dur (float) – The maximum duration of a peak in seconds.

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

  • align_on_peak (bool) – 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) – If True then the peak value is included in the EventMessage or sparse matrix payload.

  • auto_scale_tau (float) – If > 0, the data will be passed through a standard scaler prior to thresholding.

Return type:

Generator[list[EventMessage] | AxisArray, AxisArray, None]

Note: If either align_on_peak or return_peak_val are True then it is necessary to find the actual peak and not

just the threshold crossing. This will drastically increase the computational demand. It is recommended to tune max_peak_dur to a minimal-yet-reasonable value to limit the search space.

Returns:

A primed generator object that yields a list of EventMessage objects for every AxisArray it receives via send.

Parameters:
Return type:

Generator[list[EventMessage] | AxisArray, AxisArray, None]

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

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

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 ThresholdCrossing(*args, settings=None, **kwargs)[source]#

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

Parameters:

settings (Settings | None)

SETTINGS#

alias of ThresholdSettings

threshold_crossing(threshold=-3.5, max_peak_dur=0.002, refrac_dur=0.001, align_on_peak=False, return_peak_val=False, auto_scale_tau=0.0)[source]#

Detect threshold crossing events.

Parameters:

  • threshold (float) – the value the signal must cross before the peak is found.

  • max_peak_dur (float) – The maximum duration of a peak in seconds.

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

  • align_on_peak (bool) – 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) – If True then the peak value is included in the EventMessage or sparse matrix payload.

  • auto_scale_tau (float) – If > 0, the data will be passed through a standard scaler prior to thresholding.

Return type:

Generator[list[EventMessage] | AxisArray, AxisArray, None]

Note: If either align_on_peak or return_peak_val are True then it is necessary to find the actual peak and not

just the threshold crossing. This will drastically increase the computational demand. It is recommended to tune max_peak_dur to a minimal-yet-reasonable value to limit the search space.

Returns:

A primed generator object that yields a list of EventMessage objects for every AxisArray it receives via send.

Parameters:
Return type:

Generator[list[EventMessage] | AxisArray, AxisArray, None]