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 an arbitrary AggregationFunction per bin instead of only counting, 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, 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 = 'mean'#

AggregationFunction applied within each bin.

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, 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.

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.