From a2838eaf339acd484051660134c2f05a3bef57ba Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 17:58:22 -0400 Subject: [PATCH 1/2] BinnedAggregate: add a passthrough flag There was no way to leave this node in a graph but switch it off. Every other conditioning stage has one -- CommonRereference has mode="passthrough", Butterworth has order=0, SamplingDelayAlignment has filter_len=0, Slicer has an empty selection -- so a pipeline that wants to compare with and without binning had to rewire itself instead. A separate flag rather than a sentinel bin_duration, because the intended use is a runtime toggle: a consumer switching this off because the view zoomed to a range where binning would cost detail should not have to remember and restore the rate itself. A sentinel would have destroyed it. passthrough is part of _hash_message, so flipping it resets the schedule and the carry. That matters: the carry holds an open partial bin, and if it survived the gap the first bin after resuming would mix samples from either side of it. There is a test that fails without the hash change. Note the shape consequence, which is documented on the setting: with a tuple operation, toggling adds and removes a trailing axis. Downstream has to absorb a rank change -- a fixed-layout sink reallocates, a plot rebuilds. ezmsg-tools' shmem bridge and sweep widget both do, but neither does it for free. --- src/ezmsg/sigproc/binned_aggregate.py | 27 ++++++++++++++++- tests/unit/test_binned_aggregate.py | 42 +++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/src/ezmsg/sigproc/binned_aggregate.py b/src/ezmsg/sigproc/binned_aggregate.py index faba76f..43b46b2 100644 --- a/src/ezmsg/sigproc/binned_aggregate.py +++ b/src/ezmsg/sigproc/binned_aggregate.py @@ -85,6 +85,19 @@ class BinnedAggregateSettings(ez.Settings): 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. + + 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`` @@ -133,9 +146,18 @@ class BinnedAggregateTransformer( """ def _hash_message(self, message: AxisArray) -> int: - return hash((message.axes[self.settings.axis].gain, message.key)) + # passthrough is in the hash so flipping it resets the schedule and the + # carry. Without that, switching back on would splice samples from + # before the gap onto the first bin after it. + return hash((message.axes[self.settings.axis].gain, message.key, self.settings.passthrough)) def _reset_state(self, message: AxisArray) -> None: + if self.settings.passthrough: + # No schedule to build, and nothing may be carried across the gap. + self._state.schedule = None + self._state.carry = None + self._state.metric_axis = None + return axis_info = message.get_axis(self.settings.axis) schedule = BinSchedule( bin_duration=self.settings.bin_duration, @@ -219,6 +241,9 @@ def _empty_like(self, message: AxisArray, axis_idx: int, step: BinStep) -> AxisA ) def _process(self, message: AxisArray) -> AxisArray: + if self.settings.passthrough: + return message + axis = self.settings.axis axis_info = message.get_axis(axis) axis_idx = message.get_axis_idx(axis) diff --git a/tests/unit/test_binned_aggregate.py b/tests/unit/test_binned_aggregate.py index 0ca4c93..6dee475 100644 --- a/tests/unit/test_binned_aggregate.py +++ b/tests/unit/test_binned_aggregate.py @@ -487,3 +487,45 @@ def test_tuple_may_mix_value_and_coordinate_operations(): assert list(out.axes["metric"].data) == ["max", "argmax"] np.testing.assert_allclose(out.data[:, 0, 0], [3.0, 5.0]) # peak values np.testing.assert_allclose(out.data[:, 0, 1], [0.007, 0.025]) # peak times + + +# ---- passthrough ----------------------------------------------------------- + + +def replace_settings(settings, **kw): + """Mimic a runtime settings push.""" + import dataclasses + + return dataclasses.replace(settings, **kw) + + +def test_passthrough_forwards_untouched(): + proc = BinnedAggregateTransformer(axis="time", bin_duration=0.02, operation=MINMAX, passthrough=True) + msg = _sig_msgs(np.arange(100, dtype=float).reshape(50, 2), 1000.0, 50)[0] + out = proc(msg) + assert out is msg + + +def test_passthrough_keeps_the_bin_rate_for_when_it_is_switched_back(): + """A sentinel bin_duration would have destroyed the rate; a separate flag + means a caller toggling at runtime need not remember and restore it.""" + proc = BinnedAggregateTransformer(axis="time", bin_duration=0.02, operation=MINMAX, passthrough=True) + assert proc.settings.bin_duration == pytest.approx(0.02) + + +def test_toggling_off_does_not_splice_stale_samples_into_the_first_bin(): + """The carry holds an open partial bin. If it survived a passthrough gap, + the first bin after would mix samples from either side of it.""" + fs = 1000.0 + proc = BinnedAggregateTransformer(axis="time", bin_duration=0.02, operation=AggregationFunction.MAX) + + # 25 samples: bin 0 closes, 5 are carried. + proc(_sig_msgs(np.full((25, 1), 99.0), fs, 25)[0]) + + # Pass through a stretch, then resume. The 99s must not reappear. + proc.settings = replace_settings(proc.settings, passthrough=True) + proc(_sig_msgs(np.zeros((25, 1)), fs, 25)[0]) + proc.settings = replace_settings(proc.settings, passthrough=False) + out = proc(_sig_msgs(np.zeros((40, 1)), fs, 40)[0]) + + assert out.data.max() == pytest.approx(0.0) From 46e9c92d1184bc373405b06ea821ab0c304a0b9d Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Thu, 6 Aug 2026 01:21:24 -0400 Subject: [PATCH 2/2] BinnedAggregate: short-circuit passthrough before the state Putting passthrough in _hash_message worked, but it paid for the toggle on every message: each one hashed the flag, and the passthrough branches in _reset_state and _process existed only to describe a node that is supposed to be doing nothing at all. A message arriving in passthrough still walked the whole stateful path to be handed back unchanged. Short-circuit in __call__/__acall__ instead, matching EWMATransformer and AdaptiveStandardScalerTransformer, so passthrough returns the input before hashing and the state is never consulted. That removes the three branches. The reason passthrough was in the hash in the first place still holds: the carry holds an open partial bin, and if it survived the gap the first bin after resuming would mix samples from either side of it. Short-circuiting alone would have reintroduced exactly that, since _hash keeps its pre-gap value when nothing hashes it. So the passthrough branch calls _request_reset(), which invalidates _hash directly -- the next real message is guaranteed to take the reset path. _request_reset() goes in __call__ rather than only in update_settings so the guard also holds when settings are pushed by assignment (proc.settings = replace(...)) rather than through update_settings. That is what the existing toggle test does, and it is the path a runtime consumer is most likely to take. passthrough joins NONRESET_SETTINGS_FIELDS to say that update_settings need not queue a reset of its own. Note this is the opposite of what EWMA and the scaler do -- they resume from pre-gap state on purpose. The difference is that their state is a converged estimate, while the carry is raw samples that would be spliced into an output bin. Whether theirs is right is #195. --- src/ezmsg/sigproc/binned_aggregate.py | 34 +++++++++++------- tests/unit/test_binned_aggregate.py | 52 +++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 13 deletions(-) diff --git a/src/ezmsg/sigproc/binned_aggregate.py b/src/ezmsg/sigproc/binned_aggregate.py index 43b46b2..0d940a4 100644 --- a/src/ezmsg/sigproc/binned_aggregate.py +++ b/src/ezmsg/sigproc/binned_aggregate.py @@ -93,6 +93,9 @@ class BinnedAggregateSettings(ez.Settings): (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 @@ -145,19 +148,27 @@ class BinnedAggregateTransformer( guaranteed to describe the same bins. """ + # `passthrough` is read live in `__call__`/`__acall__`, which short-circuit + # before the state is ever consulted; everything else is baked into the + # schedule and the metric axis during `_reset_state`. + NONRESET_SETTINGS_FIELDS = frozenset({"passthrough"}) + + def __call__(self, message: AxisArray) -> AxisArray: + if self.settings.passthrough: + self._request_reset() + return message + return super().__call__(message) + + async def __acall__(self, message: AxisArray) -> AxisArray: + if self.settings.passthrough: + self._request_reset() + return message + return await super().__acall__(message) + def _hash_message(self, message: AxisArray) -> int: - # passthrough is in the hash so flipping it resets the schedule and the - # carry. Without that, switching back on would splice samples from - # before the gap onto the first bin after it. - return hash((message.axes[self.settings.axis].gain, message.key, self.settings.passthrough)) + return hash((message.axes[self.settings.axis].gain, message.key)) def _reset_state(self, message: AxisArray) -> None: - if self.settings.passthrough: - # No schedule to build, and nothing may be carried across the gap. - self._state.schedule = None - self._state.carry = None - self._state.metric_axis = None - return axis_info = message.get_axis(self.settings.axis) schedule = BinSchedule( bin_duration=self.settings.bin_duration, @@ -241,9 +252,6 @@ def _empty_like(self, message: AxisArray, axis_idx: int, step: BinStep) -> AxisA ) def _process(self, message: AxisArray) -> AxisArray: - if self.settings.passthrough: - return message - axis = self.settings.axis axis_info = message.get_axis(axis) axis_idx = message.get_axis_idx(axis) diff --git a/tests/unit/test_binned_aggregate.py b/tests/unit/test_binned_aggregate.py index 6dee475..522dcec 100644 --- a/tests/unit/test_binned_aggregate.py +++ b/tests/unit/test_binned_aggregate.py @@ -513,6 +513,16 @@ def test_passthrough_keeps_the_bin_rate_for_when_it_is_switched_back(): assert proc.settings.bin_duration == pytest.approx(0.02) +def test_passthrough_skips_the_state_entirely(): + """The short-circuit happens before hashing, so a message that arrives in + passthrough neither builds a schedule nor grows the carry.""" + proc = BinnedAggregateTransformer(axis="time", bin_duration=0.02, operation=MINMAX, passthrough=True) + for msg in _sig_msgs(np.zeros((100, 2)), 1000.0, 25): + assert proc(msg) is msg + assert proc._state.schedule is None + assert proc._state.carry is None + + def test_toggling_off_does_not_splice_stale_samples_into_the_first_bin(): """The carry holds an open partial bin. If it survived a passthrough gap, the first bin after would mix samples from either side of it.""" @@ -521,11 +531,53 @@ def test_toggling_off_does_not_splice_stale_samples_into_the_first_bin(): # 25 samples: bin 0 closes, 5 are carried. proc(_sig_msgs(np.full((25, 1), 99.0), fs, 25)[0]) + assert proc._state.carry.shape[0] == 5 # Pass through a stretch, then resume. The 99s must not reappear. proc.settings = replace_settings(proc.settings, passthrough=True) proc(_sig_msgs(np.zeros((25, 1)), fs, 25)[0]) + # The stale carry is still in the state, but a reset is queued so it can + # never reach the next bin. + assert proc._hash == -1 proc.settings = replace_settings(proc.settings, passthrough=False) out = proc(_sig_msgs(np.zeros((40, 1)), fs, 40)[0]) assert out.data.max() == pytest.approx(0.0) + assert out.data.shape[0] == 2 # a fresh grid: 40 samples = exactly 2 bins + + +def test_toggling_off_via_update_settings_also_resets(): + """`passthrough` is in NONRESET_SETTINGS_FIELDS, so update_settings queues + no reset of its own -- the short-circuit in __call__ has to be what does.""" + fs = 1000.0 + settings = BinnedAggregateSettings(axis="time", bin_duration=0.02, operation=AggregationFunction.MAX) + proc = BinnedAggregateTransformer(settings=settings) + + proc(_sig_msgs(np.full((25, 1), 99.0), fs, 25)[0]) + + proc.update_settings(replace_settings(settings, passthrough=True)) + proc(_sig_msgs(np.zeros((25, 1)), fs, 25)[0]) + proc.update_settings(settings) + out = proc(_sig_msgs(np.zeros((40, 1)), fs, 40)[0]) + + assert out.data.max() == pytest.approx(0.0) + + +def test_toggling_restores_the_metric_axis(): + """A tuple operation adds a trailing axis; the gap must not leave it behind.""" + fs = 1000.0 + proc = BinnedAggregateTransformer(axis="time", bin_duration=0.02, operation=MINMAX) + + out = proc(_sig_msgs(np.zeros((40, 2)), fs, 40)[0]) + assert out.data.shape == (2, 2, 2) + assert list(out.axes["metric"].data) == ["min", "max"] + + proc.settings = replace_settings(proc.settings, passthrough=True) + msg = _sig_msgs(np.zeros((40, 2)), fs, 40)[0] + assert proc(msg) is msg + assert "metric" not in msg.axes + + proc.settings = replace_settings(proc.settings, passthrough=False) + out = proc(_sig_msgs(np.zeros((40, 2)), fs, 40)[0]) + assert out.data.shape == (2, 2, 2) + assert list(out.axes["metric"].data) == ["min", "max"]