feat(mri): add extract_features_from_volume (8 ROI octants × 6 stats)

src/pipelines/mri_pipeline.py

@@ -93,3 +93,77 @@ def mask_brain(
             float(volume.min()), float(volume.max()), intensity_threshold,
         )
     return cleaned
+
+
+# Default ROI partition: split a (D, H, W) volume into 2×2×2 = 8 octant ROIs.
+# Octant index follows binary (z, y, x) ordering: 0..7.
+DEFAULT_N_ROI_AXES: tuple[int, int, int] = (2, 2, 2)
+ROI_STATS: tuple[str, ...] = ("mean", "std", "p10", "p50", "p90", "voxel_count")
+
+
+def _roi_slices(
+    shape: tuple[int, int, int],
+    n_roi_axes: tuple[int, int, int],
+) -> list[tuple[slice, slice, slice]]:
+    """Generate the ROI slice list in deterministic (z, y, x) octant order."""
+    nz, ny, nx = n_roi_axes
+    dz, dy, dx = shape
+    bins_z = np.array_split(np.arange(dz), nz)
+    bins_y = np.array_split(np.arange(dy), ny)
+    bins_x = np.array_split(np.arange(dx), nx)
+    out: list[tuple[slice, slice, slice]] = []
+    for bz in bins_z:
+        for by in bins_y:
+            for bx in bins_x:
+                out.append((
+                    slice(bz[0], bz[-1] + 1),
+                    slice(by[0], by[-1] + 1),
+                    slice(bx[0], bx[-1] + 1),
+                ))
+    return out
+
+
+def _roi_stats_for(values: np.ndarray) -> dict[str, float]:
+    """Compute the 6 ROI stats. Empty array → all 0.0 (no-NaN contract)."""
+    if values.size == 0:
+        return {stat: 0.0 for stat in ROI_STATS}
+    return {
+        "mean": float(values.mean()),
+        "std": float(values.std()),
+        "p10": float(np.percentile(values, 10)),
+        "p50": float(np.percentile(values, 50)),
+        "p90": float(np.percentile(values, 90)),
+        "voxel_count": float(values.size),
+    }
+
+
+def extract_features_from_volume(
+    volume: np.ndarray,
+    mask: np.ndarray,
+    n_roi_axes: tuple[int, int, int] = DEFAULT_N_ROI_AXES,
+) -> dict[str, float]:
+    """Compute per-ROI summary statistics from a masked volume.
+
+    The volume is partitioned into ``prod(n_roi_axes)`` axis-aligned octants
+    in deterministic (z, y, x) order. For each ROI, intensity values from
+    voxels where `mask` is True are summarized via mean / std / 10th, 50th,
+    90th percentile / voxel count. Empty ROIs (no mask voxels) report all
+    zeros so the resulting Parquet has no NaN values.
+
+    Args:
+        volume: 3-D numeric `np.ndarray` (already validated).
+        mask: Boolean `np.ndarray` of the same shape (from `mask_brain`).
+        n_roi_axes: ROI grid along (z, y, x). Default `(2, 2, 2)` → 8 ROIs.
+
+    Returns:
+        Flat dict `{"feat_roi{i}_{stat}": float}` of length
+        ``prod(n_roi_axes) * len(ROI_STATS)``.
+    """
+    feats: dict[str, float] = {}
+    slices = _roi_slices(volume.shape, n_roi_axes)
+    for i, sl in enumerate(slices):
+        roi_values = volume[sl][mask[sl]]
+        stats = _roi_stats_for(roi_values)
+        for stat_name, stat_val in stats.items():
+            feats[f"feat_roi{i}_{stat_name}"] = stat_val
+    return feats

tests/pipelines/test_mri_pipeline.py

@@ -8,6 +8,9 @@ import numpy as np
 import pytest
 
 from src.pipelines.mri_pipeline import (
+    DEFAULT_N_ROI_AXES,
+    ROI_STATS,
+    extract_features_from_volume,
     is_valid_volume,
     mask_brain,
 )
@@ -128,3 +131,61 @@ class TestMaskBrain:
         log_output = buf.getvalue()
         assert "all-False mask" in log_output
         assert "downstream features for this volume will be all-zero" in log_output
+
+
+class TestExtractFeaturesFromVolume:
+    def _load_subject(self, sid: str) -> np.ndarray:
+        return nib.load(FIXTURE_DIR / f"{sid}.nii.gz").get_fdata()
+
+    def test_returns_dict_with_correct_keys(self) -> None:
+        vol = self._load_subject("subject_0")
+        mask = mask_brain(vol)
+        feats = extract_features_from_volume(vol, mask)
+        n_roi = int(np.prod(DEFAULT_N_ROI_AXES))
+        expected = {
+            f"feat_roi{i}_{stat}"
+            for i in range(n_roi)
+            for stat in ROI_STATS
+        }
+        assert set(feats.keys()) == expected
+
+    def test_feature_count_matches_contract(self) -> None:
+        vol = self._load_subject("subject_0")
+        mask = mask_brain(vol)
+        feats = extract_features_from_volume(vol, mask)
+        n_roi = int(np.prod(DEFAULT_N_ROI_AXES))
+        assert len(feats) == n_roi * len(ROI_STATS)
+
+    def test_all_features_finite_float(self) -> None:
+        vol = self._load_subject("subject_0")
+        mask = mask_brain(vol)
+        feats = extract_features_from_volume(vol, mask)
+        for k, v in feats.items():
+            assert isinstance(v, float), f"{k}: {type(v).__name__}"
+            assert np.isfinite(v), f"{k}: {v}"
+
+    def test_voxel_count_is_integer_valued(self) -> None:
+        vol = self._load_subject("subject_0")
+        mask = mask_brain(vol)
+        feats = extract_features_from_volume(vol, mask)
+        for k, v in feats.items():
+            if k.endswith("_voxel_count"):
+                # voxel_count stored as float for column-uniformity, but must be
+                # a whole number.
+                assert v == float(int(v))
+
+    def test_empty_mask_yields_zero_features(self) -> None:
+        """If a volume has zero brain voxels (mask all False), every stat
+        must default to 0.0 — not NaN — to preserve the no-NaN Parquet contract."""
+        vol = self._load_subject("subject_0")
+        empty_mask = np.zeros_like(vol, dtype=bool)
+        feats = extract_features_from_volume(vol, empty_mask)
+        for k, v in feats.items():
+            assert v == 0.0, f"{k}: {v}"
+
+    def test_deterministic_for_same_input(self) -> None:
+        vol = self._load_subject("subject_0")
+        mask = mask_brain(vol)
+        a = extract_features_from_volume(vol, mask)
+        b = extract_features_from_volume(vol, mask)
+        assert a == b