ezmsg.sigproc.rollingscaler#

Classes

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

Bases: BaseAdaptiveTransformer[RollingScalerSettings, AxisArray, AxisArray, RollingScalerState]

Processor for rolling z-score normalization of input AxisArray messages.

The processor maintains rolling statistics (mean and variance) over the last k_samples samples received via the partial_fit() method. When processing an AxisArray message, it normalizes the data using the current rolling statistics.

The input AxisArray messages are expected to have shape (time, ch), where ch is the channel axis. The processor computes the z-score for each channel independently.

Note: You should consider instead using the AdaptiveStandardScalerTransformer which is computationally more efficient and uses less memory. This RollingScalerProcessor is primarily provided to reproduce processing in the literature.

Settings:#

k_samples: int

Number of previous samples to use for rolling statistics.

Example:#

```python processor = RollingScalerProcessor(

settings=RollingScalerSettings(

k_samples=20 # Number of previous samples to use for rolling statistics

)

)#

partial_fit(message)[source]#
Parameters:

message (SampleMessage)

Return type:

None

class RollingScalerSettings(axis: str = 'time', k_samples: int | None = 20, window_size: float | None = None, update_with_signal: bool = False, min_samples: int = 1, min_seconds: float = 1.0, artifact_z_thresh: float | None = None, clip: float | None = 10.0)[source]#

Bases: Settings

Parameters:

  • axis (str)

  • k_samples (int | None)

  • window_size (float | None)

  • update_with_signal (bool)

  • min_samples (int)

  • min_seconds (float)

  • artifact_z_thresh (float | None)

  • clip (float | None)

axis: str = 'time'#

Axis along which samples are arranged.

k_samples: int | None = 20#

Rolling window size in number of samples.

window_size: float | None = None#

Rolling window size in seconds. If set, overrides k_samples. update_with_signal likely should be True if using this option.

update_with_signal: bool = False#

If True, update rolling statistics using the incoming process stream.

min_samples: int = 1#

Minimum number of samples required to compute statistics. Used when window_size is not set.

min_seconds: float = 1.0#

Minimum duration in seconds required to compute statistics. Used when window_size is set.

artifact_z_thresh: float | None = None#

Threshold for z-score based artifact detection. If set, samples with any channel exceeding this z-score will be excluded from updating the rolling statistics.

clip: float | None = 10.0#

If set, clip the output values to the range [-clip, clip].

__init__(axis='time', k_samples=20, window_size=None, update_with_signal=False, min_samples=1, min_seconds=1.0, artifact_z_thresh=None, clip=10.0)#
Parameters:

  • axis (str)

  • k_samples (int | None)

  • window_size (float | None)

  • update_with_signal (bool)

  • min_samples (int)

  • min_seconds (float)

  • artifact_z_thresh (float | None)

  • clip (float | None)

Return type:

None

class RollingScalerState[source]#

Bases: object

mean: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#
N: int = 0#
M2: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#
samples: deque | None = None#
k_samples: int | None = None#
min_samples: int | None = None#
class RollingScalerUnit(*args, settings=None, **kwargs)[source]#

Bases: BaseAdaptiveTransformerUnit[RollingScalerSettings, AxisArray, AxisArray, RollingScalerProcessor]

Unit wrapper for RollingScalerProcessor.

This unit performs rolling z-score normalization on incoming AxisArray messages. The unit maintains rolling statistics (mean and variance) over the last k_samples samples received. When processing an AxisArray message, it normalizes the data using the current rolling statistics.

Example:#

```python unit = RollingScalerUnit(

settings=RollingScalerSettings(

k_samples=20 # Number of previous samples to use for rolling statistics

)

)#

SETTINGS#

alias of RollingScalerSettings

Parameters:

settings (Settings | None)

class RollingScalerSettings(axis: str = 'time', k_samples: int | None = 20, window_size: float | None = None, update_with_signal: bool = False, min_samples: int = 1, min_seconds: float = 1.0, artifact_z_thresh: float | None = None, clip: float | None = 10.0)[source]#

Bases: Settings

Parameters:

  • axis (str)

  • k_samples (int | None)

  • window_size (float | None)

  • update_with_signal (bool)

  • min_samples (int)

  • min_seconds (float)

  • artifact_z_thresh (float | None)

  • clip (float | None)

axis: str = 'time'#

Axis along which samples are arranged.

k_samples: int | None = 20#

Rolling window size in number of samples.

window_size: float | None = None#

Rolling window size in seconds. If set, overrides k_samples. update_with_signal likely should be True if using this option.

update_with_signal: bool = False#

If True, update rolling statistics using the incoming process stream.

min_samples: int = 1#

Minimum number of samples required to compute statistics. Used when window_size is not set.

min_seconds: float = 1.0#

Minimum duration in seconds required to compute statistics. Used when window_size is set.

artifact_z_thresh: float | None = None#

Threshold for z-score based artifact detection. If set, samples with any channel exceeding this z-score will be excluded from updating the rolling statistics.

clip: float | None = 10.0#

If set, clip the output values to the range [-clip, clip].

__init__(axis='time', k_samples=20, window_size=None, update_with_signal=False, min_samples=1, min_seconds=1.0, artifact_z_thresh=None, clip=10.0)#
Parameters:

  • axis (str)

  • k_samples (int | None)

  • window_size (float | None)

  • update_with_signal (bool)

  • min_samples (int)

  • min_seconds (float)

  • artifact_z_thresh (float | None)

  • clip (float | None)

Return type:

None

class RollingScalerState[source]#

Bases: object

mean: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#
N: int = 0#
M2: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#
samples: deque | None = None#
k_samples: int | None = None#
min_samples: int | None = None#
class RollingScalerProcessor(*args, **kwargs)[source]#

Bases: BaseAdaptiveTransformer[RollingScalerSettings, AxisArray, AxisArray, RollingScalerState]

Processor for rolling z-score normalization of input AxisArray messages.

The processor maintains rolling statistics (mean and variance) over the last k_samples samples received via the partial_fit() method. When processing an AxisArray message, it normalizes the data using the current rolling statistics.

The input AxisArray messages are expected to have shape (time, ch), where ch is the channel axis. The processor computes the z-score for each channel independently.

Note: You should consider instead using the AdaptiveStandardScalerTransformer which is computationally more efficient and uses less memory. This RollingScalerProcessor is primarily provided to reproduce processing in the literature.

Settings:#

k_samples: int

Number of previous samples to use for rolling statistics.

Example:#

```python processor = RollingScalerProcessor(

settings=RollingScalerSettings(

k_samples=20 # Number of previous samples to use for rolling statistics

)

)#

partial_fit(message)[source]#
Parameters:

message (SampleMessage)

Return type:

None

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

Bases: BaseAdaptiveTransformerUnit[RollingScalerSettings, AxisArray, AxisArray, RollingScalerProcessor]

Unit wrapper for RollingScalerProcessor.

This unit performs rolling z-score normalization on incoming AxisArray messages. The unit maintains rolling statistics (mean and variance) over the last k_samples samples received. When processing an AxisArray message, it normalizes the data using the current rolling statistics.

Example:#

```python unit = RollingScalerUnit(

settings=RollingScalerSettings(

k_samples=20 # Number of previous samples to use for rolling statistics

)

)#

SETTINGS#

alias of RollingScalerSettings

Parameters:

settings (Settings | None)