Follow chunk_dim for binning, and deprecate BinnedEventAggregatorSettings.axis

pyproject.toml

@@ -13,8 +13,8 @@ dependencies = [
     # left to the transitive requirement: uv only enables pre-releases for a
     # package named with a pre-release marker in *this* file.
     "ezmsg>=3.10.0b2",
-    "ezmsg-baseproc>=1.12.0",
-    "ezmsg-sigproc>=2.26.1",
+    "ezmsg-baseproc>=1.13.0",
+    "ezmsg-sigproc>=3.8.1",  # BinnedAggregate must resolve chunk_dim from axis=None
     "sparse>=0.17.0",
     "numpy>=2.2.6",
     "numba>=0.65.1",

src/ezmsg/event/binned.py

@@ -23,7 +23,12 @@
 import ezmsg.core as ez
 import sparse
 from array_api_compat import get_namespace
-from ezmsg.baseproc import BaseTransformer, BaseTransformerUnit
+from ezmsg.baseproc import (
+    BaseTransformer,
+    BaseTransformerUnit,
+    suppress_axis_deprecation,
+    warn_axis_deprecated,
+)
 from ezmsg.sigproc.aggregate import AggregationFunction
 from ezmsg.sigproc.binned_aggregate import BinnedAggregateSettings, BinnedAggregateTransformer
 from ezmsg.util.messages.axisarray import AxisArray, replace
@@ -36,8 +41,15 @@ class BinnedEventAggregatorSettings(ez.Settings):
     scale_output: bool = True
     """If True, divide each bin's count by ``bin_duration`` (events/second)."""
 
-    axis: str = "time"
-    """Name of the axis to bin along."""
+    axis: str | None = None
+    """.. deprecated:: 1.4
+        Scheduled for removal in 2.0. Binning carries an open partial bin across
+        message boundaries, which is only meaningful along the dimension messages
+        accumulate along; that dimension now comes from
+        :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`."""
+
+    def __post_init__(self) -> None:
+        warn_axis_deprecated(self, package="ezmsg-event", removal="2.0")
 
     fractional: bool = True
     """If True (default), bins span a fractional ``bin_duration * fs`` samples via
@@ -60,14 +72,17 @@ class BinnedEventAggregator(BaseTransformer[BinnedEventAggregatorSettings, AxisA
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
-        self._binner = BinnedAggregateTransformer(
-            BinnedAggregateSettings(
-                axis=self.settings.axis,
-                bin_duration=self.settings.bin_duration,
-                operation=AggregationFunction.SUM,
-                fractional=self.settings.fractional,
+        # Forwarding this stage's own already-warned setting; warning again would
+        # name a sigproc class for something the user set on ours.
+        with suppress_axis_deprecation():
+            self._binner = BinnedAggregateTransformer(
+                BinnedAggregateSettings(
+                    axis=self.settings.axis,
+                    bin_duration=self.settings.bin_duration,
+                    operation=AggregationFunction.SUM,
+                    fractional=self.settings.fractional,
+                )
             )
-        )
 
     def _process(self, message: AxisArray) -> AxisArray:
         data = message.data

src/ezmsg/event/peak.py

@@ -23,6 +23,7 @@
     BaseStatefulTransformer,
     BaseTransformerUnit,
     processor_state,
+    suppress_axis_deprecation,
 )
 from ezmsg.sigproc.scaler import AdaptiveStandardScalerTransformer
 from ezmsg.util.messages.axisarray import AxisArray, replace  # slice_along_axis,
@@ -126,9 +127,15 @@ def _reset_state(self, message: AxisArray) -> None:
         self._state.scaler = None
         self._state.data_raw = None
         if self.settings.auto_scale_tau > 0:
-            self._state.scaler = AdaptiveStandardScalerTransformer(
-                time_constant=self.settings.auto_scale_tau, axis="time"
-            )
+            # This module works on "time" throughout (see get_axis_idx calls
+            # below), so the scaler has to as well -- they share the state that
+            # normalises the samples being thresholded. Suppressed rather than
+            # dropped for that reason: letting the scaler follow chunk_dim while
+            # everything around it used "time" would be worse than either.
+            with suppress_axis_deprecation():
+                self._state.scaler = AdaptiveStandardScalerTransformer(
+                    time_constant=self.settings.auto_scale_tau, axis="time"
+                )
             if self.settings.return_peak_val:
                 self._state.data_raw = first_samp
 

tests/test_binned.py

@@ -1,4 +1,5 @@
 import time
+import warnings
 
 import numpy as np
 import pytest
@@ -220,3 +221,61 @@ def test_fractional_false_sample_locked(fs: float):
     spb = int(bin_dur * fs)
     assert sum(m.data.shape[0] for m in out) == n // spb
     assert out[0].axes["time"].gain == pytest.approx(spb / fs)
+
+
+class TestTheAxisSettingIsDeprecated:
+    """``axis`` is going away in 2.0: binning carries an open partial bin across
+    message boundaries, which is only meaningful along the dimension messages
+    accumulate along. That dimension comes from ``AxisArray.chunk_dim``."""
+
+    @staticmethod
+    def _msg(n=64, fs=200.0):
+        data = np.zeros((n, 2))
+        data[::8] = 1.0
+        return AxisArray(
+            data,
+            dims=["time", "ch"],
+            axes={"time": AxisArray.TimeAxis(fs=fs)},
+            key="dev",
+            chunk_dim="time",
+        )
+
+    @staticmethod
+    def _axis_warnings(records):
+        return [r for r in records if issubclass(r.category, FutureWarning) and "deprecated" in str(r.message)]
+
+    def test_setting_it_warns_about_our_class_not_sigprocs(self):
+        """The user set it on ``BinnedEventAggregatorSettings``; naming the
+        sigproc class we delegate to would send them looking in the wrong place."""
+        with warnings.catch_warnings(record=True) as rec:
+            warnings.simplefilter("always")
+            BinnedEventAggregatorSettings(bin_duration=0.02, axis="time")
+        (record,) = self._axis_warnings(rec)
+        assert "BinnedEventAggregatorSettings.axis" in str(record.message)
+
+    def test_leaving_it_unset_is_silent_through_processing(self):
+        """The delegation to BinnedAggregate forwards our setting, so without
+        suppression every use would warn about a class the user never touched."""
+        with warnings.catch_warnings(record=True) as rec:
+            warnings.simplefilter("always")
+            proc = BinnedEventAggregator(BinnedEventAggregatorSettings(bin_duration=0.02))
+            proc(self._msg())
+            proc(self._msg())
+        assert not self._axis_warnings(rec)
+
+    def test_forwarding_our_own_setting_warns_only_once(self):
+        with warnings.catch_warnings(record=True) as rec:
+            warnings.simplefilter("always")
+            proc = BinnedEventAggregator(BinnedEventAggregatorSettings(bin_duration=0.02, axis="time"))
+            proc(self._msg())
+        assert len(self._axis_warnings(rec)) == 1
+
+    def test_it_still_bins_along_a_configured_axis(self):
+        """Behaviour is unchanged until removal."""
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            explicit = BinnedEventAggregator(BinnedEventAggregatorSettings(bin_duration=0.02, axis="time"))
+            implicit = BinnedEventAggregator(BinnedEventAggregatorSettings(bin_duration=0.02))
+            a = explicit(self._msg())
+            b = implicit(self._msg())
+        assert np.array_equal(a.data, b.data)