refactor(backend): centralize per-band Butterworth filtering in BatchedBandpass

apps/backend/analyze_detection.py

@@ -1,5 +1,6 @@
 """Detection analyzers — effects, acid, reverb, vocal, supersaw, and genre."""
 
+import functools
 import sys
 
 import numpy as np
@@ -21,6 +22,7 @@
 
 from dsp_utils import _safe_db, _compute_bark_db
 from analyze_audio_io import _load_stem_mono
+from dsp_bandbank import BatchedBandpass
 
 
 # Phase 1.D #5 — bands for per-band RT60 estimation. Chosen to roughly mirror
@@ -35,20 +37,21 @@
 )
 
 
+@functools.lru_cache(maxsize=4)
+def _bandbank_for(sample_rate: int) -> BatchedBandpass:
+    """One BatchedBandpass per sample rate (4 slots covers 22050/44100/48000/96000)."""
+    return BatchedBandpass(int(sample_rate))
+
+
 def _bandpass_signal(mono: np.ndarray, sample_rate: int, lo_hz: float, hi_hz: float) -> np.ndarray | None:
-    """4th-order Butterworth bandpass via scipy.signal. Returns None on failure."""
-    if scipy_signal is None or mono.size == 0:
-        return None
-    nyquist = 0.5 * sample_rate
-    lo = max(1.0, lo_hz) / nyquist
-    hi = min(sample_rate * 0.49, hi_hz) / nyquist
-    if not (0.0 < lo < hi < 1.0):
-        return None
-    try:
-        sos = scipy_signal.butter(4, [lo, hi], btype="bandpass", output="sos")
-        return scipy_signal.sosfiltfilt(sos, mono).astype(np.float32, copy=False)
-    except Exception:
-        return None
+    """4th-order Butterworth bandpass via scipy.signal. Returns None on failure.
+
+    Thin wrapper around ``BatchedBandpass.filter_one`` kept for callers that
+    pass ``sample_rate`` per call. Output is bit-identical to the previous
+    inline implementation; new code should obtain a ``BatchedBandpass``
+    directly via ``_bandbank_for(sample_rate)`` and use ``filter_many``.
+    """
+    return _bandbank_for(sample_rate).filter_one(mono, lo_hz, hi_hz)
 
 
 # Mirrors apps/backend/analyze_core.py SPECTRAL_BALANCE_BANDS — imported lazily
@@ -545,13 +548,16 @@ def analyze_reverb_detail(
 
         # Per-band RT60: re-use the SAME transient indices on the broadband
         # envelope so each band measures the same events. Bandpass the raw
-        # signal first, recompute the per-band envelope, then run the slope
-        # fit. Requires scipy.signal — if unavailable, omit per-band.
+        # signal once (filter_many designs each SOS only once and reuses
+        # them), recompute the per-band envelope, then run the slope fit.
+        # Requires scipy.signal — if unavailable, omit per-band.
         per_band_rt60: dict[str, float] | None = None
         if scipy_signal is not None:
+            bandbank = _bandbank_for(sample_rate)
+            filtered_bands = bandbank.filter_many(mono_arr, _REVERB_BANDS, dtype=np.float32)
             per_band: dict[str, float] = {}
-            for band_name, lo_hz, hi_hz in _REVERB_BANDS:
-                filtered = _bandpass_signal(mono_arr, sample_rate, lo_hz, hi_hz)
+            for band_name, _lo, _hi in _REVERB_BANDS:
+                filtered = filtered_bands.get(band_name)
                 if filtered is None:
                     continue
                 band_envelope = np.zeros(n_frames, dtype=np.float64)

apps/backend/dsp_bandbank.py

@@ -0,0 +1,147 @@
+"""BatchedBandpass — 4th-order Butterworth bandpass bank with zero-phase filtfilt.
+
+Drop-in replacement for the inline ``butter(4, ..., output='sos') + sosfiltfilt``
+pattern that ASA's detection module uses for per-band RT60. The pattern was
+repeating across multiple call sites with slightly different glue; this module
+gives them one shared primitive.
+
+Output is bit-identical to the inline code (np.allclose atol=1e-12, rtol=1e-12).
+The ``backend`` argument is reserved for a future torch path — the audit at
+``docs/history/library-review-torchfx-2026-05-13.md`` flagged that as a
+candidate if profiling ever shows per-band scipy filtering is hot.
+"""
+
+from collections.abc import Iterable, Mapping
+from typing import Union
+
+import numpy as np
+
+try:
+    from scipy import signal as scipy_signal  # type: ignore[import-not-found]
+except ImportError:
+    scipy_signal = None  # type: ignore[assignment]
+
+
+class BatchedBandpass:
+    """Per-instance bandpass bank with cached SOS coefficients.
+
+    Matches ``analyze_detection._bandpass_signal`` byte-for-byte: Nyquist
+    clamps via ``max(1.0, lo_hz)`` and ``min(sample_rate * 0.49, hi_hz)``,
+    4th-order Butterworth via ``output='sos'``, then ``sosfiltfilt``. Any
+    deviation is a Phase 1 regression — see ``BatchedBandpassTests``.
+    """
+
+    def __init__(self, sample_rate: int, *, order: int = 4, backend: str = "scipy") -> None:
+        if backend != "scipy":
+            raise ValueError(
+                f"BatchedBandpass: backend={backend!r} not supported (PR 1 ships scipy only)"
+            )
+        self.sample_rate = int(sample_rate)
+        self.order = int(order)
+        self.backend = backend
+        # Memoised SOS coefficients per (lo_hz, hi_hz). sample_rate and order
+        # are fixed per instance. Failed designs (out-of-range etc.) are
+        # cached as None so the second caller short-circuits without retry.
+        self._sos_cache: dict[tuple[float, float], np.ndarray | None] = {}
+
+    def _design(self, lo_hz: float, hi_hz: float) -> np.ndarray | None:
+        key = (float(lo_hz), float(hi_hz))
+        if key in self._sos_cache:
+            return self._sos_cache[key]
+        if scipy_signal is None:
+            self._sos_cache[key] = None
+            return None
+        nyquist = 0.5 * self.sample_rate
+        lo = max(1.0, lo_hz) / nyquist
+        hi = min(self.sample_rate * 0.49, hi_hz) / nyquist
+        if not (0.0 < lo < hi < 1.0):
+            self._sos_cache[key] = None
+            return None
+        try:
+            sos = scipy_signal.butter(self.order, [lo, hi], btype="bandpass", output="sos")
+        except Exception:
+            sos = None
+        self._sos_cache[key] = sos
+        return sos
+
+    # Note on asymmetric dtype defaults between filter_one and filter_many:
+    # filter_one defaults to float32 because it replaces the pre-refactor
+    # _bandpass_signal which hardcoded float32 — that default keeps the
+    # existing call site bit-identical with no code change. filter_many
+    # defaults to float64 because its primary future caller
+    # (analyze_per_band_transient_density) does the upstream mono → float64
+    # cast in the outer scope and feeds the result to librosa, which accepts
+    # either dtype. Both methods accept an explicit dtype keyword so callers
+    # can override either default at the call site.
+
+    def filter_one(
+        self,
+        mono: np.ndarray,
+        lo_hz: float,
+        hi_hz: float,
+        *,
+        dtype=np.float32,
+    ) -> np.ndarray | None:
+        """Single-band filter; use ``filter_many`` for the multi-band path.
+
+        Default ``dtype=np.float32`` preserves bit-identicality with the
+        pre-refactor ``analyze_detection._bandpass_signal`` — callers that
+        pass no dtype get exactly the historical output. See the class-level
+        note above on why ``filter_one`` and ``filter_many`` differ in
+        default dtype.
+
+        Returns ``None`` on empty input, missing scipy, Nyquist-clamp
+        failure, or sosfiltfilt error.
+        """
+        if scipy_signal is None or mono is None or getattr(mono, "size", 0) == 0:
+            return None
+        sos = self._design(lo_hz, hi_hz)
+        if sos is None:
+            return None
+        try:
+            return scipy_signal.sosfiltfilt(sos, mono).astype(dtype, copy=False)
+        except Exception:
+            return None
+
+    def filter_many(
+        self,
+        mono: np.ndarray,
+        bands: Union[Mapping[str, tuple[float, float]], Iterable[tuple[str, float, float]]],
+        *,
+        dtype=np.float64,
+    ) -> dict[str, np.ndarray]:
+        """Run the bandpass over many bands and return ``{band_name: filtered}``.
+
+        Accepts two shapes for ``bands``:
+          - ``Mapping[str, (lo, hi)]`` — matches ``SPECTRAL_BALANCE_BANDS``
+            in ``analyze_core``.
+          - ``Iterable[(name, lo, hi)]`` — matches ``_REVERB_BANDS`` in
+            ``analyze_detection``.
+
+        Skipped bands (empty input, out-of-range, sosfiltfilt error) are
+        absent from the returned dict — preserves the omit-on-skip behavior
+        of the per-band reverb loop today. See the class-level note above
+        on why ``filter_many`` defaults to ``float64`` while ``filter_one``
+        defaults to ``float32``.
+        """
+        if scipy_signal is None or mono is None or getattr(mono, "size", 0) == 0:
+            return {}
+
+        if isinstance(bands, Mapping):
+            entries: Iterable[tuple[str, float, float]] = [
+                (str(name), float(lo), float(hi)) for name, (lo, hi) in bands.items()
+            ]
+        else:
+            entries = [(str(name), float(lo), float(hi)) for name, lo, hi in bands]
+
+        out: dict[str, np.ndarray] = {}
+        for name, lo_hz, hi_hz in entries:
+            sos = self._design(lo_hz, hi_hz)
+            if sos is None:
+                continue
+            try:
+                filtered = scipy_signal.sosfiltfilt(sos, mono)
+            except Exception:
+                continue
+            out[name] = filtered.astype(dtype, copy=False)
+        return out

apps/backend/tests/test_dsp_utils.py

@@ -27,6 +27,14 @@
 _DSP_SPEC.loader.exec_module(dsp_utils)
 
 
+_BANDBANK_PATH = _BACKEND_ROOT / "dsp_bandbank.py"
+_BANDBANK_SPEC = importlib.util.spec_from_file_location("dsp_bandbank_test", _BANDBANK_PATH)
+if _BANDBANK_SPEC is None or _BANDBANK_SPEC.loader is None:
+    raise AssertionError("Could not load dsp_bandbank.py for direct helper tests.")
+dsp_bandbank = importlib.util.module_from_spec(_BANDBANK_SPEC)
+_BANDBANK_SPEC.loader.exec_module(dsp_bandbank)
+
+
 class PearsonCorrTests(unittest.TestCase):
     """Sanity-check the in-house Pearson correlation used across the analyzers."""
 
@@ -274,5 +282,128 @@ def test_silent_sub_band_produces_none_sub(self):
             self.assertIsNone(point["sub"])
 
 
+class BatchedBandpassTests(unittest.TestCase):
+    """Verify ``BatchedBandpass`` is bit-identical to the inline scipy code
+    it replaces in ``analyze_detection.py``.
+
+    The load-bearing assertion is ``test_filter_one_matches_inline_scipy_bit_for_bit``:
+    if the new class deviates from ``butter(4, [lo/nyq, hi/nyq], output='sos')
+    + sosfiltfilt`` by more than 1e-12 (well below float32 last-bit precision
+    after the post-filter cast), Phase 1 numbers move and the refactor must
+    not land.
+    """
+
+    # _REVERB_BANDS, replicated here so the test doesn't drag in analyze_detection
+    # (which pulls essentia, librosa, etc.).
+    _REVERB_BANDS = (
+        ("low", 20.0, 250.0),
+        ("lowMids", 250.0, 2000.0),
+        ("highMids", 2000.0, 8000.0),
+        ("highs", 8000.0, 16000.0),
+    )
+    _SR = 44100
+
+    @classmethod
+    def setUpClass(cls):
+        # Fixed-seed signal so the parity assertion is reproducible.
+        cls._signal = np.random.default_rng(1234).standard_normal(2 * cls._SR)
+
+    def _inline_reference(self, mono, lo_hz, hi_hz):
+        """Literal pre-refactor implementation, copy-pasted from
+        ``analyze_detection._bandpass_signal``. Anything the new class
+        produces must match this byte-for-byte."""
+        from scipy import signal as scipy_signal
+
+        nyquist = 0.5 * self._SR
+        lo = max(1.0, lo_hz) / nyquist
+        hi = min(self._SR * 0.49, hi_hz) / nyquist
+        if not (0.0 < lo < hi < 1.0):
+            return None
+        sos = scipy_signal.butter(4, [lo, hi], btype="bandpass", output="sos")
+        return scipy_signal.sosfiltfilt(sos, mono).astype(np.float32, copy=False)
+
+    def test_filter_one_matches_inline_scipy_bit_for_bit(self):
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        for name, lo, hi in self._REVERB_BANDS:
+            with self.subTest(band=name):
+                expected = self._inline_reference(self._signal, lo, hi)
+                got = bb.filter_one(self._signal, lo, hi)
+                self.assertIsNotNone(expected)
+                self.assertIsNotNone(got)
+                np.testing.assert_allclose(got, expected, atol=1e-12, rtol=1e-12)
+
+    def test_filter_many_returns_same_arrays_as_per_band_loop(self):
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        many = bb.filter_many(self._signal, self._REVERB_BANDS, dtype=np.float32)
+        self.assertEqual(set(many.keys()), {n for n, *_ in self._REVERB_BANDS})
+        for name, lo, hi in self._REVERB_BANDS:
+            with self.subTest(band=name):
+                expected = self._inline_reference(self._signal, lo, hi)
+                np.testing.assert_allclose(many[name], expected, atol=1e-12, rtol=1e-12)
+
+    def test_filter_many_accepts_reverb_bands_tuple_shape(self):
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        result = bb.filter_many(self._signal, self._REVERB_BANDS, dtype=np.float32)
+        self.assertEqual(set(result.keys()), {"low", "lowMids", "highMids", "highs"})
+
+    def test_filter_many_accepts_spectral_balance_dict_shape(self):
+        """SPECTRAL_BALANCE_BANDS in analyze_core is a ``Mapping[str, (lo, hi)]`` —
+        ``filter_many`` must accept that shape too, not just the reverb triple."""
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        bands = {"subBass": (20, 80), "lowMids": (250, 500)}
+        result = bb.filter_many(self._signal, bands)
+        self.assertEqual(set(result.keys()), {"subBass", "lowMids"})
+        # Default dtype is float64 to match the upstream cast in
+        # analyze_per_band_transient_density.
+        self.assertEqual(result["subBass"].dtype, np.float64)
+
+    def test_invalid_band_returns_none_or_absent_key(self):
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        # lo > hi forces the clamp predicate to fail.
+        self.assertIsNone(bb.filter_one(self._signal, 10000.0, 100.0))
+        many = bb.filter_many(
+            self._signal,
+            [("bad", 10000.0, 100.0), ("good", 100.0, 1000.0)],
+            dtype=np.float32,
+        )
+        self.assertNotIn("bad", many)
+        self.assertIn("good", many)
+
+    def test_filter_one_returns_none_for_empty_input(self):
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        self.assertIsNone(bb.filter_one(np.array([], dtype=np.float64), 100.0, 1000.0))
+
+    def test_sos_coefficients_are_cached_per_band(self):
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        bb.filter_one(self._signal, 100.0, 1000.0)
+        self.assertIn((100.0, 1000.0), bb._sos_cache)
+        sos_first = bb._sos_cache[(100.0, 1000.0)]
+        bb.filter_one(self._signal, 100.0, 1000.0)
+        sos_second = bb._sos_cache[(100.0, 1000.0)]
+        # Identity check — second call must hit the cache, not re-design.
+        self.assertIs(sos_first, sos_second)
+
+    def test_unsupported_backend_raises(self):
+        with self.assertRaises(ValueError):
+            dsp_bandbank.BatchedBandpass(self._SR, backend="torch")
+
+    def test_filter_one_default_dtype_is_float32(self):
+        """Locks the bit-identicality contract: ``filter_one`` with no dtype
+        kwarg must return float32 to match the pre-refactor _bandpass_signal."""
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        out = bb.filter_one(self._signal, 100.0, 1000.0)
+        self.assertIsNotNone(out)
+        self.assertEqual(out.dtype, np.float32)
+
+    def test_filter_one_respects_dtype_override(self):
+        """Explicit ``dtype=np.float64`` must produce a float64 array — the
+        forward-looking override path for transient-density-style callers
+        that want to skip an upstream cast."""
+        bb = dsp_bandbank.BatchedBandpass(self._SR)
+        out = bb.filter_one(self._signal, 100.0, 1000.0, dtype=np.float64)
+        self.assertIsNotNone(out)
+        self.assertEqual(out.dtype, np.float64)
+
+
 if __name__ == "__main__":
     unittest.main()