ezmsg.sigproc.window#

Sliding and tumbling window segmentation of streaming data.

Functions

windowing(axis=None, newaxis=None, window_dur=None, window_shift=None, zero_pad_until='full', anchor=Anchor.BEGINNING, batch_windows=False, buffer_update_strategy='immediate')[source]#
Parameters:

  • axis (str | None)

  • newaxis (str | None)

  • window_dur (float | None)

  • window_shift (float | None)

  • zero_pad_until (str)

  • anchor (str | Anchor)

  • batch_windows (bool)

  • buffer_update_strategy (Literal['immediate', 'threshold', 'on_demand'])

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[AxisArray]()#
OUTPUT_SIGNAL = OutputStream:unlocated[AxisArray](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#
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 | Anchor = <Anchor.BEGINNING: 'beginning'>, batch_windows: bool = False, buffer_update_strategy: Literal['immediate', 'threshold', 'on_demand']='immediate')[source]#

Bases: Settings

Parameters:

  • axis (str | None)

  • newaxis (str | None)

  • window_dur (float | None)

  • window_shift (float | None)

  • zero_pad_until (str)

  • anchor (str | Anchor)

  • batch_windows (bool)

  • buffer_update_strategy (Literal['immediate', 'threshold', 'on_demand'])

axis: str | None = None#
newaxis: str | None = None#

Name of the axis windows are delimited on, inserted before axis.

None (default) means the published messages carry no window axis: the Window unit yields one message per window, each exactly window_dur long with its own absolute offset. The transformer still emits a win axis in that case, because a transformer is 1-in/1-out and several windows may complete at once; the unit is what unbundles them.

Set batch_windows to trade that per-window guarantee for fewer, larger messages.

window_dur: float | None = None#
window_shift: float | None = None#
zero_pad_until: str = 'full'#
anchor: str | Anchor = 'beginning'#
batch_windows: bool = False#

Emit all complete windows as one contiguous message (“batcher mode”).

Only meaningful with newaxis=None and window_shift == window_dur, where consecutive windows tile the target axis exactly; requiring both is validated at construction.

Off by default, because a message of exactly window_dur is what asking for a window_dur most obviously means – and consumers with a fixed input size (an FFT, a binned aggregation) depend on it. Turn it on to re-chunk a stream purely for throughput, accepting that an emitted message is then a whole-number multiple of window_dur, since one oversized input can complete several windows at once.

buffer_update_strategy: Literal['immediate', 'threshold', 'on_demand'] = 'immediate'#

When the backlog copies incoming samples into its own memory. See ezmsg.sigproc.util.buffer.UpdateStrategy.

"immediate" (default, matching Sampler and Resample) copies on every write. That is what makes the buffer safe behind a cross-process link: ezmsg marshals with pickle protocol 5 out-of-band buffers, so message.data is a view into a shared-memory slot that the publisher recycles every num_buffers messages. Holding the array keeps the Python object alive but not its contents.

"on_demand" defers the copy, saving ~3 us per message at 256 channels. Only safe when nothing recycles the incoming buffer – i.e. a graph you know is single-process, where messages are passed by reference (put_local) rather than serialized.

__init__(axis=None, newaxis=None, window_dur=None, window_shift=None, zero_pad_until='full', anchor=Anchor.BEGINNING, batch_windows=False, buffer_update_strategy='immediate')#
Parameters:

  • axis (str | None)

  • newaxis (str | None)

  • window_dur (float | None)

  • window_shift (float | None)

  • zero_pad_until (str)

  • anchor (str | Anchor)

  • batch_windows (bool)

  • buffer_update_strategy (Literal['immediate', 'threshold', 'on_demand'])

Return type:

None

class WindowState[source]#

Bases: object

buffer: HybridBuffer | None = None#

Backlog of samples awaiting a complete window.

A HybridBuffer writing into preallocated memory, rather than re-growing one array per message. With 30-sample chunks feeding a 600-sample window, 19 of every 20 calls produce no output, and concatenate((buffer, new)) made each of those copy the whole backlog – ~25 us at 256 channels, 85% of the call. Copying the new chunk into a fixed allocation instead is ~2 us and independent of how much is already buffered.

Whether writes copy immediately is WindowSettings.buffer_update_strategy; the default copies, because incoming message data may be a view into memory the publisher recycles.

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

Fallback backlog for namespaces HybridBuffer can’t back.

pydata/sparse arrays have no item assignment, so they cannot be written into a preallocated buffer; those streams keep the original grow-by-concatenate behaviour. Sparse windowing is not a throughput path.

buffer_len: int = 0#

Samples buffered, mirrored here so the hot path skips a method call.

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_axis: LinearAxis | None = None#

Target axis re-anchored per anchor; constant for the life of the state.

out_dims: list[str] | None = None#
empty_out: ndarray[tuple[Any, ...], dtype[_ScalarT]] | SparseArray | None = None#

Cached zero-window output, returned unchanged whenever no window is due.

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.

Several windows can complete on one input, so the transformer – being 1-in/1-out – represents them along a win axis. What reaches subscribers depends on the settings:

settings

published messages

newaxis="win"

one message, win axis, N windows

newaxis=None

N messages, no win axis, each exactly window_dur

newaxis=None +

one message, no win axis,

batch_windows=True

N * window_dur contiguous samples

The last row is “batcher mode”, available only when window_shift == window_dur so that windows tile the target axis exactly. It is the only mode the transformer can produce without a win axis, because tiling is what makes concatenation lossless.

NONRESET_SETTINGS_FIELDS: ClassVar[frozenset[str]] = frozenset({'anchor'})#
__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

property is_batcher: bool#

Whether complete windows are emitted contiguously, with no win axis.