BinnedAggregate: add a passthrough flag

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))
 

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"]