diff --git a/src/ezmsg/sigproc/binned_aggregate.py b/src/ezmsg/sigproc/binned_aggregate.py index faba76f..0d940a4 100644 --- a/src/ezmsg/sigproc/binned_aggregate.py +++ b/src/ezmsg/sigproc/binned_aggregate.py @@ -85,6 +85,22 @@ 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. + + 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`` @@ -132,6 +148,23 @@ 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: return hash((message.axes[self.settings.axis].gain, message.key)) diff --git a/tests/unit/test_binned_aggregate.py b/tests/unit/test_binned_aggregate.py index 0ca4c93..522dcec 100644 --- a/tests/unit/test_binned_aggregate.py +++ b/tests/unit/test_binned_aggregate.py @@ -487,3 +487,97 @@ 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_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.""" + 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]) + 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"]