BinnedAggregate: allow a tuple of operations, stacked on a metric axis - #193
Merged
Conversation
Reducing a high-rate signal for display wants each bin's min *and* max -- stride decimation lands between samples and clips the peaks, so a spike shrinks or vanishes depending on where the stride falls, whereas the two extremes preserve peak amplitude exactly. That is the only thing a dedicated min/max decimator would do that this transformer did not already do: bin on a shared schedule, carry the open partial bin across message boundaries, aggregate, label the output axis. Rather than write a second transformer with its own copy of that arithmetic, `operation` now accepts a tuple. Each function is applied to the same bins and the results stack on a trailing axis, `newaxis` (default "metric"), coordinate-labelled with the AggregationFunction values -- "min", "max" -- so a consumer reads names rather than relying on positional convention. Trailing rather than in place, so the binned axis keeps its position. Computing them together is not just less code: the bins are cut once and sliced once, so the results are identical by construction rather than by two transformers agreeing. There is a test comparing a tuple against separate single-op transformers over the same chunking to hold that. The output shape follows the *type* of `operation`, not the count. A scalar behaves exactly as before -- no trailing axis, no new key in `axes` -- so existing streams are untouched, and a one-element tuple does produce the axis, so a caller assembling its tuple programmatically gets a stable shape either way. The empty-payload path has to build the trailing axis at full width too; a zero-length message of the wrong rank will not concatenate with the messages around it, and at a high input rate most chunks close no bin, so that path is the common one rather than an edge case.
It depends only on settings -- the operations and the axis name -- so rebuilding it in _out_axes on every message meant constructing a numpy string array per message on the hot path, for a value that cannot have changed. Moved to _reset_state and held in state. Every output now carries the same object, so a downstream identity check on the axis is a pointer comparison rather than an array comparison.
This was referenced Aug 4, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Lets
BinnedAggregateSettings.operationbe a tuple, applying several aggregations to the same bins and stacking the results on a trailing axis.First of a stack of two. #194 builds on this branch.
Why
Putting a high-rate signal on a screen wants each bin's min and max. Stride decimation lands between samples and clips the peaks, so a spike shrinks or vanishes depending on where the stride falls; the two extremes preserve peak amplitude exactly. That was going to be a dedicated min/max decimator until it became clear the only thing it would do that
BinnedAggregatedoes not is apply two functions instead of one — everything else (binning on the sharedBinSchedule, carrying the open partial bin across message boundaries, labelling the output axis) already lives here and is better tested than a new transformer would have been.What
The trailing axis is named by
newaxis(default"metric") and coordinate-labelled with theAggregationFunctionvalues —"min","max"— so consumers read names rather than relying on position.Design notes
operation, not the count. A scalar behaves exactly as before — no trailing axis, no new key inaxes— so existing streams are untouched. A one-element tuple does produce the axis, so a caller assembling its tuple programmatically gets a stable shape either way._empty_likeis the common path, and a zero-length message of the wrong rank will not concatenate with the messages around it._reset_stateand held in state; every output carries the same object, which keeps a downstream identity check cheap.Tests
10 new, on top of the existing 20. 98 pass across
test_binned_aggregate,test_aggregate,test_bin_schedule, plustest_empty_gate(the only otherBinnedAggregateconsumer).