ezmsg.sigproc.ewma#

Exponentially weighted moving average (EWMA) utilities and parameter conversion.

Functions

ewma_step(sample, zi, alpha, beta=None)[source]#

Do an exponentially weighted moving average step.

Parameters:

sample (ndarray[tuple[Any, ...], dtype[_ScalarT]]) – The new sample.
zi (ndarray[tuple[Any, ...], dtype[_ScalarT]]) – The output of the previous step.
alpha (float) – Fading factor.
beta (float | None) – Persisting factor. If None, it is calculated as 1-alpha.

Returns:

alpha * sample + beta * zi

Classes

class EWMASettings(time_constant: float = 1.0, axis: str | None = None, accumulate: bool = True)[source]#

Bases: Settings

Parameters:

time_constant (float)
axis (str | None)
accumulate (bool)

time_constant: float = 1.0#: The amount of time for the smoothed response of a unit step function to reach 1 - 1/e approx-eq 63.2%.

axis: str | None = None#

accumulate: bool = True#: If True, update the EWMA state with each sample. If False, only apply the current EWMA estimate without updating state (useful for inference periods where you don’t want to adapt statistics).

__init__(time_constant=1.0, axis=None, accumulate=True)#

Parameters:

time_constant (float)
axis (str | None)
accumulate (bool)

Return type:

None

class EWMAState[source]#

Bases: object

alpha: float#

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

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

Bases: BaseStatefulTransformer[EWMASettings, AxisArray, AxisArray, EWMAState]

NONRESET_SETTINGS_FIELDS: ClassVar[frozenset[str]] = frozenset({'accumulate'})#

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

Bases: BaseTransformerUnit[EWMASettings, AxisArray, AxisArray, EWMATransformer]

Parameters:: settings (Settings | None)

SETTINGS#: alias of EWMASettings

class EWMA_Deprecated(alpha, max_len)[source]#

Bases: object

Grabbed these methods from https://stackoverflow.com/a/70998068 and other answers in that topic, but they ended up being slower than the scipy.signal.lfilter method. Additionally, compute and compute2 suffer from potential errors as the vector length increases and beta**n approaches zero.

Parameters:

alpha (float)
max_len (int)

__init__(alpha, max_len)[source]#

Parameters:

alpha (float)
max_len (int)

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

compute(arr, out=None)[source]#

Parameters:

arr (ndarray[tuple[Any, ...], dtype[_ScalarT]])
out (ndarray[tuple[Any, ...], dtype[_ScalarT]] | None)

Return type:

ndarray[tuple[Any, …], dtype[_ScalarT]]

compute2(arr)[source]#

Compute the Exponentially Weighted Moving Average (EWMA) of the input array.

Parameters:: arr (ndarray[tuple[Any, ...], dtype[_ScalarT]]) – The input array to be smoothed.
Returns:: The smoothed array.
Return type:: ndarray[tuple[Any, …], dtype[_ScalarT]]

compute_sample(new_sample)[source]#

Parameters:: new_sample (ndarray[tuple[Any, ...], dtype[_ScalarT]])
Return type:: ndarray[tuple[Any, …], dtype[_ScalarT]]