diff --git a/src/ezmsg/sigproc/adaptive_lattice_notch.py b/src/ezmsg/sigproc/adaptive_lattice_notch.py index 6814221..1fe37db 100644 --- a/src/ezmsg/sigproc/adaptive_lattice_notch.py +++ b/src/ezmsg/sigproc/adaptive_lattice_notch.py @@ -8,6 +8,9 @@ from ezmsg.util.messages.axisarray import AxisArray, CoordinateAxis from ezmsg.util.messages.util import replace +from .util.deprecation import warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim + class AdaptiveLatticeNotchFilterSettings(ez.Settings): """Settings for the Adaptive Lattice Notch Filter.""" @@ -18,7 +21,15 @@ class AdaptiveLatticeNotchFilterSettings(ez.Settings): """Smoothing factor""" eta: float = 0.99 """Forgetting factor""" - axis: str = "time" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) + """Axis to apply filter to""" init_notch_freq: float | None = None """Initial notch frequency. Should be < nyquist.""" @@ -28,6 +39,9 @@ class AdaptiveLatticeNotchFilterSettings(ez.Settings): @processor_state class AdaptiveLatticeNotchFilterState: + axis: str = "" + """The resolved chunk dimension, fixed at reset so every later use agrees.""" + """State for the Adaptive Lattice Notch Filter.""" s_history: npt.NDArray | None = None @@ -71,10 +85,11 @@ class AdaptiveLatticeNotchFilterTransformer( NONRESET_SETTINGS_FIELDS = frozenset({"gamma", "mu", "eta", "chunkwise"}) def _reset_state(self, message: AxisArray) -> None: - ax_idx = message.get_axis_idx(self.settings.axis) + axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") + ax_idx = message.get_axis_idx(axis) sample_shape = message.data.shape[:ax_idx] + message.data.shape[ax_idx + 1 :] - fs = 1 / message.axes[self.settings.axis].gain + fs = 1 / message.axes[axis].gain init_f = ( self.settings.init_notch_freq if self.settings.init_notch_freq is not None else 0.07178314656435313 * fs ) @@ -83,13 +98,15 @@ def _reset_state(self, message: AxisArray) -> None: """Reset filter state to initial values.""" self._state = AdaptiveLatticeNotchFilterState() + # Set after the wholesale replacement above, which would otherwise drop it. + self._state.axis = axis self._state.s_history = np.zeros((2,) + sample_shape, dtype=float) self._state.p = np.zeros(sample_shape, dtype=float) self._state.q = np.zeros(sample_shape, dtype=float) self._state.k1 = init_k1 + np.zeros(sample_shape, dtype=float) self._state.freq_template = CoordinateAxis( data=np.zeros((0,) + sample_shape, dtype=float), - dims=[self.settings.axis] + message.dims[:ax_idx] + message.dims[ax_idx + 1 :], + dims=[axis] + message.dims[:ax_idx] + message.dims[ax_idx + 1 :], unit="Hz", ) @@ -105,17 +122,18 @@ def _reset_state(self, message: AxisArray) -> None: def _process(self, message: AxisArray) -> AxisArray: x_data = message.data - ax_idx = message.get_axis_idx(self.settings.axis) + axis = self._state.axis + ax_idx = message.get_axis_idx(axis) # TODO: Time should be moved to -1th axis, not the 0th axis - if message.dims[0] != self.settings.axis: + if message.dims[0] != axis: x_data = np.moveaxis(x_data, ax_idx, 0) # Access settings once gamma = self.settings.gamma eta = self.settings.eta mu = self.settings.mu - fs = 1 / message.axes[self.settings.axis].gain + fs = 1 / message.axes[axis].gain # Pre-compute constants one_minus_eta = 1 - eta diff --git a/src/ezmsg/sigproc/adaptive_lnc.py b/src/ezmsg/sigproc/adaptive_lnc.py index cc56abc..8628ccf 100644 --- a/src/ezmsg/sigproc/adaptive_lnc.py +++ b/src/ezmsg/sigproc/adaptive_lnc.py @@ -73,6 +73,9 @@ from ezmsg.util.messages.axisarray import AxisArray from ezmsg.util.messages.util import replace +from .util.deprecation import warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim + # Optional Apple-Silicon GPU backend. The canceller is an LTI SOS notch # cascade (see `design_lnc_sos`), so on MLX arrays we dispatch to the Metal # `sosfilt` kernel; everything else runs through scipy on the array's own @@ -165,14 +168,23 @@ class AdaptiveLNCSettings(ez.Settings): unchanged. (Per-channel sampling-delay alignment is handled separately, upstream, by ``SamplingDelayAlignmentTransformer``.)""" - axis: str = "time" - """Name of the axis to filter along.""" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) @processor_state class AdaptiveLNCState: """State for :class:`AdaptiveLNCTransformer`.""" + axis: str = "" + """The resolved chunk dimension, fixed at reset so every later use agrees.""" + omega: float = 0.0 """Current NCO angular frequency in rad/sample (tracked by the FLL).""" @@ -288,11 +300,12 @@ class AdaptiveLNCTransformer( NONRESET_SETTINGS_FIELDS = frozenset({"adapt_time_constant", "freq_time_constant"}) def _reset_state(self, message: AxisArray) -> None: - ax_idx = message.get_axis_idx(self.settings.axis) + self._state.axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") + ax_idx = message.get_axis_idx(self._state.axis) sample_shape = message.data.shape[:ax_idx] + message.data.shape[ax_idx + 1 :] xp, is_mlx = _namespace(message.data) - fs = 1.0 / message.axes[self.settings.axis].gain + fs = 1.0 / message.axes[self._state.axis].gain # Seed the NCO at the nominal normalised frequency; the FLL refines it. self._state.omega = 2.0 * np.pi * self.settings.line_freq / fs max_deviation = self.settings.max_freq_deviation @@ -472,7 +485,7 @@ def _process(self, message: AxisArray) -> AxisArray: # No cancellation and no frequency tracking; emit the input as-is. return message - ax_idx = message.get_axis_idx(self.settings.axis) + ax_idx = message.get_axis_idx(self._state.axis) x_data = message.data xp, is_mlx = _namespace(x_data) moved = ax_idx != 0 @@ -482,7 +495,7 @@ def _process(self, message: AxisArray) -> AxisArray: n = x_data.shape[0] st = self._state dtype = x_data.dtype - fs = 1.0 / message.axes[self.settings.axis].gain + fs = 1.0 / message.axes[self._state.axis].gain # Time constants -> gains (independent of chunk size and fs). # mu = 2 / (tau_adapt * fs); beta = 1 - exp(-window_dt / tau_freq). diff --git a/src/ezmsg/sigproc/affinetransform.py b/src/ezmsg/sigproc/affinetransform.py index 38da4d5..5f4638b 100644 --- a/src/ezmsg/sigproc/affinetransform.py +++ b/src/ezmsg/sigproc/affinetransform.py @@ -36,7 +36,7 @@ from ezmsg.sigproc.util.array import array_device, is_float_dtype, xp_asarray, xp_copy, xp_create, xp_empty from ezmsg.sigproc.util.blockdiag import plan_block_matmul from ezmsg.sigproc.util.channels import ChannelGroupSpec, resolve_channel_groups -from ezmsg.sigproc.util.message import with_fingerprint +from ezmsg.sigproc.util.message import resolve_feature_dim, with_fingerprint from ezmsg.sigproc.util.rereference import RereferenceKind, rereference_matrix KERNELS = ("auto", "dense", "blocks") @@ -242,7 +242,7 @@ def _reset_state(self, message: AxisArray) -> None: if self.settings.kernel not in KERNELS: raise ValueError(f"kernel must be one of {KERNELS}, got {self.settings.kernel!r}") - axis = self.settings.axis or message.dims[-1] + axis = self.settings.axis or resolve_feature_dim(message) axis_idx = message.get_axis_idx(axis) n_in = message.data.shape[axis_idx] xp = get_namespace(message.data) @@ -447,7 +447,7 @@ def _stacked_split(self, xp): def _process(self, message: AxisArray) -> AxisArray: xp = get_namespace(message.data) - axis = self.settings.axis or message.dims[-1] + axis = self.settings.axis or resolve_feature_dim(message) axis_idx = message.get_axis_idx(axis) data = message.data @@ -588,7 +588,7 @@ class CommonRereferenceTransformer( def _reset_state(self, message: AxisArray) -> None: xp = get_namespace(message.data) dev = array_device(message.data) - axis = self.settings.axis or message.dims[-1] + axis = self.settings.axis or resolve_feature_dim(message) axis_idx = message.get_axis_idx(axis) n_ch = message.data.shape[axis_idx] include_current = self.settings.include_current @@ -638,7 +638,7 @@ def _process(self, message: AxisArray) -> AxisArray: return message xp = get_namespace(message.data) - axis = self.settings.axis or message.dims[-1] + axis = self.settings.axis or resolve_feature_dim(message) axis_idx = message.get_axis_idx(axis) state = self._state data = message.data diff --git a/src/ezmsg/sigproc/aggregate.py b/src/ezmsg/sigproc/aggregate.py index 997a00c..e4f4556 100644 --- a/src/ezmsg/sigproc/aggregate.py +++ b/src/ezmsg/sigproc/aggregate.py @@ -29,7 +29,7 @@ ) from .spectral import OptionsEnum -from .util.message import with_fingerprint +from .util.message import resolve_feature_dim, with_fingerprint class AggregationFunction(OptionsEnum): @@ -260,7 +260,7 @@ async def __acall__(self, message: AxisArray) -> AxisArray: return await super().__acall__(message) def _reset_state(self, message: AxisArray) -> None: - axis = self.settings.axis or message.dims[0] + axis = self.settings.axis or resolve_feature_dim(message, 0) target_axis = message.get_axis(axis) ax_idx = message.get_axis_idx(axis) @@ -293,7 +293,7 @@ def _reset_state(self, message: AxisArray) -> None: ) def _process(self, message: AxisArray) -> AxisArray: - axis = self.settings.axis or message.dims[0] + axis = self.settings.axis or resolve_feature_dim(message, 0) ax_idx = message.get_axis_idx(axis) # The bands are already resolved to slices in _reset_state, and ax_vec diff --git a/src/ezmsg/sigproc/align.py b/src/ezmsg/sigproc/align.py index e41e7c9..559c4fc 100644 --- a/src/ezmsg/sigproc/align.py +++ b/src/ezmsg/sigproc/align.py @@ -12,11 +12,19 @@ from ezmsg.util.messages.axisarray import AxisArray from .util.axisarray_buffer import HybridAxisArrayBuffer +from .util.deprecation import warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim class AlignAlongAxisSettings(ez.Settings): - axis: str = "time" - """Axis used for alignment (typically the time axis).""" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) buffer_dur: float = 10.0 """Buffer duration in seconds for each input stream.""" @@ -79,7 +87,7 @@ def _hash_message(self, message: AxisArray) -> int: return hash(self._extract_gain(message)) def _extract_gain(self, message: AxisArray) -> float | None: - align_name = self.settings.axis or message.dims[0] + align_name = resolve_configured_chunk_dim(self, message, self.settings.axis) ax = message.axes.get(align_name) if ax is not None and hasattr(ax, "gain"): return ax.gain @@ -131,7 +139,7 @@ def _request_reset(self) -> None: super()._request_reset() def _reset_state(self, message: AxisArray) -> None: - align_axis = self.settings.axis or message.dims[0] + align_axis = resolve_configured_chunk_dim(self, message, self.settings.axis) if self._hash == -1 and not getattr(self, "_force_full_reset", False): self._state.align_axis = align_axis if self._state.buf_a is None: @@ -159,7 +167,7 @@ def _process(self, message: AxisArray) -> _AlignPair | None: def push_b(self, message: AxisArray) -> _AlignPair | None: """Process input B: check gain, detect shape changes, buffer, try align.""" - align_axis = self.settings.axis or message.dims[0] + align_axis = resolve_configured_chunk_dim(self, message, self.settings.axis) # Gain compatibility check. Skipped when B's gain can't be estimated # (e.g. a single-sample CoordinateAxis yields None) — there is nothing diff --git a/src/ezmsg/sigproc/binned_aggregate.py b/src/ezmsg/sigproc/binned_aggregate.py index baeddca..d695925 100644 --- a/src/ezmsg/sigproc/binned_aggregate.py +++ b/src/ezmsg/sigproc/binned_aggregate.py @@ -56,14 +56,21 @@ from .aggregate import AggregationFunction, aggregate_slices, needs_coordinates from .util.array import xp_copy from .util.binning import BinSchedule, BinStep -from .util.message import is_empty_along, with_fingerprint +from .util.deprecation import warn_axis_deprecated +from .util.message import is_empty_along, resolve_configured_chunk_dim, with_fingerprint class BinnedAggregateSettings(ez.Settings): """Settings for :obj:`BinnedAggregate`.""" - axis: str = "time" - """The name of the axis to bin and aggregate along.""" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) bin_duration: float = 0.02 """Output bin duration in seconds.""" @@ -113,6 +120,9 @@ class BinnedAggregateSettings(ez.Settings): @processor_state class BinnedAggregateState: + axis: str = "" + """The resolved chunk dimension, fixed at reset so every later use agrees.""" + schedule: BinSchedule | None = None """Shared bin-boundary schedule (see :obj:`ezmsg.sigproc.util.binning`). Owns the sample rate, samples-per-bin, output gain, global bin index, and carried @@ -167,7 +177,8 @@ async def __acall__(self, message: AxisArray) -> AxisArray: return await super().__acall__(message) def _reset_state(self, message: AxisArray) -> None: - axis_info = message.get_axis(self.settings.axis) + self._state.axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") + axis_info = message.get_axis(self._state.axis) schedule = BinSchedule( bin_duration=self.settings.bin_duration, fractional=self.settings.fractional, @@ -227,10 +238,10 @@ def _out_dims(self, message: AxisArray) -> list[str]: return dims + [self.settings.newaxis] if self._multi else dims def _out_axes(self, message: AxisArray, step: BinStep) -> dict: - axis_info = message.get_axis(self.settings.axis) + axis_info = message.get_axis(self._state.axis) axes = { **message.axes, - self.settings.axis: replace(axis_info, gain=step.output_gain, offset=step.output_offset), + self._state.axis: replace(axis_info, gain=step.output_gain, offset=step.output_offset), } if self._multi: axes[self.settings.newaxis] = self._state.metric_axis @@ -252,7 +263,7 @@ def _empty_like(self, message: AxisArray, axis_idx: int, step: BinStep) -> AxisA ) def _process(self, message: AxisArray) -> AxisArray: - axis = self.settings.axis + axis = self._state.axis axis_info = message.get_axis(axis) axis_idx = message.get_axis_idx(axis) xp = get_namespace(message.data) @@ -321,5 +332,5 @@ async def on_signal(self, message: AxisArray) -> typing.AsyncGenerator: cadence. """ result = await self.processor.__acall__(message) - if result is not None and not is_empty_along(result, (self.SETTINGS.axis,)): + if result is not None and not is_empty_along(result, (self.processor.state.axis,)): yield self.OUTPUT_SIGNAL, result diff --git a/src/ezmsg/sigproc/butterworthzerophase.py b/src/ezmsg/sigproc/butterworthzerophase.py index 38f2ab4..687cc22 100644 --- a/src/ezmsg/sigproc/butterworthzerophase.py +++ b/src/ezmsg/sigproc/butterworthzerophase.py @@ -32,6 +32,7 @@ _sosfilt_mlx_metal_xp, ) from .util.array import xp_asarray, xp_copy, xp_empty, xp_flip +from .util.message import resolve_configured_chunk_dim if _HAS_MLX_METAL: import mlx.core as _mx @@ -188,7 +189,7 @@ def _reset_state(self, message: AxisArray) -> None: self._tail = None self._tail_offset = 0.0 # Compute pad_length based on the message's sampling rate - axis = message.dims[0] if self.settings.axis is None else self.settings.axis + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) fs = 1 / message.axes[axis].gain self._pad_length = self._compute_pad_length(fs) self.state.needs_redesign = True @@ -230,7 +231,7 @@ def _initialize_zi(self, data, ax_idx: int, xp): return self._zi_tiled * first_sample def _process(self, message: AxisArray) -> AxisArray: - axis = message.dims[0] if self.settings.axis is None else self.settings.axis + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) ax_idx = message.get_axis_idx(axis) fs = 1 / message.axes[axis].gain diff --git a/src/ezmsg/sigproc/coordinatespaces.py b/src/ezmsg/sigproc/coordinatespaces.py index 57597de..303ef01 100644 --- a/src/ezmsg/sigproc/coordinatespaces.py +++ b/src/ezmsg/sigproc/coordinatespaces.py @@ -22,7 +22,7 @@ ) from ezmsg.util.messages.axisarray import AxisArray, replace -from .util.message import with_fingerprint +from .util.message import resolve_feature_dim, with_fingerprint # -- Utility functions for coordinate transformations -- @@ -109,7 +109,7 @@ class CoordinateSpacesTransformer(BaseTransformer[CoordinateSpacesSettings, Axis def _process(self, message: AxisArray) -> AxisArray: xp = get_namespace(message.data) - axis = self.settings.axis or message.dims[-1] + axis = self.settings.axis or resolve_feature_dim(message) axis_idx = message.get_axis_idx(axis) if message.data.shape[axis_idx] != 2: diff --git a/src/ezmsg/sigproc/decimate.py b/src/ezmsg/sigproc/decimate.py index d89d803..bc35d00 100644 --- a/src/ezmsg/sigproc/decimate.py +++ b/src/ezmsg/sigproc/decimate.py @@ -9,6 +9,7 @@ from .cheby import ChebyshevFilterSettings, ChebyshevFilterTransformer from .downsample import Downsample, DownsampleSettings from .filter import BACoeffs, SOSCoeffs +from .util.deprecation import suppress_axis_deprecation, warn_axis_deprecated class ChebyForDecimateTransformer(ChebyshevFilterTransformer[BACoeffs | SOSCoeffs]): @@ -47,8 +48,14 @@ class DecimateSettings(DownsampleSettings): decimating another is not decimation. """ - axis: str = "time" - """Axis for the anti-aliasing lowpass filter.""" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) class Decimate(ez.Collection): @@ -66,14 +73,17 @@ class Decimate(ez.Collection): DOWNSAMPLE = Downsample() def configure(self) -> None: - cheby_settings = ChebyshevFilterSettings( - order=8, - ripple_tol=0.05, - Wn=0.4 * self.SETTINGS.target_rate, - btype="lowpass", - axis=self.SETTINGS.axis, - wn_hz=True, - ) + # Already warned about on DecimateSettings, whose `axis` exists only to + # reach this filter. + with suppress_axis_deprecation(): + cheby_settings = ChebyshevFilterSettings( + order=8, + ripple_tol=0.05, + Wn=0.4 * self.SETTINGS.target_rate, + btype="lowpass", + axis=self.SETTINGS.axis, + wn_hz=True, + ) self.FILTER.apply_settings(cheby_settings) # `axis` is the filter's, not the downsampler's -- pass only what # DownsampleSettings still declares. diff --git a/src/ezmsg/sigproc/diff.py b/src/ezmsg/sigproc/diff.py index 609c0d1..d8f2da2 100644 --- a/src/ezmsg/sigproc/diff.py +++ b/src/ezmsg/sigproc/diff.py @@ -19,10 +19,20 @@ from ezmsg.util.messages.util import replace from .util.array import xp_copy +from .util.deprecation import warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim class DiffSettings(ez.Settings): axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) + scale_by_fs: bool = False @@ -33,26 +43,30 @@ class DiffState: class DiffTransformer(BaseStatefulTransformer[DiffSettings, AxisArray, AxisArray, DiffState]): + def _axis(self, message: AxisArray) -> str: + return resolve_configured_chunk_dim(self, message, self.settings.axis) + def __call__(self, message: AxisArray) -> AxisArray: - ax_idx = message.get_axis_idx(self.settings.axis) + ax_idx = message.get_axis_idx(self._axis(message)) if message.data.shape[ax_idx] == 0: return message return super().__call__(message) async def __acall__(self, message: AxisArray) -> AxisArray: - ax_idx = message.get_axis_idx(self.settings.axis) + ax_idx = message.get_axis_idx(self._axis(message)) if message.data.shape[ax_idx] == 0: return message return await super().__acall__(message) def _reset_state(self, message) -> None: - ax_idx = message.get_axis_idx(self.settings.axis) + axis = self._axis(message) + ax_idx = message.get_axis_idx(axis) # Copied for the same reason as in `_process`: state must never alias the # message's (possibly shared-memory-backed) buffer, even though this one # happens to be overwritten before the call returns. self.state.last_dat = xp_copy(slice_along_axis(message.data, slice(0, 1), axis=ax_idx)) if self.settings.scale_by_fs: - ax_info = message.get_axis(self.settings.axis) + ax_info = message.get_axis(axis) if hasattr(ax_info, "data"): if len(ax_info.data) > 1: self.state.last_time = 2 * ax_info.data[0] - ax_info.data[1] @@ -61,7 +75,7 @@ def _reset_state(self, message) -> None: def _process(self, message: AxisArray) -> AxisArray: xp = get_namespace(message.data) - axis = self.settings.axis or message.dims[0] + axis = self._axis(message) ax_idx = message.get_axis_idx(axis) diffs = xp.diff( @@ -90,5 +104,5 @@ class DiffUnit(BaseTransformerUnit[DiffSettings, AxisArray, AxisArray, DiffTrans SETTINGS = DiffSettings -def diff(axis: str = "time", scale_by_fs: bool = False) -> DiffTransformer: +def diff(axis: str | None = None, scale_by_fs: bool = False) -> DiffTransformer: return DiffTransformer(DiffSettings(axis=axis, scale_by_fs=scale_by_fs)) diff --git a/src/ezmsg/sigproc/downsample.py b/src/ezmsg/sigproc/downsample.py index 626b767..6148250 100644 --- a/src/ezmsg/sigproc/downsample.py +++ b/src/ezmsg/sigproc/downsample.py @@ -14,7 +14,7 @@ slice_along_axis, ) -from .util.message import is_empty_along +from .util.message import is_empty_along, resolve_chunk_dim class DownsampleSettings(ez.Settings): @@ -72,12 +72,7 @@ class DownsampleTransformer(BaseStatefulTransformer[DownsampleSettings, AxisArra def _resolve_axis(self, message: AxisArray) -> str: """The dimension messages accumulate along, which is the only one to downsample. Falls back to :attr:`STREAMING_DIMS` when undeclared.""" - if message.chunk_dim is not None: - return message.chunk_dim - for name in self.STREAMING_DIMS: - if name in message.dims: - return name - return message.dims[0] + return resolve_chunk_dim(message, self.STREAMING_DIMS) def _hash_message(self, message: AxisArray) -> int: # The whole state is a decimation factor and the phase counter that walks diff --git a/src/ezmsg/sigproc/ewma.py b/src/ezmsg/sigproc/ewma.py index f33e1ce..6302a91 100644 --- a/src/ezmsg/sigproc/ewma.py +++ b/src/ezmsg/sigproc/ewma.py @@ -15,6 +15,9 @@ from ezmsg.sigproc.util.array import np_finfo +from .util.deprecation import warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim + def _ewma_mlx_metal_xp(data, axis_idx: int, zi, alpha: float, chunk_sizes: tuple[int, ...]): """Run EWMA through the MLX Metal helper while preserving scipy zi layout.""" @@ -165,6 +168,13 @@ class EWMASettings(ez.Settings): """The amount of time for the smoothed response of a unit step function to reach 1 - 1/e approx-eq 63.2%.""" axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) accumulate: bool = True """If True, update the EWMA state with each sample. If False, only apply @@ -274,7 +284,7 @@ async def __acall__(self, message: AxisArray) -> AxisArray: return await super().__acall__(message) def _reset_state(self, message: AxisArray) -> None: - axis = self.settings.axis or message.dims[0] + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) axis_idx = message.get_axis_idx(axis) self._state.alpha = _alpha_from_tau(self.settings.time_constant, message.axes[axis].gain) # Start from zero; _process divides out the missing-history bias. @@ -330,7 +340,7 @@ def _lfilter_axis_last( ) def _process(self, message: AxisArray) -> AxisArray: - axis = self.settings.axis or message.dims[0] + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) axis_idx = message.get_axis_idx(axis) xp = np if is_numpy_array(message.data) else get_namespace(message.data) diff --git a/src/ezmsg/sigproc/ewmfilter.py b/src/ezmsg/sigproc/ewmfilter.py index 4ae8189..72a59a9 100644 --- a/src/ezmsg/sigproc/ewmfilter.py +++ b/src/ezmsg/sigproc/ewmfilter.py @@ -8,6 +8,8 @@ from ezmsg.util.messages.axisarray import AxisArray from ezmsg.util.messages.util import replace +from .util.deprecation import suppress_axis_deprecation +from .util.message import resolve_chunk_dim from .window import Window, WindowSettings @@ -62,7 +64,7 @@ async def sync_output(self) -> typing.AsyncGenerator: axis_name = self.SETTINGS.axis if axis_name is None: - axis_name = signal.dims[0] + axis_name = resolve_chunk_dim(signal) axis_idx = signal.get_axis_idx(axis_name) @@ -131,20 +133,23 @@ class EWMFilter(ez.Collection): EWM = EWM() def configure(self) -> None: - self.EWM.apply_settings( - EWMSettings( - axis=self.SETTINGS.axis, - zero_offset=self.SETTINGS.zero_offset, + # Already warned about on EWMFilterSettings; fanning it out to the two + # children is one setting, not three. + with suppress_axis_deprecation(): + self.EWM.apply_settings( + EWMSettings( + axis=self.SETTINGS.axis, + zero_offset=self.SETTINGS.zero_offset, + ) ) - ) - self.WINDOW.apply_settings( - WindowSettings( - axis=self.SETTINGS.axis, - window_dur=self.SETTINGS.history_dur, - window_shift=None, # 1:1 mode + self.WINDOW.apply_settings( + WindowSettings( + axis=self.SETTINGS.axis, + window_dur=self.SETTINGS.history_dur, + window_shift=None, # 1:1 mode + ) ) - ) def network(self) -> ez.NetworkDefinition: return ( diff --git a/src/ezmsg/sigproc/fbcca.py b/src/ezmsg/sigproc/fbcca.py index 05f8667..eb74279 100644 --- a/src/ezmsg/sigproc/fbcca.py +++ b/src/ezmsg/sigproc/fbcca.py @@ -22,6 +22,7 @@ ) from .kaiser import KaiserFilterSettings from .sampler import SampleTriggerMessage +from .util.deprecation import suppress_axis_deprecation from .util.message import with_fingerprint from .window import WindowSettings, WindowTransformer @@ -239,32 +240,36 @@ def _initialize_processors( ) -> dict[str, BaseProcessor | BaseStatefulProcessor]: pipeline = {} - if settings.filterbank_dim is not None: - cut_freqs = (np.arange(settings.subbands + 1) * settings.filter_bw) + settings.filter_low - filters = [ - KaiserFilterSettings( - axis=settings.time_dim, - cutoff=(c - settings.trans_bw, cut_freqs[-1]), - ripple=settings.ripple_db, - width=settings.trans_bw, - pass_zero=False, - ) - for c in cut_freqs[:-1] - ] + # Every child below is configured from FBCCA's own `time_dim`, which is + # not deprecated; warning about the children's `axis` would name settings + # the user never touched and cannot drop. + with suppress_axis_deprecation(): + if settings.filterbank_dim is not None: + cut_freqs = (np.arange(settings.subbands + 1) * settings.filter_bw) + settings.filter_low + filters = [ + KaiserFilterSettings( + axis=settings.time_dim, + cutoff=(c - settings.trans_bw, cut_freqs[-1]), + ripple=settings.ripple_db, + width=settings.trans_bw, + pass_zero=False, + ) + for c in cut_freqs[:-1] + ] - pipeline["filterbank"] = FilterbankDesignTransformer( - FilterbankDesignSettings(filters=filters, new_axis=settings.filterbank_dim) - ) + pipeline["filterbank"] = FilterbankDesignTransformer( + FilterbankDesignSettings(filters=filters, new_axis=settings.filterbank_dim) + ) - pipeline["window"] = WindowTransformer( - WindowSettings( - axis=settings.time_dim, - newaxis=settings.window_dim, - window_dur=settings.window_dur, - window_shift=settings.window_shift, - zero_pad_until="shift", + pipeline["window"] = WindowTransformer( + WindowSettings( + axis=settings.time_dim, + newaxis=settings.window_dim, + window_dur=settings.window_dur, + window_shift=settings.window_shift, + zero_pad_until="shift", + ) ) - ) pipeline["fbcca"] = FBCCATransformer(settings) diff --git a/src/ezmsg/sigproc/filter.py b/src/ezmsg/sigproc/filter.py index e0c5d20..2e6188d 100644 --- a/src/ezmsg/sigproc/filter.py +++ b/src/ezmsg/sigproc/filter.py @@ -24,6 +24,8 @@ from .util import sosfilt_direct from .util.array import array_device, xp_asarray, xp_create +from .util.deprecation import suppress_axis_deprecation, warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim from .util.threaded_filt import DEFAULT_MIN_BYTES as _DEFAULT_THREAD_MIN_BYTES from .util.threaded_filt import filt_threaded, should_thread @@ -303,7 +305,13 @@ def _fir_filt_conv(b_1d, data, zi, axis_idx, xp): class FilterBaseSettings(ez.Settings): axis: str | None = None - """The name of the axis to operate on.""" + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) coef_type: str = "ba" """The type of filter coefficients. One of "ba" or "sos".""" @@ -438,7 +446,7 @@ def _reset_state(self, message: AxisArray) -> None: # edge-scaled by the first sample x0 -- the scipy ``lfilter_zi * x[0]`` # idiom -- treating the pre-stream signal as constant x0 so that a DC # offset does not ring through as a start-up transient. - axis = message.dims[0] if self.settings.axis is None else self.settings.axis + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) axis_idx = message.get_axis_idx(axis) n_tail = message.data.ndim - axis_idx - 1 _, coefs = _normalize_coefs(self.settings.coefs) @@ -650,7 +658,7 @@ def _sos_direct_for(self, data_np: npt.NDArray, zi_np: npt.NDArray): def _process(self, message: AxisArray) -> AxisArray: if message.data.size > 0: - axis = message.dims[0] if self.settings.axis is None else self.settings.axis + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) axis_idx = message.get_axis_idx(axis) if self.state.fir_method is not None and self.state.fir_b_1d is None: self._refresh_fir_taps(message, axis_idx) @@ -838,7 +846,7 @@ def __call__(self, message: AxisArray) -> AxisArray: def _reset_state(self, message: AxisArray) -> None: design_fun = self.get_design_function() - axis = message.dims[0] if self.settings.axis is None else self.settings.axis + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) fs = 1 / message.axes[axis].gain coefs = design_fun(fs) @@ -849,14 +857,20 @@ def _reset_state(self, message: AxisArray) -> None: b, a = coefs coefs = scipy.signal.tf2sos(b, a) - new_settings = FilterSettings( - axis=axis, - coef_type=self.settings.coef_type, - coefs=coefs, - use_mlx_metal=self.settings.use_mlx_metal, - mlx_metal_chunk_sizes=self.settings.mlx_metal_chunk_sizes, - ) - self.state.filter = FilterTransformer(settings=new_settings) + # The child is handed the axis this transformer already resolved, so it + # filters the same dimension. Suppressed because this runs on every reset + # -- mid-stream, where the warning would name the pipeline driver -- and + # because the value forwarded is ours, not necessarily anything the user + # set. + with suppress_axis_deprecation(): + new_settings = FilterSettings( + axis=axis, + coef_type=self.settings.coef_type, + coefs=coefs, + use_mlx_metal=self.settings.use_mlx_metal, + mlx_metal_chunk_sizes=self.settings.mlx_metal_chunk_sizes, + ) + self.state.filter = FilterTransformer(settings=new_settings) self.state.needs_redesign = False def _process(self, message: AxisArray) -> AxisArray: diff --git a/src/ezmsg/sigproc/filterbank.py b/src/ezmsg/sigproc/filterbank.py index 5bf439e..c0aaf43 100644 --- a/src/ezmsg/sigproc/filterbank.py +++ b/src/ezmsg/sigproc/filterbank.py @@ -20,6 +20,8 @@ from scipy.special import lambertw from .spectrum import OptionsEnum +from .util.deprecation import warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim from .window import WindowTransformer @@ -65,8 +67,14 @@ class FilterbankSettings(ez.Settings): See `scipy.signal.minimum_phase` for details. """ - axis: str = "time" - """The name of the axis to operate on. This should usually be "time".""" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) new_axis: str = "kernel" """The name of the new axis corresponding to the kernel index.""" @@ -98,7 +106,7 @@ def _hash_message(self, message: AxisArray) -> int: return self._message_hash(message, extra=(message.data.dtype.kind,)) def _reset_state(self, message: AxisArray) -> None: - axis = self.settings.axis or message.dims[0] + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) gain = message.axes[axis].gain if axis in message.axes else 1.0 targ_ax_ix = message.get_axis_idx(axis) in_shape = message.data.shape[:targ_ax_ix] + message.data.shape[targ_ax_ix + 1 :] @@ -203,7 +211,7 @@ def _reset_state(self, message: AxisArray) -> None: # TODO: If fft_kernels have significant stretches of zeros, convert to sparse array. def _process(self, message: AxisArray) -> AxisArray: - axis = self.settings.axis or message.dims[0] + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) targ_ax_ix = message.get_axis_idx(axis) # Make sure target axis is in -1th position. @@ -275,7 +283,7 @@ def filterbank( kernels: list[npt.NDArray] | tuple[npt.NDArray, ...], mode: FilterbankMode = FilterbankMode.CONV, min_phase: MinPhaseMode = MinPhaseMode.NONE, - axis: str = "time", + axis: str | None = None, new_axis: str = "kernel", ) -> FilterbankTransformer: """ diff --git a/src/ezmsg/sigproc/filterbankdesign.py b/src/ezmsg/sigproc/filterbankdesign.py index 8aa59be..5c7df21 100644 --- a/src/ezmsg/sigproc/filterbankdesign.py +++ b/src/ezmsg/sigproc/filterbankdesign.py @@ -19,6 +19,8 @@ MinPhaseMode, ) from .kaiser import KaiserFilterSettings, kaiser_design_fun +from .util.deprecation import suppress_axis_deprecation, warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim class FilterbankDesignSettings(ez.Settings): @@ -40,8 +42,14 @@ class FilterbankDesignSettings(ez.Settings): See `scipy.signal.minimum_phase` for details. """ - axis: str = "time" - """The name of the axis to operate on. This should usually be "time".""" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) new_axis: str = "kernel" """The name of the new axis corresponding to the kernel index.""" @@ -49,6 +57,9 @@ class FilterbankDesignSettings(ez.Settings): @processor_state class FilterbankDesignState: + axis: str = "" + """The resolved chunk dimension, fixed at reset so every later use agrees.""" + filterbank: FilterbankTransformer | None = None needs_redesign: bool = False @@ -111,21 +122,29 @@ def _hash_message(self, message: AxisArray) -> int: # of the sample rate. That inner transformer keeps its own hash and # rebuilds itself when the channels change, so folding the channel # fingerprint in here would only redesign kernels that came out the same. - return hash((message.key, getattr(message.axes.get(self.settings.axis), "gain", None))) + # Runs before `_reset_state`, so the axis is resolved from the message + # rather than read back off state that does not exist yet. + axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") + return hash((message.key, getattr(message.axes.get(axis), "gain", None))) def _reset_state(self, message: AxisArray) -> None: - axis_obj = message.axes[self.settings.axis] + self.state.axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") + axis_obj = message.axes[self.state.axis] assert isinstance(axis_obj, AxisArray.LinearAxis) fs = 1 / axis_obj.gain kernels = self._calculate_kernels(fs) - new_settings = FilterbankSettings( - kernels=kernels, - mode=self.settings.mode, - min_phase=self.settings.min_phase, - axis=self.settings.axis, - new_axis=self.settings.new_axis, - ) - self.state.filterbank = FilterbankTransformer(settings=new_settings) + # Forwards this stage's own `axis`, which is not the deprecated setting; + # warning here would name FilterbankSettings for something the user set on + # FilterbankDesignSettings, and would do it on every reset. + with suppress_axis_deprecation(): + new_settings = FilterbankSettings( + kernels=kernels, + mode=self.settings.mode, + min_phase=self.settings.min_phase, + axis=self.state.axis, + new_axis=self.settings.new_axis, + ) + self.state.filterbank = FilterbankTransformer(settings=new_settings) def _process(self, message: AxisArray) -> AxisArray: return self.state.filterbank(message) diff --git a/src/ezmsg/sigproc/fir_hilbert.py b/src/ezmsg/sigproc/fir_hilbert.py index c095650..3c67777 100644 --- a/src/ezmsg/sigproc/fir_hilbert.py +++ b/src/ezmsg/sigproc/fir_hilbert.py @@ -18,6 +18,8 @@ FilterByDesignTransformer, ) +from .util.message import resolve_configured_chunk_dim + class FIRHilbertFilterSettings(FilterBaseSettings): """Settings for :obj:`FIRHilbertFilter`.""" @@ -266,7 +268,7 @@ def _process(self, message: AxisArray) -> AxisArray: y_imag_msg = self._state.filter(message) y_imag = y_imag_msg.data - axis_name = self.settings.axis or message.dims[0] + axis_name = resolve_configured_chunk_dim(self, message, self.settings.axis) axis_idx = message.get_axis_idx(axis_name) if self._state.dly is None: taps = self._state.filter.get_taps() diff --git a/src/ezmsg/sigproc/flatten.py b/src/ezmsg/sigproc/flatten.py index 5eac834..1d0c691 100644 --- a/src/ezmsg/sigproc/flatten.py +++ b/src/ezmsg/sigproc/flatten.py @@ -42,7 +42,7 @@ ) from ezmsg.util.messages.axisarray import AxisArray, CoordinateAxis, replace -from .util.message import with_fingerprint +from .util.message import resolve_chunk_dim, with_fingerprint def normalize_axis_label(label): @@ -80,8 +80,13 @@ class FlattenSettings(ez.Settings): """ preserve_axis: str | None = None - """Axis kept as the leading dim of the output (typically - ``"time"``). Defaults to ``message.dims[0]``.""" + """Axis kept as the leading dim of the output (typically ``"time"``). + + Defaults to the dimension messages accumulate along + (:attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`). Unlike the + state-carrying stages, this one stays configurable: Flatten holds no data + between messages, and folding the chunk dimension into the merged axis is a + coherent request -- the output simply declares no chunk dimension.""" sample_axis: str | None = None """Optional rename for ``preserve_axis`` on the output @@ -243,7 +248,7 @@ def _expand(arr: np.ndarray, axis_idx: int) -> np.ndarray: class FlattenTransformer(BaseStatefulTransformer[FlattenSettings, AxisArray, AxisArray, FlattenState]): def _reset_state(self, message: AxisArray) -> None: - preserve_axis = self.settings.preserve_axis or message.dims[0] + preserve_axis = self.settings.preserve_axis or resolve_chunk_dim(message, self.STREAMING_DIMS) if preserve_axis not in message.dims: raise ValueError(f"preserve_axis {preserve_axis!r} not found in dims {message.dims}") diff --git a/src/ezmsg/sigproc/merge.py b/src/ezmsg/sigproc/merge.py index b82c0de..518e666 100644 --- a/src/ezmsg/sigproc/merge.py +++ b/src/ezmsg/sigproc/merge.py @@ -12,6 +12,7 @@ from .align import AlignAlongAxis, AlignAlongAxisProcessor, AlignAlongAxisSettings from .concat import Concat, ConcatProcessor, ConcatSettings +from .util.deprecation import suppress_axis_deprecation class MergeSettings(ez.Settings): @@ -56,12 +57,16 @@ class MergeProcessor: def __init__(self, settings: MergeSettings): self.settings = settings - self._align = AlignAlongAxisProcessor( - settings=AlignAlongAxisSettings( - axis=settings.align_axis or "time", - buffer_dur=settings.buffer_dur, + # `align_axis` is Merge's own setting; forwarding it must not warn about + # AlignAlongAxisSettings. Passed through rather than defaulted to "time", + # so that leaving it unset follows the stream's chunk dimension. + with suppress_axis_deprecation(): + self._align = AlignAlongAxisProcessor( + settings=AlignAlongAxisSettings( + axis=settings.align_axis, + buffer_dur=settings.buffer_dur, + ) ) - ) self._concat = ConcatProcessor( settings=ConcatSettings( axis=settings.axis, @@ -119,12 +124,13 @@ class Merge(ez.Collection): CONCAT = Concat() def configure(self) -> None: - self.ALIGN.apply_settings( - AlignAlongAxisSettings( - axis=self.SETTINGS.align_axis or "time", - buffer_dur=self.SETTINGS.buffer_dur, + with suppress_axis_deprecation(): + self.ALIGN.apply_settings( + AlignAlongAxisSettings( + axis=self.SETTINGS.align_axis, + buffer_dur=self.SETTINGS.buffer_dur, + ) ) - ) self.CONCAT.apply_settings( ConcatSettings( axis=self.SETTINGS.axis, diff --git a/src/ezmsg/sigproc/resample.py b/src/ezmsg/sigproc/resample.py index feb2576..9669d01 100644 --- a/src/ezmsg/sigproc/resample.py +++ b/src/ezmsg/sigproc/resample.py @@ -19,7 +19,8 @@ from .util.axisarray_buffer import HybridAxisArrayBuffer, HybridAxisBuffer from .util.buffer import UpdateStrategy -from .util.message import has_samples_along +from .util.deprecation import warn_axis_deprecated +from .util.message import has_samples_along, resolve_configured_chunk_dim def _as_limit(value: float | None) -> float | None: @@ -35,7 +36,14 @@ def _as_limit(value: float | None) -> float | None: class ResampleSettings(ez.Settings): - axis: str = "time" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) resample_rate: float | None = None """target resample rate in Hz. If None, the resample rate will be determined by the reference signal.""" @@ -103,6 +111,9 @@ class ResampleSettings(ez.Settings): @processor_state class ResampleState: + axis: str = "" + """The resolved chunk dimension, fixed at reset so every later use agrees.""" + src_buffer: HybridAxisArrayBuffer | None = None """ Buffer for the incoming signal data. This is the source for training the interpolation function. @@ -164,13 +175,29 @@ class ResampleProcessor(BaseStatefulProcessor[ResampleSettings, AxisArray, AxisA # `resample_rate` / `buffer_duration` / `axis` all size cached buffers. NONRESET_SETTINGS_FIELDS = frozenset({"max_chunk_delay", "fill_value", "reference_reset_after_chunks"}) + def _seed_axis(self, message: AxisArray) -> str: + """Resolve the chunk dimension, seeding it if the reference stream got here first. + + :meth:`push_reference` is an independent entry point and can be called + before any signal message has arrived, while :meth:`__next__` has no + message at all. Both read the resolved value off state, so it is fixed + once and shared rather than re-derived from whichever message happens to + be in hand. + """ + if not self.state.axis: + self.state.axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") + return self.state.axis + def _reset_state(self, message: AxisArray) -> None: """ Reset the internal state based on the incoming message. """ + # The signal stream is authoritative, so this overwrites any value the + # reference stream seeded above. + self.state.axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") self.state.src_buffer = HybridAxisArrayBuffer( duration=self.settings.buffer_duration, - axis=self.settings.axis, + axis=self.state.axis, update_strategy=self.settings.buffer_update_strategy, overflow_strategy="grow", ) @@ -179,7 +206,7 @@ def _reset_state(self, message: AxisArray) -> None: self.state.ref_axis_buffer = HybridAxisBuffer( duration=self.settings.buffer_duration, ) - in_ax = message.axes[self.settings.axis] + in_ax = message.axes[self.state.axis] out_gain = 1 / self.settings.resample_rate t0 = in_ax.data[0] if hasattr(in_ax, "data") else in_ax.value(0) self.state.last_ref_ax_val = t0 - out_gain @@ -204,8 +231,9 @@ def _axis_first_step(ax) -> tuple[float, float]: return first, step def push_reference(self, message: AxisArray) -> None: - ax = message.axes[self.settings.axis] - ax_idx = message.get_axis_idx(self.settings.axis) + axis = self._seed_axis(message) + ax = message.axes[axis] + ax_idx = message.get_axis_idx(axis) n = message.data.shape[ax_idx] if self.state.ref_axis_buffer is None: self.state.ref_axis_buffer = HybridAxisBuffer( @@ -220,7 +248,7 @@ def push_reference(self, message: AxisArray) -> None: # buffer above so a shared index addresses the same sample in both. self.state.ref_data_buffer = HybridAxisArrayBuffer( duration=self.settings.buffer_duration, - axis=self.settings.axis, + axis=axis, update_strategy=self.settings.buffer_update_strategy, overflow_strategy="grow", ) @@ -262,9 +290,9 @@ def _process(self, message: AxisArray) -> None: # If we are resampling at a prescribed rate (i.e., not by reference msgs), # then we use this opportunity to extend our synthetic reference axis. - ax_idx = message.get_axis_idx(self.settings.axis) + ax_idx = message.get_axis_idx(self.state.axis) if self.settings.resample_rate is not None and message.data.shape[ax_idx] > 0: - in_ax = message.axes[self.settings.axis] + in_ax = message.axes[self.state.axis] in_t_end = in_ax.data[-1] if hasattr(in_ax, "data") else in_ax.value(message.data.shape[ax_idx] - 1) out_gain = 1 / self.settings.resample_rate prev_t_end = self.state.last_ref_ax_val @@ -293,7 +321,7 @@ def __next__(self) -> AxisArray: src_axarr, axes={ **src_axarr.axes, - self.settings.axis: ref.peek(0), + self.state.axis: ref.peek(0), }, ) @@ -322,8 +350,8 @@ def __next__(self) -> AxisArray: # Get source to train interpolation. The buffer preserves the input layout, # so the resample axis is wherever the incoming messages put it. src_axarr = src.peek() - src_ax_idx = src_axarr.get_axis_idx(self.settings.axis) - src_axis = src_axarr.axes[self.settings.axis] + src_ax_idx = src_axarr.get_axis_idx(self.state.axis) + src_axis = src_axarr.axes[self.state.axis] x = src_axis.data if hasattr(src_axis, "data") else src_axis.value(np.arange(src_axarr.data.shape[src_ax_idx])) # Only resample at reference values that have not been interpolated over previously. @@ -339,7 +367,7 @@ def __next__(self) -> AxisArray: return replace( src_axarr, data=slice_along_axis(src_axarr.data, slice(0, 0), src_ax_idx), - axes={**src_axarr.axes, self.settings.axis: null_ref}, + axes={**src_axarr.axes, self.state.axis: null_ref}, ) xnew = ref_xvec[ref_idx] @@ -389,7 +417,7 @@ def __next__(self) -> AxisArray: data=resampled_data, axes={ **src_axarr.axes, - self.settings.axis: out_ax, + self.state.axis: out_ax, }, ) @@ -399,13 +427,13 @@ def __next__(self) -> AxisArray: if self.state.ref_data_buffer is not None: ref_axarr = self.state.ref_data_buffer.peek() if ref_axarr is not None: - ref_ax_idx = ref_axarr.get_axis_idx(self.settings.axis) + ref_ax_idx = ref_axarr.get_axis_idx(self.state.axis) if ref_axarr.data.shape[ref_ax_idx] == ref.available(): ref_xp = get_namespace(ref_axarr.data) self.state.reference_output = replace( ref_axarr, data=ref_xp.take(ref_axarr.data, ref_idx, axis=ref_ax_idx), - axes={**ref_axarr.axes, self.settings.axis: out_ax}, + axes={**ref_axarr.axes, self.state.axis: out_ax}, ) # Update the state. For state buffers, seek beyond samples that are no longer needed. @@ -480,7 +508,7 @@ async def gen_resampled(self): # pre-init null template (which lacks the axis entirely) means # "nothing ready". A chunk that is empty only along other axes # (e.g. zero channels) is still real output and must be published. - if not has_samples_along(result, self.SETTINGS.axis): + if not has_samples_along(result, self.processor.state.axis): break yield self.OUTPUT_SIGNAL, result ref_out = self.processor.state.reference_output diff --git a/src/ezmsg/sigproc/rollingscaler.py b/src/ezmsg/sigproc/rollingscaler.py index 9e27f45..7e86f0a 100644 --- a/src/ezmsg/sigproc/rollingscaler.py +++ b/src/ezmsg/sigproc/rollingscaler.py @@ -14,9 +14,20 @@ from ezmsg.util.messages.axisarray import AxisArray from ezmsg.util.messages.util import replace +from .util.deprecation import warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim + class RollingScalerSettings(ez.Settings): - axis: str = "time" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) + """ Axis along which samples are arranged. """ @@ -65,6 +76,9 @@ class RollingScalerSettings(ez.Settings): @processor_state class RollingScalerState: + axis: str = "" + """The resolved chunk dimension, fixed at reset so every later use agrees.""" + mean: npt.NDArray | None = None N: int = 0 M2: npt.NDArray | None = None @@ -109,13 +123,14 @@ class RollingScalerProcessor(BaseAdaptiveTransformer[RollingScalerSettings, Axis NONRESET_SETTINGS_FIELDS = frozenset({"update_with_signal", "artifact_z_thresh", "clip"}) def _reset_state(self, message: AxisArray) -> None: + self._state.axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") xp = get_namespace(message.data) ch = message.data.shape[-1] self._state.mean = xp.zeros(ch, dtype=xp.float64) self._state.N = 0 self._state.M2 = xp.zeros(ch, dtype=xp.float64) self._state.k_samples = ( - math.ceil(self.settings.window_size / message.axes[self.settings.axis].gain) + math.ceil(self.settings.window_size / message.axes[self._state.axis].gain) if self.settings.window_size is not None else self.settings.k_samples ) @@ -126,7 +141,7 @@ def _reset_state(self, message: AxisArray) -> None: ez.logger.warning("k_samples is None; z-score accumulation will be unbounded.") self._state.samples = deque(maxlen=self._state.k_samples) self._state.min_samples = ( - math.ceil(self.settings.min_seconds / message.axes[self.settings.axis].gain) + math.ceil(self.settings.min_seconds / message.axes[self._state.axis].gain) if self.settings.window_size is not None else self.settings.min_samples ) diff --git a/src/ezmsg/sigproc/sampler.py b/src/ezmsg/sigproc/sampler.py index 69a3360..154af3a 100644 --- a/src/ezmsg/sigproc/sampler.py +++ b/src/ezmsg/sigproc/sampler.py @@ -23,7 +23,8 @@ from .util.axisarray_buffer import HybridAxisArrayBuffer from .util.buffer import UpdateStrategy -from .util.message import SampleTriggerMessage +from .util.deprecation import warn_axis_deprecated +from .util.message import SampleTriggerMessage, resolve_configured_chunk_dim from .util.profile import profile_subpub @@ -41,11 +42,13 @@ class SamplerSettings(ez.Settings): """ axis: str | None = None - """ - The axis along which to sample the data. - None (default) will choose the first axis in the first input. - Note: (for now) the axis must exist in the msg .axes and be of type AxisArray.LinearAxis - """ + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) period: tuple[float, float] | None = None """Optional default period (in seconds) if unspecified in SampleTriggerMessage.""" @@ -90,7 +93,7 @@ def __call__(self, message: AxisArray | SampleTriggerMessage) -> list[AxisArray] def _reset_state(self, message: AxisArray) -> None: self._state.buffer = HybridAxisArrayBuffer( duration=self.settings.buffer_dur, - axis=self.settings.axis or message.dims[0], + axis=resolve_configured_chunk_dim(self, message, self.settings.axis), update_strategy=self.settings.buffer_update_strategy, overflow_strategy="warn-overwrite", # True circular buffer ) diff --git a/src/ezmsg/sigproc/scaler.py b/src/ezmsg/sigproc/scaler.py index 19ddb98..f223562 100644 --- a/src/ezmsg/sigproc/scaler.py +++ b/src/ezmsg/sigproc/scaler.py @@ -13,19 +13,28 @@ from ezmsg.util.messages.axisarray import AxisArray from ezmsg.util.messages.util import replace -# Imports for backwards compatibility with previous module location from .ewma import EWMA_Deprecated as EWMA_Deprecated from .ewma import EWMASettings, EWMATransformer, _alpha_from_tau from .ewma import _tau_from_alpha as _tau_from_alpha from .ewma import ewma_step as ewma_step +# Imports for backwards compatibility with previous module location +from .util.deprecation import suppress_axis_deprecation, warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim + class RiverAdaptiveStandardScalerSettings(ez.Settings): time_constant: float = 1.0 """Decay constant ``tau`` in seconds.""" axis: str | None = None - """The name of the axis to accumulate statistics over.""" + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) @processor_state @@ -55,12 +64,11 @@ class RiverAdaptiveStandardScalerTransformer( def _reset_state(self, message: AxisArray) -> None: from river import preprocessing - axis = self.settings.axis - if axis is None: - axis = message.dims[0] - self._state.axis_idx = 0 - else: - self._state.axis_idx = message.get_axis_idx(axis) + # The index is looked up unconditionally. The resolved axis is the one + # messages accumulate along, which is not necessarily the leading one: + # assuming index 0 for it transposed the data silently. + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) + self._state.axis_idx = message.get_axis_idx(axis) self._state.axis = axis alpha = _alpha_from_tau(self.settings.time_constant, message.axes[axis].gain) @@ -147,16 +155,21 @@ def _hash_message(self, message: AxisArray) -> int: return 0 def _reset_state(self, message: AxisArray) -> None: - self._state.samps_ewma = EWMATransformer( - time_constant=self.settings.time_constant, - axis=self.settings.axis, - accumulate=self.settings.accumulate, - ) - self._state.vars_sq_ewma = EWMATransformer( - time_constant=self.settings.time_constant, - axis=self.settings.axis, - accumulate=self.settings.accumulate, - ) + # One user-visible setting, already warned about when this transformer's + # own settings were built. Forwarding it must not warn again -- and this + # runs mid-stream, so the warning would name the pipeline driver rather + # than any call site. + with suppress_axis_deprecation(): + self._state.samps_ewma = EWMATransformer( + time_constant=self.settings.time_constant, + axis=self.settings.axis, + accumulate=self.settings.accumulate, + ) + self._state.vars_sq_ewma = EWMATransformer( + time_constant=self.settings.time_constant, + axis=self.settings.axis, + accumulate=self.settings.accumulate, + ) @property def accumulate(self) -> bool: diff --git a/src/ezmsg/sigproc/slicer.py b/src/ezmsg/sigproc/slicer.py index 2080df9..ccc2b74 100644 --- a/src/ezmsg/sigproc/slicer.py +++ b/src/ezmsg/sigproc/slicer.py @@ -17,7 +17,7 @@ slice_along_axis, ) -from .util.message import with_fingerprint +from .util.message import resolve_feature_dim, with_fingerprint """ Slicer:Select a subset of data along a particular axis. @@ -240,7 +240,7 @@ def _reset_state(self, message: AxisArray) -> None: raise ValueError(f"on_empty must be 'raise' or 'warn', got {self.settings.on_empty!r}") if self.settings.order not in ("axis", "selection"): raise ValueError(f"order must be 'axis' or 'selection', got {self.settings.order!r}") - axis = self.settings.axis or message.dims[-1] + axis = self.settings.axis or resolve_feature_dim(message) axis_idx = message.get_axis_idx(axis) axinfo = message.axes.get(axis, None) self._state.new_axis = None @@ -296,7 +296,7 @@ def _reset_state(self, message: AxisArray) -> None: self._state.new_axis = with_fingerprint(replace(message.axes[axis], data=out_data)) def _process(self, message: AxisArray) -> AxisArray: - axis = self.settings.axis or message.dims[-1] + axis = self.settings.axis or resolve_feature_dim(message) axis_idx = message.get_axis_idx(axis) replace_kwargs = {} diff --git a/src/ezmsg/sigproc/spectrum.py b/src/ezmsg/sigproc/spectrum.py index 93a2861..bae5f76 100644 --- a/src/ezmsg/sigproc/spectrum.py +++ b/src/ezmsg/sigproc/spectrum.py @@ -20,6 +20,7 @@ ) from .util.array import is_complex_dtype +from .util.message import resolve_transform_dim class OptionsEnum(enum.Enum): @@ -84,6 +85,11 @@ class SpectrumSettings(ez.Settings): """ The name of the axis on which to calculate the spectrum. Note: The axis must have an .axes entry of type LinearAxis, not CoordinateAxis. + + Defaults to the innermost non-chunk dimension carrying a LinearAxis, else + the chunk dimension itself -- ``"time"`` for both a raw ``(time, ch)`` + stream and a windowed ``(win, time, ch)`` one, where each window's + spectrum is taken over ``time`` while ``win`` is what accumulates. """ # n: int | None = None # n parameter for fft @@ -143,7 +149,7 @@ def _hash_message(self, message: AxisArray) -> int: back in. The dtype matters because a complex input takes a different branch and produces a different frequency axis. """ - axis = self.settings.axis or message.dims[0] + axis = self.settings.axis or resolve_transform_dim(message, self.STREAMING_DIMS) return self._message_hash( message, extra=( @@ -153,7 +159,7 @@ def _hash_message(self, message: AxisArray) -> int: ) def _reset_state(self, message: AxisArray) -> None: - axis = self.settings.axis or message.dims[0] + axis = self.settings.axis or resolve_transform_dim(message, self.STREAMING_DIMS) ax_idx = message.get_axis_idx(axis) ax_info = message.axes[axis] targ_len = message.data.shape[ax_idx] @@ -258,7 +264,7 @@ def f_transform(x): self.state.f_transform = f_transform def _process(self, message: AxisArray) -> AxisArray: - axis = self.settings.axis or message.dims[0] + axis = self.settings.axis or resolve_transform_dim(message, self.STREAMING_DIMS) new_axes = {k: v for k, v in message.axes.items() if k not in [self.settings.out_axis, axis]} new_axes[self.settings.out_axis or axis] = self.state.freq_axis diff --git a/src/ezmsg/sigproc/util/channels.py b/src/ezmsg/sigproc/util/channels.py index 5ce6f44..bc4c6de 100644 --- a/src/ezmsg/sigproc/util/channels.py +++ b/src/ezmsg/sigproc/util/channels.py @@ -40,6 +40,8 @@ import numpy as np from ezmsg.util.messages.axisarray import AxisArray +from .message import resolve_feature_dim + # Whether AxisFingerprintMemo counts its hits and misses. Off unless the env var # is set, because the answer it gives -- do axis objects survive, or is every # message a fresh deserialization? -- is a property of a *deployed* graph's @@ -139,7 +141,7 @@ def channel_groups_from_field( than a single all-channel group lets callers distinguish "no metadata, fall back to my default" from "one bank". """ - axis = axis or message.dims[-1] + axis = axis or resolve_feature_dim(message) ax = message.axes.get(axis) data = getattr(ax, "data", None) names = getattr(getattr(data, "dtype", None), "names", None) @@ -177,13 +179,13 @@ def resolve_channel_groups( raise ValueError(f"channel group spec mixes field names with index groups: {spec!r}") groups = channel_groups_from_field(message, axis, fields) elif callable(spec): - groups = spec(message, axis or message.dims[-1]) + groups = spec(message, axis or resolve_feature_dim(message)) else: groups = spec if groups is None: return None - axis = axis or message.dims[-1] + axis = axis or resolve_feature_dim(message) out = [np.asarray(group, dtype=np.intp).reshape(-1) for group in groups] validate_channel_groups(out, message.data.shape[message.get_axis_idx(axis)]) return out @@ -240,7 +242,7 @@ def group_spec_fingerprint( fields = (spec,) if isinstance(spec, str) else group_spec_fields(spec) if fields is None: return () - ax = message.axes.get(axis or message.dims[-1]) + ax = message.axes.get(axis or resolve_feature_dim(message)) names = getattr(getattr(getattr(ax, "data", None), "dtype", None), "names", None) return (bool(names) and all(field in names for field in fields),) @@ -331,7 +333,7 @@ def coord_value_fingerprint( itemsize, so its ``tobytes()`` is the entire 27.6 kB -- asking for two of eight fields would otherwise cost more than asking for all of them. """ - ax = message.axes.get(axis or message.dims[-1]) + ax = message.axes.get(axis or resolve_feature_dim(message)) data = getattr(ax, "data", None) if data is None: return () diff --git a/src/ezmsg/sigproc/util/deprecation.py b/src/ezmsg/sigproc/util/deprecation.py new file mode 100644 index 0000000..388c9b9 --- /dev/null +++ b/src/ezmsg/sigproc/util/deprecation.py @@ -0,0 +1,119 @@ +"""Deprecation of the per-processor ``axis`` setting. + +A processor that carries state *between* messages -- filter initial conditions, +a running mean, a sample buffer, a previous-sample cache -- can only do so along +the dimension messages accumulate along. Carrying it along a static axis is not +a smaller error but a different operation: that axis has the same length every +message, so the carried state applies message N's tail to message N+1's head at +the same coordinate, forever. + +Which dimension that is belongs to the producer, and +:attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim` is where it says so. +A setting that lets a consumer disagree can only be used to be wrong, so it is +going away; see :func:`~ezmsg.sigproc.util.message.resolve_chunk_dim`. + +During the deprecation window the setting is still honoured, so nothing changes +behaviour until it is removed. Two warnings partition the call sites: + +* This module's construction-time :class:`FutureWarning` fires for *every* use, + including a harmless ``axis="time"`` on a raw stream. It means "delete this". +* :func:`~ezmsg.sigproc.util.message.resolve_configured_chunk_dim`'s runtime + warning fires only when the configured axis disagrees with a *declared* + ``chunk_dim``. It means "deleting this will change what this stage computes". + +To find every remaining call site in a pipeline, run its tests with +``-W error::FutureWarning``. +""" + +import sys +import typing +import warnings +from contextlib import contextmanager +from contextvars import ContextVar + +__all__ = [ + "AXIS_REMOVAL_VERSION", + "suppress_axis_deprecation", + "warn_axis_deprecated", +] + +AXIS_REMOVAL_VERSION = "4.0" +"""Release that drops the deprecated ``axis`` settings. Deprecated in 3.8.""" + +_suppressed: ContextVar[bool] = ContextVar("_axis_deprecation_suppressed", default=False) + + +@contextmanager +def suppress_axis_deprecation() -> typing.Iterator[None]: + """Silence the construction-time warning while forwarding a setting internally. + + A stage that builds a child processor from its own already-warned settings + (:obj:`~ezmsg.sigproc.scaler.AdaptiveStandardScalerTransformer` and its two + EWMAs, :obj:`~ezmsg.sigproc.decimate.Decimate` and its anti-alias filter) + would otherwise warn a second time for one user-visible setting -- and, since + some of that forwarding happens in ``_reset_state``, would warn mid-stream + pointing at whatever is driving the pipeline rather than at any call site. + + Deleted along with the settings themselves; see :data:`AXIS_REMOVAL_VERSION`. + """ + token = _suppressed.set(True) + try: + yield + finally: + _suppressed.reset(token) + + +def _user_stacklevel() -> int: + """``stacklevel`` that makes a ``warn()`` in this module's caller point at + the first frame outside ezmsg. + + A fixed level cannot work: the depth differs between + ``WindowSettings(axis=...)`` (the dataclass ``__init__``) and + ``Window(axis=...)`` (through ``_unify_settings``), and the functional + factories build the settings object inside this package, so a fixed level + would blame ``scaler.py`` for a call the user made. + """ + # Frames are identified by module rather than by filename. The dataclass + # __init__ every one of these settings objects is constructed through is + # generated by exec, so its co_filename is "" -- indistinguishable + # from a user running `python -c`. Its globals, though, are the defining + # module's, so f_globals says "ezmsg.sigproc.window" and the walk continues + # correctly in both cases. + # + # 0 is this function, 1 is warn_axis_deprecated -- the frame a stacklevel of + # 1 would name. Walk out from there until we leave the package. + frame: typing.Any = sys._getframe(1) + level = 1 + while frame is not None: + module = frame.f_globals.get("__name__", "") + if module != "ezmsg" and not module.startswith("ezmsg."): + return level + if frame.f_back is None: + # Never left the package (a settings object built at import time, or + # from a bare exec). Blaming the outermost frame beats pointing the + # user at `sys:1`. + return level + frame = frame.f_back + level += 1 + return level + + +def warn_axis_deprecated(settings: typing.Any, field: str = "axis") -> None: + """Warn that *settings*' ``field`` is deprecated, if it was actually set. + + Call from a ``__post_init__``: ``ez.Settings`` classes are frozen dataclasses + and every construction path -- the settings class, the transformer, the unit, + and the functional factory -- funnels through their ``__init__``, so one hook + covers all four. + """ + if getattr(settings, field, None) is None or _suppressed.get(): + return + warnings.warn( + f"{type(settings).__name__}.{field} is deprecated and will be removed in " + f"ezmsg-sigproc {AXIS_REMOVAL_VERSION}. This processor carries state between " + f"messages, which is only meaningful along the dimension they accumulate " + f"along; that dimension now comes from AxisArray.chunk_dim. Drop the setting. " + f"If the stream's chunk_dim is wrong, fix it at the producer.", + FutureWarning, + stacklevel=_user_stacklevel(), + ) diff --git a/src/ezmsg/sigproc/util/message.py b/src/ezmsg/sigproc/util/message.py index 5011b63..49f732b 100644 --- a/src/ezmsg/sigproc/util/message.py +++ b/src/ezmsg/sigproc/util/message.py @@ -7,6 +7,7 @@ import typing +import ezmsg.core as ez from ezmsg.baseproc.util.message import ( SampleMessage, SampleTriggerMessage, @@ -20,9 +21,147 @@ "has_samples_along", "is_empty_along", "is_sample_message", + "resolve_chunk_dim", + "resolve_configured_chunk_dim", + "resolve_feature_dim", + "resolve_transform_dim", "with_fingerprint", ] +STREAMING_DIMS: tuple[str, ...] = ("time",) +"""Default fallback chunk dimension, matching ``BaseStatefulTransformer``.""" + + +def resolve_chunk_dim(message: AxisArray, streaming_dims: typing.Iterable[str] = STREAMING_DIMS) -> str: + """The dimension successive messages accumulate along. + + This is the axis a processor that carries state *between* messages must + operate on -- filter initial conditions, a running mean, a sample buffer, + a previous-sample cache. Carrying such state along any other dimension is + not a smaller error but a different operation: a static axis has the same + length every message, so state carried across it applies message N's tail + to message N+1's head at the same coordinate, forever. + + The producer renamed the dims and so is the only party that reliably knows + which one grows; ``message.chunk_dim`` is that declaration. When a producer + is silent, *streaming_dims* supplies the guess -- ``("time",)`` is right for + a raw signal and wrong downstream of a windowing stage, where the message is + ``(win, time, ch)`` and ``win`` is what grows. + + ``dims[0]`` is the last resort only. It is a position, not a meaning, and it + breaks under :meth:`~ezmsg.util.messages.axisarray.AxisArray.transpose`. + """ + if message.chunk_dim is not None: + return message.chunk_dim + for name in streaming_dims: + if name in message.dims: + return name + return message.dims[0] + + +def resolve_configured_chunk_dim( + processor: typing.Any, + message: AxisArray, + configured: str | None, + legacy_default: str | None = None, +) -> str: + """Resolve a state-carrying processor's axis, honouring an explicit setting. + + *configured* wins when set -- an explicit axis is an instruction, and + removing that escape hatch would break every pipeline that passes the + common ``axis="time"``. But when the producer *declared* a different chunk + dimension, that disagreement is worth surfacing exactly once: the + processor's cross-message state is about to be carried along an axis whose + length is fixed, which is a different operation from the one the caller + almost certainly meant. + + The warning fires only against a declared ``chunk_dim``, never against the + :attr:`STREAMING_DIMS` guess -- warning on a guess would fire on every + correctly-configured windowed pipeline whose producer is merely silent. + + :param legacy_default: The dimension this processor's ``axis`` setting used + to default to, for the stages whose default was a hardcoded ``"time"`` + rather than a positional guess. Flipping those to follow ``chunk_dim`` + changes results wherever the chunk dimension is not ``"time"`` -- most + obviously downstream of a windowing stage, where it is ``"win"`` -- and + unlike an explicitly configured axis there is nothing in the settings to + warn about. Passing the old default here surfaces exactly that + population, once, and is dropped when the setting is removed. + """ + resolved = resolve_chunk_dim(message, getattr(processor, "STREAMING_DIMS", STREAMING_DIMS)) + if configured is None: + if ( + legacy_default is not None + and resolved != legacy_default + and legacy_default in message.dims + and not getattr(processor, "_legacy_axis_default_warned", False) + ): + processor._legacy_axis_default_warned = True + ez.logger.warning( + f"{type(processor).__name__} used to operate on axis={legacy_default!r} by default; it now " + f"follows the stream's chunk_dim={resolved!r}. This changes its output. The old behaviour was " + f"carrying state across messages along {legacy_default!r}, whose length does not grow, so this " + f"is a fix -- but pass axis={legacy_default!r} explicitly to keep the previous behaviour." + ) + return resolved + if ( + message.chunk_dim is not None + and configured != message.chunk_dim + and configured in message.dims + and not getattr(processor, "_chunk_dim_mismatch_warned", False) + ): + processor._chunk_dim_mismatch_warned = True + ez.logger.warning( + f"{type(processor).__name__} is configured with axis={configured!r} but messages declare " + f"chunk_dim={message.chunk_dim!r}. State carried between messages will be applied along " + f"{configured!r}, whose length does not grow. Set axis=None to follow the declared chunk dimension." + ) + return configured + + +def resolve_feature_dim(message: AxisArray, position: int = -1) -> str: + """The dimension at *position*, skipping the chunk dimension. + + For processors whose axis is a *static* one -- channels, coordinate + components, feature labels. ``chunk_dim`` is emphatically not the answer + here, but the naive ``dims[position]`` can silently *be* the chunk + dimension: a ``(ch, time)`` stream makes ``dims[-1]`` the accumulating axis, + and an affine transform would then matmul across time while a slicer would + discard samples. + + Falls back to ``dims[position]`` when the chunk dimension is all there is, + which keeps 1-D messages working rather than raising on them. + """ + candidates = [d for d in message.dims if d != message.chunk_dim] + if not candidates: + return message.dims[position] + return candidates[position] + + +def resolve_transform_dim(message: AxisArray, streaming_dims: typing.Iterable[str] = STREAMING_DIMS) -> str: + """The regularly-sampled dimension a transform consumes. + + Neither :func:`resolve_chunk_dim` nor :func:`resolve_feature_dim` fits a + stage like :obj:`~ezmsg.sigproc.spectrum.Spectrum`, which needs the axis + whose ``gain`` is a sample period and whose extent is the transform length: + + * On a raw ``(time, ch)`` stream that *is* the chunk dimension. + * On windowed ``(win, time, ch)`` it is ``time`` -- ``win`` is what + accumulates, but each window's spectrum is taken over ``time``. + + So: prefer the innermost non-chunk dimension carrying a ``LinearAxis``, and + fall back to the chunk dimension when there is none. ``ch`` carries a + ``CoordinateAxis`` (or no axis at all), so the raw case falls through + correctly rather than transforming across channels. + """ + chunk_dim = resolve_chunk_dim(message, streaming_dims) + for name in reversed(message.dims): + if name == chunk_dim: + continue + if isinstance(message.axes.get(name), AxisArray.LinearAxis): + return name + return chunk_dim + def with_fingerprint(axis: AxisArray.CoordinateAxis) -> AxisArray.CoordinateAxis: """Compute *axis*'s fingerprint now, and return the axis. diff --git a/src/ezmsg/sigproc/wavelets.py b/src/ezmsg/sigproc/wavelets.py index 3890b71..9300660 100644 --- a/src/ezmsg/sigproc/wavelets.py +++ b/src/ezmsg/sigproc/wavelets.py @@ -16,7 +16,8 @@ from ezmsg.util.messages.util import replace from .filterbank import FilterbankMode, MinPhaseMode, filterbank -from .util.message import with_fingerprint +from .util.deprecation import suppress_axis_deprecation, warn_axis_deprecated +from .util.message import resolve_configured_chunk_dim, with_fingerprint class CWTSettings(ez.Settings): @@ -28,12 +29,23 @@ class CWTSettings(ez.Settings): frequencies: list | tuple | npt.NDArray | None wavelet: str | pywt.ContinuousWavelet | pywt.Wavelet min_phase: MinPhaseMode = MinPhaseMode.NONE - axis: str = "time" + axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) + scales: list | tuple | npt.NDArray | None = None @processor_state class CWTState: + axis: str = "" + """The resolved chunk dimension, fixed at reset so every later use agrees.""" + neg_rt_scales: npt.NDArray | None = None int_psi_scales: list[npt.NDArray] | None = None template: AxisArray | None = None @@ -47,6 +59,7 @@ def _hash_message(self, message: AxisArray) -> int: return self._message_hash(message, extra=(message.data.dtype.kind,)) def _reset_state(self, message: AxisArray) -> None: + self._state.axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") if "freq" in message.dims: raise ValueError( "CWT appends a 'freq' axis to its output, but the input already has one " @@ -70,7 +83,7 @@ def _reset_state(self, message: AxisArray) -> None: frequencies = np.sort(np.array(self.settings.frequencies)) scales = pywt.frequency2scale( wavelet, - frequencies * message.axes[self.settings.axis].gain, + frequencies * message.axes[self._state.axis].gain, precision=precision, ) else: @@ -97,21 +110,25 @@ def _reset_state(self, message: AxisArray) -> None: self._state.int_psi_scales.append(int_psi[reix][::-1]) # Setup filterbank generator - self._state.fbgen = filterbank( - self._state.int_psi_scales, - mode=FilterbankMode.CONV, - min_phase=self.settings.min_phase, - axis=self.settings.axis, - ) + # The child filterbank is handed the axis this stage already resolved, + # so it convolves the same dimension; that forwarding is ours, not a + # setting the user can drop. + with suppress_axis_deprecation(): + self._state.fbgen = filterbank( + self._state.int_psi_scales, + mode=FilterbankMode.CONV, + min_phase=self.settings.min_phase, + axis=self._state.axis, + ) # Create output template. - ax_idx = message.get_axis_idx(self.settings.axis) + ax_idx = message.get_axis_idx(self._state.axis) in_shape = message.data.shape[:ax_idx] + message.data.shape[ax_idx + 1 :] - freqs = pywt.scale2frequency(wavelet, scales, precision) / message.axes[self.settings.axis].gain + freqs = pywt.scale2frequency(wavelet, scales, precision) / message.axes[self._state.axis].gain dummy_shape = in_shape + (len(scales), 0) self._state.template = AxisArray( np.zeros(dummy_shape, dtype=dt_cplx if wavelet.complex_cwt else dt_data), - dims=message.dims[:ax_idx] + message.dims[ax_idx + 1 :] + ["freq", self.settings.axis], + dims=message.dims[:ax_idx] + message.dims[ax_idx + 1 :] + ["freq", self._state.axis], axes={ **{k: deepcopy(v) for k, v in message.axes.items()}, "freq": with_fingerprint(AxisArray.CoordinateAxis(unit="Hz", data=freqs, dims=["freq"])), @@ -141,7 +158,7 @@ def _process(self, message: AxisArray) -> AxisArray: data=coef, axes={ **self._state.template.axes, - self.settings.axis: message.axes[self.settings.axis], + self._state.axis: message.axes[self._state.axis], }, ) diff --git a/src/ezmsg/sigproc/window.py b/src/ezmsg/sigproc/window.py index ba0b213..d42c1da 100644 --- a/src/ezmsg/sigproc/window.py +++ b/src/ezmsg/sigproc/window.py @@ -22,7 +22,8 @@ from .util.array import xp_empty from .util.buffer import HybridBuffer, UpdateStrategy -from .util.message import is_empty_along +from .util.deprecation import warn_axis_deprecated +from .util.message import is_empty_along, resolve_configured_chunk_dim from .util.profile import profile_subpub from .util.sparse import sliding_win_oneaxis as sparse_sliding_win_oneaxis @@ -35,6 +36,13 @@ class Anchor(enum.Enum): class WindowSettings(ez.Settings): axis: str | None = None + """.. deprecated:: 3.8 + Scheduled for removal in 4.0. The dimension messages accumulate along + now comes from :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`; + see :mod:`ezmsg.sigproc.util.deprecation`.""" + + def __post_init__(self) -> None: + warn_axis_deprecated(self) newaxis: str | None = None """Name of the axis windows are delimited on, inserted before ``axis``. @@ -262,7 +270,7 @@ def _reset_state(self, message: AxisArray) -> None: # disagree with the axes key _process writes. _newaxis = self.settings.newaxis - axis = self.settings.axis or message.dims[0] + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) axis_idx = message.get_axis_idx(axis) axis_info = message.get_axis(axis) fs = 1.0 / axis_info.gain @@ -367,7 +375,7 @@ def _drop_front(self, n: int, axis_idx: int) -> None: self._state.buffer_len -= min(n, self._state.buffer_len) def _process(self, message: AxisArray) -> AxisArray: - axis = self.settings.axis or message.dims[0] + axis = resolve_configured_chunk_dim(self, message, self.settings.axis) axis_idx = message.get_axis_idx(axis) axis_info = message.get_axis(axis) @@ -500,9 +508,10 @@ async def on_signal(self, message: AxisArray) -> typing.AsyncGenerator: # TODO: The transfomer overwrites settings.newaxis from None to "win", # then we no longer know if the user wants to trim out the newaxis from the unit. xp = get_namespace(message.data) - # `axis` defaults to the input's first dim, matching WindowTransformer. - # Resolve it from the *input*, since the output has `win` prepended. - axis = self.SETTINGS.axis or message.dims[0] + # Must resolve exactly as WindowTransformer does, or the emptiness gate + # below checks a different dim than the one that was windowed. Resolved + # from the *input*, since the output has `win` prepended. + axis = resolve_configured_chunk_dim(self.processor, message, self.SETTINGS.axis) try: ret = self.processor(message) # Swallow only when no complete windows (or, in pass-through mode, no diff --git a/tests/unit/test_axis_deprecation.py b/tests/unit/test_axis_deprecation.py new file mode 100644 index 0000000..5d62c32 --- /dev/null +++ b/tests/unit/test_axis_deprecation.py @@ -0,0 +1,502 @@ +"""The deprecation window for the per-processor ``axis`` setting. + +Deprecated in 3.8, removed in 4.0. See :mod:`ezmsg.sigproc.util.deprecation`. + +The contract under test has four parts: + +* setting ``axis`` warns, once, pointing at the *caller's* line; +* leaving it unset is silent, including across message processing; +* the setting is still honoured, so nothing changes behaviour until removal; +* stages that forward the setting internally do not warn on the user's behalf. + +The last one is what makes the window usable: without it, every filter-by-design +would warn on every state reset, mid-stream, naming whatever happened to be +driving the pipeline. +""" + +import inspect +import sys +import warnings + +import ezmsg.core as ez +import numpy as np +import pytest +from ezmsg.util.messages.axisarray import AxisArray + +from ezmsg.sigproc.butterworthfilter import ( + ButterworthFilterSettings, + ButterworthFilterTransformer, +) +from ezmsg.sigproc.decimate import DecimateSettings +from ezmsg.sigproc.diff import DiffSettings, DiffTransformer +from ezmsg.sigproc.flatten import FlattenSettings, FlattenTransformer +from ezmsg.sigproc.gaussiansmoothing import ( + GaussianSmoothingFilterTransformer, + GaussianSmoothingSettings, +) +from ezmsg.sigproc.merge import MergeProcessor, MergeSettings +from ezmsg.sigproc.scaler import ( + AdaptiveStandardScalerSettings, + AdaptiveStandardScalerTransformer, +) +from ezmsg.sigproc.util.deprecation import ( + AXIS_REMOVAL_VERSION, + suppress_axis_deprecation, +) +from ezmsg.sigproc.window import WindowSettings, WindowTransformer + +FS = 100.0 + +# Every Settings class carrying the deprecation hook. Pinned so that adding or +# dropping one is a deliberate edit rather than a side effect, and so the 4.0 +# removal has a checklist. +DEPRECATED_SETTINGS = { + "AdaptiveLNCSettings", + "AdaptiveLatticeNotchFilterSettings", + "AdaptiveStandardScalerSettings", + "AlignAlongAxisSettings", + "BinnedAggregateSettings", + "ButterworthFilterSettings", + "ButterworthZeroPhaseSettings", + "CWTSettings", + "ChebyshevFilterSettings", + "CombFilterSettings", + "DecimateSettings", + "DiffSettings", + "EWMASettings", + "FIRFilterSettings", + "FIRHilbertFilterSettings", + "FilterBaseSettings", + "FilterSettings", + "FilterbankDesignSettings", + "FilterbankSettings", + "GaussianSmoothingSettings", + "KaiserFilterSettings", + "ParksMcClellanFIRSettings", + "ResampleSettings", + "RiverAdaptiveStandardScalerSettings", + "RollingScalerSettings", + "SamplerSettings", + "WindowSettings", +} + +# The second wave: these defaulted to a hardcoded ``axis="time"`` rather than to +# a positional guess, so flipping them to follow ``chunk_dim`` changes results +# wherever the chunk dimension is not ``"time"``. They pass ``legacy_default`` +# to surface that population; the rest do not. +LEGACY_TIME_DEFAULT = { + "AdaptiveLNCSettings", + "AdaptiveLatticeNotchFilterSettings", + "BinnedAggregateSettings", + "CWTSettings", + "FilterbankDesignSettings", + "ResampleSettings", + "RollingScalerSettings", +} + + +def msg(n_time=32, n_ch=3, chunk_dim="time"): + kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} + return AxisArray( + np.random.default_rng(0).standard_normal((n_time, n_ch)), + dims=["time", "ch"], + axes={"time": AxisArray.TimeAxis(fs=FS)}, + key="dev", + **kwargs, + ) + + +def axis_warnings(records): + return [r for r in records if issubclass(r.category, FutureWarning) and "deprecated" in str(r.message)] + + +class TestTheInventoryIsPinned: + def test_exactly_these_classes_carry_the_hook(self): + import importlib + import pkgutil + + import ezmsg.sigproc + + # The walk reads sys.modules, so every submodule has to be imported + # first -- otherwise this passes by simply not looking at most of them. + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + for info in pkgutil.walk_packages(ezmsg.sigproc.__path__, "ezmsg.sigproc."): + try: + importlib.import_module(info.name) + except Exception: # optional deps (river, mlx, ...) may be absent + pass + + found = set() + for name, mod in list(sys.modules.items()): + if not name.startswith("ezmsg.sigproc"): + continue + for obj in vars(mod).values(): + if not inspect.isclass(obj): + continue + try: + if not issubclass(obj, ez.Settings): + continue + except TypeError: + # On 3.10 `isinstance(tuple[int, str], type)` is True, so a + # module-level generic alias (align.py's `_AlignPair`) gets + # past isclass and then blows up in issubclass. 3.11 made + # that False, which is why this only bit on one matrix job. + continue + if "__post_init__" not in dir(obj): + continue + try: + src = inspect.getsource(obj.__post_init__) + except (OSError, TypeError): + continue + if "warn_axis_deprecated" in src: + found.add(obj.__name__) + assert found == DEPRECATED_SETTINGS + + def test_flatten_is_deliberately_excluded(self): + """Flatten carries no data between messages and already handles the chunk + dimension being folded away, so preserving a non-chunk axis is coherent.""" + assert "FlattenSettings" not in DEPRECATED_SETTINGS + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + FlattenTransformer(FlattenSettings(preserve_axis="time"))(msg()) + assert not axis_warnings(rec) + + +class TestSettingItWarns: + @pytest.mark.parametrize( + "build", + [ + pytest.param(lambda: WindowSettings(axis="time", window_dur=0.1, window_shift=0.05), id="window"), + pytest.param(lambda: DiffSettings(axis="time"), id="diff"), + pytest.param(lambda: ButterworthFilterSettings(axis="time", order=2, cutoff=20.0), id="butterworth"), + pytest.param(lambda: AdaptiveStandardScalerSettings(axis="time"), id="scaler"), + pytest.param(lambda: DecimateSettings(axis="time", target_rate=50.0), id="decimate"), + ], + ) + def test_it_warns_once(self, build): + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + build() + assert len(axis_warnings(rec)) == 1 + + def test_the_warning_names_the_removal_version(self): + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + DiffSettings(axis="time") + assert AXIS_REMOVAL_VERSION in str(axis_warnings(rec)[0].message) + + def test_it_is_a_futurewarning_not_a_deprecationwarning(self): + """DeprecationWarning is suppressed by default outside __main__, so a + pipeline -- library code -- would never see it. ezmsg core uses + FutureWarning for its own deprecations.""" + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + DiffSettings(axis="time") + assert rec[0].category is FutureWarning + + def test_it_blames_the_callers_line_not_ours_via_settings(self): + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + WindowSettings(axis="time", window_dur=0.1, window_shift=0.05) + expected_line = inspect.currentframe().f_lineno - 1 + (record,) = axis_warnings(rec) + assert (record.filename, record.lineno) == (__file__, expected_line) + + def test_it_blames_the_callers_line_not_ours_via_transformer(self): + """The depth differs from the path above -- the transformer builds its + settings through ``_unify_settings`` -- so a fixed stacklevel would be + wrong for at least one of the two.""" + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + WindowTransformer(axis="time", window_dur=0.1, window_shift=0.05) + expected_line = inspect.currentframe().f_lineno - 1 + (record,) = axis_warnings(rec) + assert (record.filename, record.lineno) == (__file__, expected_line) + + def test_the_factory_functions_blame_the_caller_too(self): + from ezmsg.sigproc.scaler import scaler_np + + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + scaler_np(time_constant=1.0, axis="time") + expected_line = inspect.currentframe().f_lineno - 1 + (record,) = axis_warnings(rec) + assert (record.filename, record.lineno) == (__file__, expected_line) + + +class TestLeavingItUnsetIsSilent: + @pytest.mark.parametrize( + "build", + [ + pytest.param(lambda: WindowTransformer(window_dur=0.1, window_shift=0.05), id="window"), + pytest.param(lambda: DiffTransformer(DiffSettings()), id="diff"), + pytest.param( + lambda: ButterworthFilterTransformer(ButterworthFilterSettings(order=2, cutoff=20.0)), + id="butterworth", + ), + pytest.param( + lambda: GaussianSmoothingFilterTransformer(GaussianSmoothingSettings(sigma=0.01)), + id="gaussian", + ), + pytest.param( + lambda: AdaptiveStandardScalerTransformer(AdaptiveStandardScalerSettings()), + id="scaler", + ), + ], + ) + def test_construction_and_processing_are_both_silent(self, build): + """Processing matters as much as construction: filter-by-design rebuilds a + child FilterSettings on every state reset.""" + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + proc = build() + proc(msg()) + proc(msg()) + assert not axis_warnings(rec) + + def test_merge_with_no_align_axis_is_silent(self): + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + MergeProcessor(MergeSettings(axis="ch")) + assert not axis_warnings(rec) + + +class TestInternalForwardingDoesNotMultiply: + def test_the_scaler_warns_once_for_its_two_child_ewmas(self): + """AdaptiveStandardScaler builds two EWMATransformers from its own axis, + inside ``_reset_state`` -- so without suppression this would warn twice + per reset, mid-stream, naming the pipeline driver.""" + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + proc = AdaptiveStandardScalerTransformer(AdaptiveStandardScalerSettings(axis="time")) + proc(msg()) + proc(msg()) + assert len(axis_warnings(rec)) == 1 + + def test_filter_by_design_warns_once_across_resets(self): + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + proc = ButterworthFilterTransformer(ButterworthFilterSettings(axis="time", order=2, cutoff=20.0)) + proc(msg()) + proc(msg(n_ch=5)) # a channel change forces a fresh reset + assert len(axis_warnings(rec)) == 1 + + def test_the_context_manager_silences_and_restores(self): + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + with suppress_axis_deprecation(): + DiffSettings(axis="time") + assert not axis_warnings(rec) + DiffSettings(axis="time") + assert len(axis_warnings(rec)) == 1 + + +class TestBehaviourIsUnchangedUntilRemoval: + """A deprecation window that changed behaviour would not be a window.""" + + def test_a_configured_axis_is_still_honoured(self): + transposed = AxisArray( + np.arange(24, dtype=float).reshape(3, 8), + dims=["ch", "time"], + axes={"time": AxisArray.TimeAxis(fs=FS)}, + key="dev", + chunk_dim="time", + ) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + # `ch` is not the chunk dim, and is exactly what removal will stop + # allowing -- but today it must still be obeyed. + proc = DiffTransformer(DiffSettings(axis="ch")) + out = proc(transposed) + assert out.data.shape == (3, 8) + # Differences along `ch` are 8 apart in this row-major fixture. + assert np.allclose(out.data[1:, :], 8.0) + + def test_the_runtime_mismatch_warning_still_fires_alongside(self, caplog): + """The construction-time warning says "delete this"; the runtime one says + "deleting this will change what the stage computes".""" + windowed = AxisArray( + np.zeros((4, 8, 3)), + dims=["win", "time", "ch"], + axes={"win": AxisArray.TimeAxis(fs=FS / 8), "time": AxisArray.TimeAxis(fs=FS)}, + key="dev", + chunk_dim="win", + ) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + proc = DiffTransformer(DiffSettings(axis="time")) + with caplog.at_level("WARNING"): + proc(windowed) + assert any("chunk_dim" in r.message for r in caplog.records) + + +class TestTheLegacyTimeDefault: + """The second wave defaulted to a hardcoded ``axis="time"``, not to a + positional guess. Flipping those to follow ``chunk_dim`` changes results + wherever the chunk dimension is not ``"time"`` -- most obviously downstream + of a windowing stage -- and no setting exists to warn about, because the + affected caller set nothing. ``legacy_default`` is what surfaces them.""" + + @staticmethod + def _windowed(): + """``(win, time, ch)``: the old default would pick ``time``, the new one + picks ``win``.""" + return AxisArray( + np.random.default_rng(0).standard_normal((4, 8, 2)), + dims=["win", "time", "ch"], + axes={"win": AxisArray.TimeAxis(fs=FS / 8), "time": AxisArray.TimeAxis(fs=FS)}, + key="dev", + chunk_dim="win", + ) + + def test_it_warns_when_the_resolved_dim_is_not_the_old_default(self, caplog): + from ezmsg.sigproc.util.message import resolve_configured_chunk_dim + + class Proc: + STREAMING_DIMS = ("time",) + + proc = Proc() + with caplog.at_level("WARNING"): + resolved = resolve_configured_chunk_dim(proc, self._windowed(), None, legacy_default="time") + assert resolved == "win" + assert any("used to operate on axis='time'" in r.message for r in caplog.records) + + def test_it_warns_only_once(self, caplog): + from ezmsg.sigproc.util.message import resolve_configured_chunk_dim + + class Proc: + STREAMING_DIMS = ("time",) + + proc = Proc() + with caplog.at_level("WARNING"): + for _ in range(3): + resolve_configured_chunk_dim(proc, self._windowed(), None, legacy_default="time") + assert sum("used to operate on" in r.message for r in caplog.records) == 1 + + def test_a_raw_stream_is_silent(self, caplog): + """The overwhelmingly common case: chunk_dim is already "time", so + nothing changed and there is nothing to say.""" + from ezmsg.sigproc.util.message import resolve_configured_chunk_dim + + class Proc: + STREAMING_DIMS = ("time",) + + with caplog.at_level("WARNING"): + resolved = resolve_configured_chunk_dim(Proc(), msg(), None, legacy_default="time") + assert resolved == "time" + assert not caplog.records + + def test_a_stream_without_the_old_default_dim_is_silent(self, caplog): + """If ``time`` is not even present, the old default could not have been + operating on it, so there is no behaviour change to report.""" + from ezmsg.sigproc.util.message import resolve_configured_chunk_dim + + class Proc: + STREAMING_DIMS = ("time",) + + no_time = AxisArray( + np.zeros((4, 2)), + dims=["win", "ch"], + axes={"win": AxisArray.TimeAxis(fs=FS)}, + key="dev", + chunk_dim="win", + ) + with caplog.at_level("WARNING"): + resolve_configured_chunk_dim(Proc(), no_time, None, legacy_default="time") + assert not caplog.records + + def test_stages_still_follow_the_declaration_end_to_end(self, caplog): + """RollingScaler is the cheapest of the seven to drive; the point is that + the resolved axis reaches the state, not the arithmetic.""" + from ezmsg.sigproc.rollingscaler import RollingScalerProcessor, RollingScalerSettings + + proc = RollingScalerProcessor(RollingScalerSettings(window_size=0.1)) + with caplog.at_level("WARNING"): + proc(self._windowed()) + assert proc.state.axis == "win" + + def test_passing_the_old_default_explicitly_preserves_behaviour(self): + """The escape hatch the warning points at: during the window, callers can + pin the old axis rather than accept the new resolution.""" + from ezmsg.sigproc.rollingscaler import RollingScalerProcessor, RollingScalerSettings + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + proc = RollingScalerProcessor(RollingScalerSettings(window_size=0.1, axis="time")) + proc(self._windowed()) + assert proc.state.axis == "time" + + +class TestTheSecondWaveWarnsOnExplicitUse: + """The seven stages whose default was a hardcoded ``"time"``. Their settings + now default to ``None`` like the rest, so an explicit ``axis=`` is what warns.""" + + @pytest.mark.parametrize( + "build", + [ + pytest.param(lambda a: _lnc(a), id="adaptive_lnc"), + pytest.param(lambda a: _lattice(a), id="adaptive_lattice_notch"), + pytest.param(lambda a: _binned(a), id="binned_aggregate"), + pytest.param(lambda a: _rolling(a), id="rollingscaler"), + pytest.param(lambda a: _resample(a), id="resample"), + pytest.param(lambda a: _cwt(a), id="wavelets"), + pytest.param(lambda a: _fbdesign(a), id="filterbankdesign"), + ], + ) + def test_explicit_warns_and_unset_is_silent(self, build): + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + build("time") + assert len(axis_warnings(rec)) == 1 + + with warnings.catch_warnings(record=True) as rec: + warnings.simplefilter("always") + build(None) + assert not axis_warnings(rec) + + @pytest.mark.parametrize("name", sorted(LEGACY_TIME_DEFAULT)) + def test_every_legacy_default_class_is_also_deprecated(self, name): + assert name in DEPRECATED_SETTINGS + + +def _lnc(axis): + from ezmsg.sigproc.adaptive_lnc import AdaptiveLNCSettings + + return AdaptiveLNCSettings(axis=axis) + + +def _lattice(axis): + from ezmsg.sigproc.adaptive_lattice_notch import AdaptiveLatticeNotchFilterSettings + + return AdaptiveLatticeNotchFilterSettings(axis=axis) + + +def _binned(axis): + from ezmsg.sigproc.binned_aggregate import BinnedAggregateSettings + + return BinnedAggregateSettings(axis=axis) + + +def _rolling(axis): + from ezmsg.sigproc.rollingscaler import RollingScalerSettings + + return RollingScalerSettings(axis=axis) + + +def _resample(axis): + from ezmsg.sigproc.resample import ResampleSettings + + return ResampleSettings(axis=axis) + + +def _cwt(axis): + from ezmsg.sigproc.wavelets import CWTSettings + + return CWTSettings(axis=axis, wavelet="morl", frequencies=[10.0, 20.0]) + + +def _fbdesign(axis): + from ezmsg.sigproc.filterbankdesign import FilterbankDesignSettings + + return FilterbankDesignSettings(filters=[], axis=axis) diff --git a/tests/unit/test_chunk_dim_resolution.py b/tests/unit/test_chunk_dim_resolution.py new file mode 100644 index 0000000..06897db --- /dev/null +++ b/tests/unit/test_chunk_dim_resolution.py @@ -0,0 +1,275 @@ +"""The axis a processor operates on, when the settings do not name one. + +Every stage used to guess positionally -- ``dims[0]`` for "the streaming axis", +``dims[-1]`` for "the channel axis". Both are positions rather than meanings, +and ``AxisArray.chunk_dim`` is the producer's declaration of which dimension +messages actually accumulate along. These tests pin the three resolution rules +(:mod:`ezmsg.sigproc.util.message`) and then check that the stages that carry +state between messages really follow the declaration. + +The fixtures are deliberately transposed or windowed, because that is the only +place the old guess and the new rule disagree. +""" + +import numpy as np +import pytest +from ezmsg.util.messages.axisarray import AxisArray, CoordinateAxis + +from ezmsg.sigproc.util.message import ( + resolve_chunk_dim, + resolve_configured_chunk_dim, + resolve_feature_dim, + resolve_transform_dim, +) + +FS = 100.0 + + +def _ch_axis(n): + return CoordinateAxis(data=np.array([f"ch{i}" for i in range(n)]), dims=["ch"]) + + +def transposed(n_ch=3, n_time=8, chunk_dim="time"): + """``(ch, time)``: the first dim is static, the accumulating one is second.""" + kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} + return AxisArray( + np.arange(n_ch * n_time, dtype=float).reshape(n_ch, n_time), + dims=["ch", "time"], + axes={"ch": _ch_axis(n_ch), "time": AxisArray.TimeAxis(fs=FS)}, + key="dev", + **kwargs, + ) + + +def windowed(n_win=4, n_time=8, n_ch=3): + """``(win, time, ch)``: ``win`` accumulates, ``time`` is within-window.""" + return AxisArray( + np.zeros((n_win, n_time, n_ch), dtype=float), + dims=["win", "time", "ch"], + axes={ + "win": AxisArray.TimeAxis(fs=FS / n_time), + "time": AxisArray.TimeAxis(fs=FS), + "ch": _ch_axis(n_ch), + }, + key="dev", + chunk_dim="win", + ) + + +def raw(n_time=8, n_ch=3, chunk_dim="time"): + kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} + return AxisArray( + np.zeros((n_time, n_ch), dtype=float), + dims=["time", "ch"], + axes={"time": AxisArray.TimeAxis(fs=FS), "ch": _ch_axis(n_ch)}, + key="dev", + **kwargs, + ) + + +class TestResolveChunkDim: + def test_the_declaration_wins_over_position(self): + msg = transposed() + assert msg.dims[0] != msg.chunk_dim, "the fixture must distinguish the two" + assert resolve_chunk_dim(msg) == "time" + + def test_it_follows_a_windowing_stage_onto_win(self): + assert resolve_chunk_dim(windowed()) == "win" + + def test_an_undeclared_chunk_dim_falls_back_to_streaming_dims(self): + assert resolve_chunk_dim(transposed(chunk_dim=None)) == "time" + + def test_the_streaming_dims_fallback_is_configurable(self): + msg = windowed() + object.__setattr__(msg, "chunk_dim", None) + assert resolve_chunk_dim(msg, ("win",)) == "win" + + def test_dims_zero_is_the_last_resort_only(self): + """Nothing declared and nothing recognised: the position is all there is.""" + msg = AxisArray(np.zeros((4, 2)), dims=["a", "b"], key="dev") + assert resolve_chunk_dim(msg) == "a" + + +class TestResolveFeatureDim: + def test_it_skips_the_chunk_dim_on_a_transposed_stream(self): + """``dims[-1]`` here is ``time``. A slicer defaulting to it would drop + samples, and an affine transform would matmul across time.""" + msg = transposed() + assert msg.dims[-1] == msg.chunk_dim, "the fixture must make the naive guess wrong" + assert resolve_feature_dim(msg) == "ch" + + def test_it_is_unchanged_on_a_conventional_stream(self): + assert resolve_feature_dim(raw()) == "ch" + + def test_position_zero_skips_the_chunk_dim_too(self): + """RangedAggregate's case: ``dims[0]`` is usually the chunk dim, which is + the worst possible default for an axis that must carry band values.""" + assert resolve_feature_dim(windowed(), 0) == "time" + + def test_a_chunk_only_message_falls_back_rather_than_raising(self): + msg = AxisArray(np.zeros(8), dims=["time"], axes={"time": AxisArray.TimeAxis(fs=FS)}, chunk_dim="time") + assert resolve_feature_dim(msg) == "time" + + +class TestResolveTransformDim: + def test_windowed_input_transforms_within_the_window(self): + """``win`` accumulates, but each window's spectrum is over ``time``.""" + assert resolve_transform_dim(windowed()) == "time" + + def test_raw_input_falls_through_to_the_chunk_dim(self): + """``ch`` carries a CoordinateAxis, not a LinearAxis, so it is not a + candidate and the rule lands back on ``time``.""" + assert resolve_transform_dim(raw()) == "time" + + def test_it_holds_under_transposition(self): + assert resolve_transform_dim(transposed()) == "time" + + +class TestResolveConfiguredChunkDim: + def test_an_explicit_axis_still_wins(self): + """The escape hatch stays open: an explicit axis is an instruction.""" + + class Proc: + STREAMING_DIMS = ("time",) + + assert resolve_configured_chunk_dim(Proc(), windowed(), "time") == "time" + + def test_a_disagreement_warns_once(self, caplog): + class Proc: + STREAMING_DIMS = ("time",) + + proc = Proc() + with caplog.at_level("WARNING"): + for _ in range(3): + resolve_configured_chunk_dim(proc, windowed(), "time") + assert sum("chunk_dim" in r.message for r in caplog.records) == 1 + + def test_agreement_is_silent(self, caplog): + class Proc: + STREAMING_DIMS = ("time",) + + with caplog.at_level("WARNING"): + resolve_configured_chunk_dim(Proc(), raw(), "time") + assert not caplog.records + + def test_a_mere_guess_never_warns(self, caplog): + """Warning against STREAMING_DIMS rather than a declaration would fire on + every correctly-configured windowed pipeline whose producer is silent.""" + + class Proc: + STREAMING_DIMS = ("time",) + + msg = windowed() + object.__setattr__(msg, "chunk_dim", None) + with caplog.at_level("WARNING"): + resolve_configured_chunk_dim(Proc(), msg, "win") + assert not caplog.records + + +class TestStagesFollowTheDeclaration: + """The point of the exercise: state carried between messages must be carried + along the dimension that actually grows.""" + + def test_window_buffers_along_the_declared_dim(self): + from ezmsg.sigproc.window import WindowSettings, WindowTransformer + + proc = WindowTransformer(WindowSettings(window_dur=0.04, window_shift=0.02)) + out = proc(transposed(n_ch=3, n_time=8)) + # 8 samples of a 4-sample window shifting by 2 -> windows along `win`, + # each holding 4 time samples and all 3 channels. + assert "win" in out.dims + assert out.data.shape[out.get_axis_idx("time")] == 4 + assert out.data.shape[out.get_axis_idx("ch")] == 3 + + def test_scaler_accumulates_along_the_declared_dim(self): + from ezmsg.sigproc.scaler import ( + AdaptiveStandardScalerSettings, + AdaptiveStandardScalerTransformer, + ) + + proc = AdaptiveStandardScalerTransformer(AdaptiveStandardScalerSettings(time_constant=1.0)) + out = proc(transposed(n_ch=3, n_time=8)) + assert out.data.shape == (3, 8) + assert out.dims == ["ch", "time"] + + def test_scaler_looks_up_the_axis_index_rather_than_assuming_zero(self): + """The old code hardcoded ``axis_idx = 0`` whenever ``axis`` was unset, + which silently transposed the data on a ``(ch, time)`` stream.""" + from ezmsg.sigproc.scaler import ( + RiverAdaptiveStandardScalerSettings, + RiverAdaptiveStandardScalerTransformer, + ) + + pytest.importorskip("river") + proc = RiverAdaptiveStandardScalerTransformer(RiverAdaptiveStandardScalerSettings(time_constant=1.0)) + proc(transposed(n_ch=3, n_time=8)) + assert proc.state.axis == "time" + assert proc.state.axis_idx == 1 + + def test_filter_carries_zi_along_the_declared_dim(self): + from ezmsg.sigproc.butterworthfilter import ( + ButterworthFilterSettings, + ButterworthFilterTransformer, + ) + + proc = ButterworthFilterTransformer(ButterworthFilterSettings(order=2, cuton=None, cutoff=20.0, coef_type="ba")) + first = proc(transposed(n_ch=3, n_time=8)) + second = proc(transposed(n_ch=3, n_time=8)) + assert first.data.shape == second.data.shape == (3, 8) + # zi is per-channel, so it has one entry per channel and the two chunks + # differ: continuity was carried across the message boundary. + assert not np.allclose(first.data, second.data) + + def test_diff_carries_the_previous_sample_along_the_declared_dim(self): + from ezmsg.sigproc.diff import DiffSettings, DiffTransformer + + proc = DiffTransformer(DiffSettings()) + out = proc(transposed(n_ch=3, n_time=8)) + assert out.data.shape == (3, 8), "diff prepends the carried sample, preserving length" + # Row-major arange over (3, 8): consecutive samples along `time` differ + # by 1, so every within-row difference is 1. + assert np.allclose(out.data[:, 1:], 1.0) + + +class TestSpectrumPicksTheTransformAxis: + def test_windowed_input_transforms_time_not_win(self): + """Previously ``dims[0]`` picked ``win`` here, FFT-ing across windows.""" + from ezmsg.sigproc.spectrum import SpectrumSettings, SpectrumTransformer + + out = SpectrumTransformer(SpectrumSettings())(windowed(n_win=4, n_time=8, n_ch=3)) + assert "freq" in out.dims + assert out.dims.index("freq") == 1, "the freq axis replaces `time`, not `win`" + assert out.data.shape[0] == 4, "the win dimension survives" + + def test_windowed_output_keeps_accumulating_along_win(self): + from ezmsg.sigproc.spectrum import SpectrumSettings, SpectrumTransformer + + out = SpectrumTransformer(SpectrumSettings())(windowed()) + assert out.chunk_dim == "win" + + def test_raw_input_still_transforms_time(self): + from ezmsg.sigproc.spectrum import SpectrumSettings, SpectrumTransformer + + out = SpectrumTransformer(SpectrumSettings())(raw(n_time=16, n_ch=3)) + assert "freq" in out.dims + assert "time" not in out.dims + assert out.chunk_dim is None, "the transformed axis is consumed" + + +class TestFeatureStagesSkipTheChunkDim: + def test_slicer_defaults_to_channels_on_a_transposed_stream(self): + from ezmsg.sigproc.slicer import SlicerSettings, SlicerTransformer + + out = SlicerTransformer(SlicerSettings(selection="0:2"))(transposed(n_ch=3, n_time=8)) + assert out.data.shape == (2, 8), "channels sliced, samples untouched" + + def test_affine_transform_defaults_to_channels_on_a_transposed_stream(self): + from ezmsg.sigproc.affinetransform import ( + AffineTransformSettings, + AffineTransformTransformer, + ) + + weights = np.eye(3) * 2.0 + out = AffineTransformTransformer(AffineTransformSettings(weights=weights))(transposed(n_ch=3, n_time=8)) + assert out.data.shape == (3, 8) + assert np.allclose(out.data, transposed(n_ch=3, n_time=8).data * 2.0)