Add passthrough setting to EWMASettings for identity bypass

src/ezmsg/sigproc/ewma.py

@@ -169,6 +169,12 @@ class EWMASettings(ez.Settings):
     the current EWMA estimate without updating state (useful for inference
     periods where you don't want to adapt statistics)."""
 
+    passthrough: bool = False
+    """If True, return the input unchanged (identity) without touching the
+    EWMA. Unlike a very large time_constant -- which still applies a (stale)
+    baseline estimate -- passthrough leaves the data untouched. May be toggled
+    at runtime without resetting the filter state."""
+
 
 @processor_state
 class EWMAState:
@@ -177,17 +183,18 @@ class EWMAState:
 
 
 class EWMATransformer(BaseStatefulTransformer[EWMASettings, AxisArray, AxisArray, EWMAState]):
-    # `accumulate` is read live in `_process` to gate state updates; other
-    # fields are cached into state (alpha, zi) during `_reset_state`.
-    NONRESET_SETTINGS_FIELDS = frozenset({"accumulate"})
+    # `accumulate` is read live in `_process` to gate state updates and
+    # `passthrough` is read live in `__call__`/`__acall__`; other fields are
+    # cached into state (alpha, zi) during `_reset_state`.
+    NONRESET_SETTINGS_FIELDS = frozenset({"accumulate", "passthrough"})
 
     def __call__(self, message: AxisArray) -> AxisArray:
-        if np.prod(message.data.shape) == 0:
+        if self.settings.passthrough or np.prod(message.data.shape) == 0:
             return message
         return super().__call__(message)
 
     async def __acall__(self, message: AxisArray) -> AxisArray:
-        if np.prod(message.data.shape) == 0:
+        if self.settings.passthrough or np.prod(message.data.shape) == 0:
             return message
         return await super().__acall__(message)
 

src/ezmsg/sigproc/scaler.py

@@ -104,9 +104,20 @@ class AdaptiveStandardScalerTransformer(
     ]
 ):
     # `accumulate` can be live-propagated into the child EWMAs (see
-    # `update_settings` below); `time_constant` and `axis` are baked into
+    # `update_settings` below) and `passthrough` is read live in
+    # `__call__`/`__acall__`; `time_constant` and `axis` are baked into
     # the children during `_reset_state`.
-    NONRESET_SETTINGS_FIELDS = frozenset({"accumulate"})
+    NONRESET_SETTINGS_FIELDS = frozenset({"accumulate", "passthrough"})
+
+    def __call__(self, message: AxisArray) -> AxisArray:
+        if self.settings.passthrough:
+            return message
+        return super().__call__(message)
+
+    async def __acall__(self, message: AxisArray) -> AxisArray:
+        if self.settings.passthrough:
+            return message
+        return await super().__acall__(message)
 
     def update_settings(self, new_settings: AdaptiveStandardScalerSettings) -> None:
         # Propagate accumulate into the existing child EWMAs before deferring

tests/unit/test_detrend.py

@@ -403,3 +403,39 @@ def test_detrend_accumulate_toggle():
     _ = proc(_make_msg(np.ones((10, 2)) * 100.0))
     zi_after_resume = proc._state.zi.copy()
     assert not np.allclose(zi_after_frozen, zi_after_resume)
+
+
+# --- Passthrough flag ---
+
+
+def test_detrend_passthrough_identity():
+    """passthrough=True returns the input unchanged — no mean subtraction."""
+    proc = DetrendTransformer(settings=EWMASettings(time_constant=0.1, passthrough=True))
+
+    msg = _make_msg(np.arange(20, dtype=float).reshape(10, 2) + 50.0)
+    out = proc(msg)
+
+    assert out is msg
+    assert proc._state.zi is None
+
+
+def test_detrend_passthrough_toggle_preserves_state():
+    """Toggling passthrough does not reset the EWMA baseline state."""
+    proc = DetrendTransformer(settings=EWMASettings(time_constant=0.1, accumulate=True))
+
+    # Initialize baseline
+    _ = proc(_make_msg(np.ones((50, 2)) * 10.0))
+    zi_before = proc._state.zi.copy()
+
+    # Passthrough: identity output, state untouched even with wild input.
+    proc.settings = dc_replace(proc.settings, passthrough=True)
+    msg = _make_msg(np.ones((10, 2)) * 1000.0)
+    out = proc(msg)
+    assert out is msg
+    assert np.allclose(proc._state.zi, zi_before)
+
+    # Resume detrending: baseline picks up where it left off.
+    proc.settings = dc_replace(proc.settings, passthrough=False)
+    out2 = proc(_make_msg(np.ones((10, 2)) * 10.0))
+    # Input near the preserved baseline of ~10 → detrended output near 0.
+    assert np.all(np.abs(out2.data) < 1.0)

tests/unit/test_ewma.py

@@ -269,6 +269,44 @@ def test_ewma_empty_first():
     check_state_not_corrupted(proc, normal)
 
 
+def test_ewma_passthrough_identity():
+    """passthrough=True returns the input message unchanged."""
+    ewma = EWMATransformer(settings=EWMASettings(time_constant=0.1, passthrough=True))
+
+    msg = _make_ewma_test_msg(np.arange(20, dtype=float).reshape(10, 2))
+    out = ewma(msg)
+
+    assert out is msg
+    # No state was ever initialized — the EWMA was skipped entirely.
+    assert ewma._state.zi is None
+
+
+def test_ewma_passthrough_toggle_preserves_state():
+    """Toggling passthrough does not reset state; processing resumes from prior zi."""
+    proc_toggled = EWMATransformer(settings=EWMASettings(time_constant=0.1))
+    proc_ref = EWMATransformer(settings=EWMASettings(time_constant=0.1))
+
+    msg1 = _make_ewma_test_msg(np.ones((10, 2)))
+    _ = proc_toggled(msg1)
+    _ = proc_ref(msg1)
+    zi_before = proc_toggled._state.zi.copy()
+
+    # Passthrough: input returned as-is, state untouched.
+    proc_toggled.settings = dc_replace(proc_toggled.settings, passthrough=True)
+    msg2 = _make_ewma_test_msg(np.ones((10, 2)) * 100.0)
+    out2 = proc_toggled(msg2)
+    assert out2 is msg2
+    assert np.allclose(proc_toggled._state.zi, zi_before)
+
+    # Resume: continues from the preserved state, matching the reference
+    # that never went through passthrough.
+    proc_toggled.settings = dc_replace(proc_toggled.settings, passthrough=False)
+    msg3 = _make_ewma_test_msg(np.ones((10, 2)) * 2.0)
+    out_toggled = proc_toggled(msg3)
+    out_ref = proc_ref(msg3)
+    assert np.allclose(out_toggled.data, out_ref.data)
+
+
 class TestEWMAUpdateSettings:
     """Live settings updates via BaseProcessor.update_settings."""
 
@@ -290,6 +328,21 @@ def test_accumulate_toggle_preserves_state(self):
         _ = proc(_make_ewma_test_msg(np.ones((10, 2)) * 100.0))
         assert np.allclose(proc._state.zi, zi_before)
 
+    def test_passthrough_toggle_preserves_state(self):
+        proc = EWMATransformer(settings=EWMASettings(time_constant=0.1))
+
+        _ = proc(_make_ewma_test_msg(np.ones((10, 2))))
+        zi_before = proc._state.zi.copy()
+
+        # `passthrough` is in NONRESET_SETTINGS_FIELDS — no reset queued.
+        proc.update_settings(EWMASettings(time_constant=0.1, passthrough=True))
+        assert proc._hash != -1
+        assert np.allclose(proc._state.zi, zi_before)
+
+        msg = _make_ewma_test_msg(np.ones((10, 2)) * 100.0)
+        assert proc(msg) is msg
+        assert np.allclose(proc._state.zi, zi_before)
+
     def test_time_constant_change_recomputes_alpha(self):
         proc = EWMATransformer(settings=EWMASettings(time_constant=0.1, accumulate=True))
 

tests/unit/test_scaler.py

@@ -316,6 +316,46 @@ def test_initial_accumulate_false(self):
         assert scaler._state.vars_sq_ewma.settings.accumulate is False
 
 
+class TestAdaptiveStandardScalerPassthrough:
+    """Tests for the passthrough setting on AdaptiveStandardScalerTransformer."""
+
+    def test_passthrough_identity(self):
+        """passthrough=True returns the input unchanged — no scaling."""
+        scaler = AdaptiveStandardScalerTransformer(
+            settings=AdaptiveStandardScalerSettings(time_constant=0.1, passthrough=True)
+        )
+
+        msg = _make_scaler_test_msg(np.arange(20, dtype=float).reshape(10, 2) + 50.0)
+        out = scaler(msg)
+
+        assert out is msg
+        # No state was ever initialized — the scaler was skipped entirely.
+        assert scaler._state.samps_ewma is None
+
+    def test_passthrough_toggle_preserves_state(self):
+        """Toggling passthrough does not reset the child EWMA states."""
+        scaler = AdaptiveStandardScalerTransformer(settings=AdaptiveStandardScalerSettings(time_constant=0.1))
+
+        np.random.seed(42)
+        _ = scaler(_make_scaler_test_msg(np.random.randn(100, 4)))
+        zi_samps = scaler._state.samps_ewma._state.zi.copy()
+        zi_vars = scaler._state.vars_sq_ewma._state.zi.copy()
+
+        # Passthrough: identity output, state untouched even with wild input.
+        scaler.update_settings(AdaptiveStandardScalerSettings(time_constant=0.1, passthrough=True))
+        assert scaler._hash != -1
+        msg = _make_scaler_test_msg(np.random.randn(50, 4) + 1000.0)
+        assert scaler(msg) is msg
+        assert np.allclose(scaler._state.samps_ewma._state.zi, zi_samps)
+        assert np.allclose(scaler._state.vars_sq_ewma._state.zi, zi_vars)
+
+        # Resume scaling: statistics pick up where they left off.
+        scaler.update_settings(AdaptiveStandardScalerSettings(time_constant=0.1, passthrough=False))
+        out = scaler(_make_scaler_test_msg(np.random.randn(50, 4)))
+        assert out.data.shape == (50, 4)
+        assert not np.any(np.isnan(out.data))
+
+
 class TestAdaptiveStandardScalerUpdateSettings:
     """Live settings updates via update_settings."""