ezmsg.sigproc.window#

Functions

windowing(axis=None, newaxis=None, window_dur=None, window_shift=None, zero_pad_until='full', anchor=Anchor.BEGINNING)[source]#
Parameters:
Return type:

WindowTransformer

Classes

class Anchor(*values)[source]#

Bases: Enum

BEGINNING = 'beginning'#
END = 'end'#
MIDDLE = 'middle'#
class Window(*args, settings=None, **kwargs)[source]#

Bases: BaseTransformerUnit[WindowSettings, AxisArray, AxisArray, WindowTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of WindowSettings

INPUT_SIGNAL = InputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>]()#
OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#
async on_signal(message)[source]#

override superclass on_signal so we can opt to yield once or multiple times after dropping the win axis.

Parameters:

message (AxisArray)

Return type:

AsyncGenerator

class WindowSettings(axis: str | None = None, newaxis: str | None = None, window_dur: float | None = None, window_shift: float | None = None, zero_pad_until: str = 'full', anchor: str | ezmsg.sigproc.window.Anchor = <Anchor.BEGINNING: 'beginning'>)[source]#

Bases: Settings

Parameters:
axis: str | None = None#
newaxis: str | None = None#
window_dur: float | None = None#
window_shift: float | None = None#
zero_pad_until: str = 'full'#
anchor: str | Anchor = 'beginning'#
__init__(axis=None, newaxis=None, window_dur=None, window_shift=None, zero_pad_until='full', anchor=Anchor.BEGINNING)#
Parameters:
Return type:

None

class WindowState[source]#

Bases: object

buffer: ndarray[tuple[Any, ...], dtype[_ScalarT]] | SparseArray | None = None#
window_samples: int | None = None#
window_shift_samples: int | None = None#
shift_deficit: int = 0#

Number of incoming samples to ignore. Only relevant when shift > window.

newaxis_warned: bool = False#
out_newaxis: LinearAxis | None = None#
out_dims: list[str] | None = None#
class WindowTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[WindowSettings, AxisArray, AxisArray, WindowState]

Apply a sliding window along the specified axis to input streaming data. The windowing method is perhaps the most useful and versatile method in ezmsg.sigproc, but its parameterization can be difficult. Please read the argument descriptions carefully.

__init__(*args, **kwargs)[source]#
Parameters:

  • axis – The axis along which to segment windows. If None, defaults to the first dimension of the first seen AxisArray. Note: The windowed axis must be an AxisArray.LinearAxis, not an AxisArray.CoordinateAxis.

  • newaxis – New axis on which windows are delimited, immediately preceding the target windowed axis. The data length along newaxis may be 0 if this most recent push did not provide enough data for a new window. If window_shift is None then the newaxis length will always be 1.

  • window_dur – The duration of the window in seconds. If None, the function acts as a passthrough and all other parameters are ignored.

  • window_shift – The shift of the window in seconds. If None (default), windowing operates in “1:1 mode”, where each input yields exactly one most-recent window.

  • zero_pad_until –

    Determines how the function initializes the buffer. Can be one of “input” (default), “full”, “shift”, or “none”. If window_shift is None then this field is ignored and “input” is always used.

    • ”input” (default) initializes the buffer with the input then prepends with zeros to the window size. The first input will always yield at least one output.

    • ”shift” fills the buffer until window_shift. No outputs will be yielded until at least window_shift data has been seen.

    • ”none” does not pad the buffer. No outputs will be yielded until at least window_dur data has been seen.

  • anchor – Determines the entry in axis that gets assigned 0, which references the value in newaxis. Can be of class Anchor or a string representation of an Anchor.

Return type:

None

class Anchor(*values)[source]#

Bases: Enum

BEGINNING = 'beginning'#
END = 'end'#
MIDDLE = 'middle'#
class WindowSettings(axis: str | None = None, newaxis: str | None = None, window_dur: float | None = None, window_shift: float | None = None, zero_pad_until: str = 'full', anchor: str | ezmsg.sigproc.window.Anchor = <Anchor.BEGINNING: 'beginning'>)[source]#

Bases: Settings

Parameters:
axis: str | None = None#
newaxis: str | None = None#
window_dur: float | None = None#
window_shift: float | None = None#
zero_pad_until: str = 'full'#
anchor: str | Anchor = 'beginning'#
__init__(axis=None, newaxis=None, window_dur=None, window_shift=None, zero_pad_until='full', anchor=Anchor.BEGINNING)#
Parameters:
Return type:

None

class WindowState[source]#

Bases: object

buffer: ndarray[tuple[Any, ...], dtype[_ScalarT]] | SparseArray | None = None#
window_samples: int | None = None#
window_shift_samples: int | None = None#
shift_deficit: int = 0#

Number of incoming samples to ignore. Only relevant when shift > window.

newaxis_warned: bool = False#
out_newaxis: LinearAxis | None = None#
out_dims: list[str] | None = None#
class WindowTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[WindowSettings, AxisArray, AxisArray, WindowState]

Apply a sliding window along the specified axis to input streaming data. The windowing method is perhaps the most useful and versatile method in ezmsg.sigproc, but its parameterization can be difficult. Please read the argument descriptions carefully.

__init__(*args, **kwargs)[source]#
Parameters:

  • axis – The axis along which to segment windows. If None, defaults to the first dimension of the first seen AxisArray. Note: The windowed axis must be an AxisArray.LinearAxis, not an AxisArray.CoordinateAxis.

  • newaxis – New axis on which windows are delimited, immediately preceding the target windowed axis. The data length along newaxis may be 0 if this most recent push did not provide enough data for a new window. If window_shift is None then the newaxis length will always be 1.

  • window_dur – The duration of the window in seconds. If None, the function acts as a passthrough and all other parameters are ignored.

  • window_shift – The shift of the window in seconds. If None (default), windowing operates in “1:1 mode”, where each input yields exactly one most-recent window.

  • zero_pad_until –

    Determines how the function initializes the buffer. Can be one of “input” (default), “full”, “shift”, or “none”. If window_shift is None then this field is ignored and “input” is always used.

    • ”input” (default) initializes the buffer with the input then prepends with zeros to the window size. The first input will always yield at least one output.

    • ”shift” fills the buffer until window_shift. No outputs will be yielded until at least window_shift data has been seen.

    • ”none” does not pad the buffer. No outputs will be yielded until at least window_dur data has been seen.

  • anchor – Determines the entry in axis that gets assigned 0, which references the value in newaxis. Can be of class Anchor or a string representation of an Anchor.

Return type:

None

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

Bases: BaseTransformerUnit[WindowSettings, AxisArray, AxisArray, WindowTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of WindowSettings

INPUT_SIGNAL = InputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>]()#
OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#
async on_signal(message)[source]#

override superclass on_signal so we can opt to yield once or multiple times after dropping the win axis.

Parameters:

message (AxisArray)

Return type:

AsyncGenerator

windowing(axis=None, newaxis=None, window_dur=None, window_shift=None, zero_pad_until='full', anchor=Anchor.BEGINNING)[source]#
Parameters:
Return type:

WindowTransformer