feat(eeg): add run_pipeline orchestrator + CLI (FIF/EDF → Parquet)

src/pipelines/eeg_pipeline.py

@@ -12,6 +12,8 @@ a logged WARNING), determinism (seeded ICA + sklearn RNG), traceability
 """
 from __future__ import annotations
 
+from pathlib import Path
+
 import mne
 import numpy as np
 import pandas as pd
@@ -30,6 +32,11 @@ logger = get_logger(__name__)
 _EOG_CORR_THRESHOLD: float = 0.9
 
 
+# Default I/O paths for the EEG pipeline. Override via run_pipeline() args.
+DEFAULT_INPUT = Path("data/raw/eeg.fif")
+DEFAULT_OUTPUT = Path("data/processed/eeg_features.parquet")
+
+
 def is_valid_epoch(epoch: np.ndarray | None) -> bool:
     """Return True iff `epoch` is a non-empty 2-D numeric array with no NaN/inf.
 
@@ -390,3 +397,77 @@ def extract_features_from_recording(
         100.0 * n_dropped / max(n_total_epochs, 1),
     )
     return out
+
+
+def run_pipeline(
+    input_path: Path = DEFAULT_INPUT,
+    output_path: Path = DEFAULT_OUTPUT,
+    epoch_duration_s: float = 2.0,
+    eog_ch_name: str | None = None,
+    n_components: int = 15,
+    random_state: int = 97,
+) -> None:
+    """Run the EEG pipeline end-to-end: raw FIF/EDF -> processed feature Parquet.
+
+    Reads `input_path` via MNE, applies bandpass + ICA + epoching + feature
+    extraction, then writes a model-ready Parquet at `output_path` (preserves
+    float64 dtype; satisfies AGENTS.md §6).
+
+    Args:
+        input_path: Path to the raw recording (.fif or .edf).
+        output_path: Where to write the processed feature Parquet file.
+            Parent directory is created if missing.
+        epoch_duration_s: Length of each fixed-duration epoch (seconds).
+        eog_ch_name: Name of the EOG channel for ICA-based artifact rejection.
+            None disables ICA.
+        n_components: Cap on ICA components.
+        random_state: Seed for ICA's solver. Required for §4 Determinism.
+
+    Raises:
+        FileNotFoundError: if `input_path` does not exist.
+        IsADirectoryError: if `output_path` resolves to an existing directory.
+    """
+    input_path = Path(input_path)
+    output_path = Path(output_path)
+    if not input_path.exists():
+        raise FileNotFoundError(f"Raw EEG file not found: {input_path}")
+
+    logger.info("Reading raw EEG from %s", input_path)
+    if input_path.suffix.lower() == ".edf":
+        raw = mne.io.read_raw_edf(input_path, preload=True, verbose="ERROR")
+    else:
+        raw = mne.io.read_raw_fif(input_path, preload=True, verbose="ERROR")
+    logger.info(
+        "Loaded %d channels, sfreq=%.1f Hz, n_times=%d",
+        len(raw.ch_names), raw.info["sfreq"], raw.n_times,
+    )
+
+    features = extract_features_from_recording(
+        raw,
+        epoch_duration_s=epoch_duration_s,
+        eog_ch_name=eog_ch_name,
+        n_components=n_components,
+        random_state=random_state,
+    )
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    if output_path.is_dir():
+        raise IsADirectoryError(
+            f"output_path must be a file, got a directory: {output_path}"
+        )
+    # Parquet preserves dtypes (float64 features stay float64) and is
+    # byte-deterministic with single-threaded snappy. AGENTS.md §6.
+    features.to_parquet(
+        output_path, index=False, engine="pyarrow", compression="snappy",
+    )
+    logger.info(
+        "Wrote processed features to %s (rows=%d, cols=%d)",
+        output_path, len(features), features.shape[1],
+    )
+
+
+if __name__ == "__main__":
+    # Day-2 CLI entrypoint — runs with default paths against `data/raw/eeg.fif`.
+    # Argument parsing (argparse / click) will land in a later task.
+    #   python -m src.pipelines.eeg_pipeline
+    run_pipeline()

tests/pipelines/test_eeg_pipeline.py

@@ -1,6 +1,7 @@
 """Unit + integration tests for the EEG pipeline."""
 from __future__ import annotations
 
+import shutil
 from pathlib import Path
 
 import mne
@@ -14,6 +15,7 @@ from src.pipelines.eeg_pipeline import (
     extract_features_from_recording,
     is_valid_epoch,
     remove_artifacts_with_ica,
+    run_pipeline,
 )
 
 
@@ -341,3 +343,87 @@ class TestExtractFeaturesFromRecording:
                 raw, epoch_duration_s=1e-6, eog_ch_name="EOG061",
                 n_components=4, random_state=97,
             )
+
+
+class TestRunPipeline:
+    def test_end_to_end_writes_processed_parquet(self, tmp_path: Path) -> None:
+        raw_dir = tmp_path / "data" / "raw"
+        proc_dir = tmp_path / "data" / "processed"
+        raw_dir.mkdir(parents=True)
+        proc_dir.mkdir(parents=True)
+        input_path = raw_dir / "rec.fif"
+        output_path = proc_dir / "eeg_features.parquet"
+        shutil.copy(FIXTURE, input_path)
+
+        run_pipeline(
+            input_path=input_path, output_path=output_path,
+            epoch_duration_s=2.0, eog_ch_name="EOG061",
+            n_components=4, random_state=97,
+        )
+
+        assert output_path.exists()
+        df = pd.read_parquet(output_path)
+        assert len(df) == 5
+        assert all(c.startswith("feat_") for c in df.columns)
+
+    def test_run_pipeline_preserves_float64_dtype(self, tmp_path: Path) -> None:
+        raw_dir = tmp_path / "data" / "raw"
+        proc_dir = tmp_path / "data" / "processed"
+        raw_dir.mkdir(parents=True)
+        proc_dir.mkdir(parents=True)
+        input_path = raw_dir / "rec.fif"
+        output_path = proc_dir / "eeg_features.parquet"
+        shutil.copy(FIXTURE, input_path)
+
+        run_pipeline(
+            input_path=input_path, output_path=output_path,
+            epoch_duration_s=2.0, eog_ch_name="EOG061",
+            n_components=4, random_state=97,
+        )
+        df = pd.read_parquet(output_path)
+        for col in df.columns:
+            assert df[col].dtype == np.float64, f"{col} widened to {df[col].dtype}"
+
+    def test_run_pipeline_is_idempotent(self, tmp_path: Path) -> None:
+        raw_dir = tmp_path / "data" / "raw"
+        proc_dir = tmp_path / "data" / "processed"
+        raw_dir.mkdir(parents=True)
+        proc_dir.mkdir(parents=True)
+        input_path = raw_dir / "rec.fif"
+        output_path = proc_dir / "eeg_features.parquet"
+        shutil.copy(FIXTURE, input_path)
+
+        run_pipeline(
+            input_path=input_path, output_path=output_path,
+            epoch_duration_s=2.0, eog_ch_name="EOG061",
+            n_components=4, random_state=97,
+        )
+        first = output_path.read_bytes()
+        run_pipeline(
+            input_path=input_path, output_path=output_path,
+            epoch_duration_s=2.0, eog_ch_name="EOG061",
+            n_components=4, random_state=97,
+        )
+        second = output_path.read_bytes()
+        assert first == second, "EEG pipeline output must be byte-deterministic"
+
+    def test_run_pipeline_raises_when_input_missing(self, tmp_path: Path) -> None:
+        with pytest.raises(FileNotFoundError):
+            run_pipeline(
+                input_path=tmp_path / "nope.fif",
+                output_path=tmp_path / "out.parquet",
+            )
+
+    def test_run_pipeline_rejects_directory_as_output(self, tmp_path: Path) -> None:
+        raw_dir = tmp_path / "data" / "raw"
+        raw_dir.mkdir(parents=True)
+        input_path = raw_dir / "rec.fif"
+        shutil.copy(FIXTURE, input_path)
+        bad_output = tmp_path / "out_dir"
+        bad_output.mkdir()
+        with pytest.raises(IsADirectoryError, match="must be a file"):
+            run_pipeline(
+                input_path=input_path, output_path=bad_output,
+                epoch_duration_s=2.0, eog_ch_name="EOG061",
+                n_components=4, random_state=97,
+            )