feat(eeg): add non-mutating bandpass_filter (default 1-40 Hz)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/pipelines/eeg_pipeline.py

@@ -12,6 +12,7 @@ a logged WARNING), determinism (seeded ICA + sklearn RNG), traceability
 """
 from __future__ import annotations
 
+import mne
 import numpy as np
 
 from src.core.logger import get_logger
@@ -37,3 +38,27 @@ def is_valid_epoch(epoch: np.ndarray | None) -> bool:
     if not np.all(np.isfinite(epoch)):
         return False
     return True
+
+
+def bandpass_filter(
+    raw: mne.io.BaseRaw,
+    l_freq: float = 1.0,
+    h_freq: float = 40.0,
+) -> mne.io.BaseRaw:
+    """Apply a non-mutating bandpass filter to an MNE Raw.
+
+    Default 1-40 Hz removes drift below 1 Hz and high-frequency noise / line
+    artifacts above 40 Hz. Returns a copy; the input `raw` is unchanged.
+
+    Args:
+        raw: Loaded `mne.io.BaseRaw` (call `.load_data()` first if from disk).
+        l_freq: Low-cut frequency in Hz.
+        h_freq: High-cut frequency in Hz.
+
+    Returns:
+        A filtered copy of `raw`.
+    """
+    out = raw.copy()
+    out.filter(l_freq=l_freq, h_freq=h_freq, picks="all", verbose="ERROR")
+    logger.info("Bandpass filter applied: %.1f-%.1f Hz", l_freq, h_freq)
+    return out

tests/pipelines/test_eeg_pipeline.py

@@ -3,10 +3,14 @@ from __future__ import annotations
 
 from pathlib import Path
 
+import mne
 import numpy as np
 import pytest
 
-from src.pipelines.eeg_pipeline import is_valid_epoch
+from src.pipelines.eeg_pipeline import (
+    bandpass_filter,
+    is_valid_epoch,
+)
 
 
 FIXTURE = Path(__file__).parent.parent / "fixtures" / "eeg_sample.fif"
@@ -45,3 +49,35 @@ class TestIsValidEpoch:
         """String / object dtype arrays must be rejected without raising."""
         epoch = np.array([["a", "b"], ["c", "d"]])
         assert is_valid_epoch(epoch) is False
+
+
+class TestBandpassFilter:
+    def _load(self) -> mne.io.BaseRaw:
+        return mne.io.read_raw_fif(FIXTURE, preload=True, verbose="ERROR")
+
+    def test_returns_raw_instance(self) -> None:
+        raw = self._load()
+        out = bandpass_filter(raw, l_freq=1.0, h_freq=40.0)
+        assert isinstance(out, mne.io.BaseRaw)
+
+    def test_preserves_shape(self) -> None:
+        raw = self._load()
+        n_ch_before, n_t_before = raw.get_data().shape
+        out = bandpass_filter(raw, l_freq=1.0, h_freq=40.0)
+        assert out.get_data().shape == (n_ch_before, n_t_before)
+
+    def test_attenuates_dc_component(self) -> None:
+        """A bandpass with l_freq=1.0 must remove a DC offset."""
+        raw = self._load()
+        # Inject a large DC offset on every channel.
+        data = raw.get_data() + 1e-3
+        raw_dc = mne.io.RawArray(data, raw.info, verbose="ERROR")
+        out = bandpass_filter(raw_dc, l_freq=1.0, h_freq=40.0)
+        # Mean on each channel should be near zero (much smaller than 1e-3).
+        assert np.all(np.abs(out.get_data().mean(axis=1)) < 1e-4)
+
+    def test_does_not_mutate_input(self) -> None:
+        raw = self._load()
+        original_mean = raw.get_data().mean()
+        _ = bandpass_filter(raw, l_freq=1.0, h_freq=40.0)
+        assert raw.get_data().mean() == pytest.approx(original_mean, rel=1e-12)