ezmsg.sigproc.binned_aggregate#

Resample a signal axis to a lower rate by aggregating fixed-duration bins.

This is the dense-signal counterpart to the binning used by ezmsg.event.rate.EventRate. Both reduce a high-rate axis to a lower-rate “bin” axis, but historically they disagreed when bin_duration * fs is not an integer:

  • EventRate bins by a fractional samples_per_bin = bin_duration * fs with a carry accumulator, so each bin spans exactly bin_duration in the input’s time base and it labels the output gain as the nominal bin_duration.

  • Window (Pow -> Window -> Aggregate) bins by a fixed int(bin_duration * fs) sample count, so its gain is int(bin_duration * fs) / fs.

At a clean rate (e.g. 30000 Hz) those coincide, but at an off-nominal rate (e.g. 30012 Hz) they diverge in both gain and bin count, so two such streams never share a grid.

BinnedAggregateTransformer applies one or more arbitrary AggregationFunctions per bin instead of only counting – a tuple of functions stacks its results on a trailing axis, e.g. (MIN, MAX) for a display envelope – but it does not define its own bin boundaries: it drives them through the shared ezmsg.sigproc.util.binning.BinSchedule, the single source of truth for the grid. Any consumer that goes through the same schedule at the same bin_duration lands on the same grid by construction, so two such streams align downstream (e.g. with ezmsg.sigproc.merge.Merge). Set fractional=False to instead bin by a fixed int(bin_duration * fs) sample count (sample-locked grid, gain int(bin_duration * fs) / fs), matching Window.

Note

The schedule reproduces EventRate’s grid by shared code, not by a copied formula: BinSchedule is the boundary primitive both consume. test_bin_schedule.py pins fractional=True against a faithful port of EventRate’s algorithm and fractional=False against Window’s; the cross-package test against the real EventRate lives in ezmsg-event.

Classes

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

Bases: BaseTransformerUnit[BinnedAggregateSettings, AxisArray, AxisArray, BinnedAggregateTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of BinnedAggregateSettings

async on_signal(message)[source]#

Suppress empty publishes when a chunk spans less than one bin.

As with Downsample, most input chunks at a high input rate close no new bin, yielding a zero-length payload; broadcasting those wastes a round-trip across SHM/socket. Only emptiness along the binned axis is suppressed: a message that is empty along other axes (e.g. all channels sliced away upstream) still flows so downstream consumers keep its cadence.

Parameters:

message (AxisArray)

Return type:

AsyncGenerator

class BinnedAggregateSettings(axis='time', bin_duration=0.02, operation=AggregationFunction.MEAN, newaxis='metric', passthrough=False, fractional=True)[source]#

Bases: Settings

Settings for BinnedAggregate.

Parameters:
axis: str = 'time'#

The name of the axis to bin and aggregate along.

bin_duration: float = 0.02#

Output bin duration in seconds.

operation: AggregationFunction | tuple[AggregationFunction, ...] = 'mean'#

AggregationFunction applied within each bin.

A tuple applies several aggregations to the same bins and stacks the results on a new trailing axis named by newaxis, coordinate-labelled with each function’s value ("min", "max", …). (MIN, MAX) is the envelope used to put a high-rate signal on a screen without stride-decimation clipping the peaks: it keeps each bin’s extremes exactly, so a spike shows at full amplitude no matter where in the bin it fell.

A one-element tuple still produces the trailing axis – the output shape follows the type of this field, not the number of functions, so a caller that builds its tuple programmatically gets a stable shape.

newaxis: str = 'metric'#

Name of the trailing axis added when operation is a tuple.

passthrough: bool = False#

Forward messages untouched, as if this node were not in the graph.

A separate flag rather than a sentinel bin_duration so the bin rate survives being switched off and on – a consumer toggling this at runtime (say, because the view zoomed to a range where binning would cost detail) should not have to remember and restore the rate itself.

Switching back off starts a fresh schedule and an empty carry: nothing from before the gap is spliced onto the first bin after it.

Note that toggling changes the shape of the output: a tuple operation adds a trailing axis, and turning it off takes that axis away again. Everything downstream has to be able to absorb that – a fixed-layout sink will have to reallocate, and a plot will have to rebuild.

fractional: bool = True#

If True (default), bins span a fractional bin_duration * fs samples with a carry accumulator across chunks; each bin spans exactly bin_duration in the input’s time base and the output gain is the nominal bin_duration. This matches ezmsg.event.rate.EventRate. If False, bins span a fixed int(bin_duration * fs) samples and the output gain is int(bin_duration * fs) / fs (sample-locked, matching Window).

__init__(axis='time', bin_duration=0.02, operation=AggregationFunction.MEAN, newaxis='metric', passthrough=False, fractional=True)#
Parameters:
Return type:

None

class BinnedAggregateState[source]#

Bases: object

schedule: BinSchedule | None = None#

Shared bin-boundary schedule (see ezmsg.sigproc.util.binning). Owns the sample rate, samples-per-bin, output gain, global bin index, and carried sample count – the boundary arithmetic this transformer shares with EventRate. This transformer adds only the data carry below.

carry: Any = None#

Raw leftover samples of the open partial bin, carried across chunks so aggregation works for any operation (not just sums). Its length is kept in sync with schedule.carry_count.

metric_axis: CoordinateAxis | None = None#

The trailing axis attached to every multi-operation output.

It depends only on settings, so it is built once here rather than rebuilt per message – and since it is the same object every time, downstream identity checks on the axis stay cheap. None for a scalar operation.

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

Bases: BaseStatefulTransformer[BinnedAggregateSettings, AxisArray, AxisArray, BinnedAggregateState]

Bin a signal axis at a fixed bin rate and aggregate within each bin.

Unlike AggregateTransformer (which collapses a whole axis) or RangedAggregateTransformer (which aggregates static coordinate bands), this reduces a high-rate axis to a regularly-binned lower-rate axis, carrying the open partial bin across message boundaries.

settings.operation may be a tuple, in which case every function is applied to the same bins and the results stack on a trailing newaxis. The bins are computed once and sliced once, so N aggregations cost far less than N copies of this transformer, and – more importantly – they are guaranteed to describe the same bins.

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