models/m1_segmentation/preprocessing.py

"""
preprocessing.py
----------------
Load and preprocess BraTS volumes for inference.

Supports two input modes:
  1. Pre-built .npz files  (same format used during training)
  2. Raw NIfTI files       (one file per MRI modality)

The four expected BraTS modalities and their array order:
  Index 0 → T1
  Index 1 → T1ce
  Index 2 → T2
  Index 3 → FLAIR
"""

import os
import numpy as np
import torch
from typing import Tuple, Optional, List


# ── Normalisation ─────────────────────────────────────────────────────────────
def z_score_normalise(volume: np.ndarray) -> np.ndarray:
    """
    Per-channel Z-score normalisation, computed only over non-zero voxels
    so that the large background region does not skew the statistics.

    Args:
        volume: float32 array of shape (C, D, H, W)

    Returns:
        Normalised array of the same shape.
    """
    out = np.zeros_like(volume, dtype=np.float32)
    for c in range(volume.shape[0]):
        ch   = volume[c]
        mask = ch > 0
        if mask.any():
            mu    = ch[mask].mean()
            sigma = ch[mask].std()
            out[c] = np.where(mask, (ch - mu) / (sigma + 1e-8), 0.0)
        # else: channel stays zero (no signal)
    return out


# ── Padding helpers ───────────────────────────────────────────────────────────
def pad_to_multiple(
    volume: np.ndarray,
    multiple: int = 16,
) -> Tuple[np.ndarray, Tuple[int, ...]]:
    """
    Pad a (C, D, H, W) volume so that every spatial dimension is a multiple
    of `multiple`.  Padding is added at the *end* of each axis.

    Returns:
        padded_volume : padded array
        pad_amounts   : (pd, ph, pw) — amount added to each spatial axis,
                        needed to crop back after inference.
    """
    _, D, H, W = volume.shape
    pad = lambda n: (multiple - n % multiple) % multiple
    pd, ph, pw = pad(D), pad(H), pad(W)
    if pd or ph or pw:
        volume = np.pad(volume, ((0, 0), (0, pd), (0, ph), (0, pw)))
    return volume, (pd, ph, pw)


def unpad(mask: np.ndarray, pad_amounts: Tuple[int, ...]) -> np.ndarray:
    """
    Remove the padding that was added by pad_to_multiple.

    Args:
        mask        : (D', H', W') integer array
        pad_amounts : (pd, ph, pw) as returned by pad_to_multiple

    Returns:
        Cropped mask of shape (D, H, W).
    """
    pd, ph, pw = pad_amounts
    D, H, W = mask.shape
    return mask[
        : D - pd if pd else D,
        : H - ph if ph else H,
        : W - pw if pw else W,
    ]


# ── NPZ loader ────────────────────────────────────────────────────────────────
def load_npz(
    path: str,
    normalise: bool = True,
) -> Tuple[np.ndarray, Optional[np.ndarray]]:
    """
    Load a pre-built .npz file that contains keys ``X`` and optionally ``Y``.

    Args:
        path      : path to the .npz file
        normalise : apply Z-score normalisation to X

    Returns:
        X : float32 (C, D, H, W)
        Y : int64   (D, H, W)  or None if the file has no ground-truth label
    """
    data = np.load(path)
    X    = data["X"].astype(np.float32)
    Y    = data["Y"].astype(np.int64) if "Y" in data else None
    data.close()

    if normalise:
        X = z_score_normalise(X)

    return X, Y


# ── NIfTI loader ──────────────────────────────────────────────────────────────
def load_nifti(
    t1_path:    str,
    t1ce_path:  str,
    t2_path:    str,
    flair_path: str,
    label_path: Optional[str] = None,
    normalise:  bool = True,
) -> Tuple[np.ndarray, Optional[np.ndarray]]:
    """
    Load four BraTS NIfTI modalities and optionally a segmentation label.

    Requires ``nibabel``.  Install with:  pip install nibabel

    Args:
        t1_path    : path to T1    .nii / .nii.gz
        t1ce_path  : path to T1ce  .nii / .nii.gz
        t2_path    : path to T2    .nii / .nii.gz
        flair_path : path to FLAIR .nii / .nii.gz
        label_path : (optional) path to segmentation .nii / .nii.gz
        normalise  : apply Z-score normalisation to X

    Returns:
        X : float32 (4, D, H, W)
        Y : int64   (D, H, W)  or None
    """
    try:
        import nibabel as nib
    except ImportError as exc:
        raise ImportError(
            "nibabel is required for NIfTI loading.  "
            "Install it with:  pip install nibabel"
        ) from exc

    channels: List[np.ndarray] = []
    for p in [t1_path, t1ce_path, t2_path, flair_path]:
        vol = nib.load(p).get_fdata(dtype=np.float32)
        channels.append(vol)

    # Stack → (C, D, H, W).  NIfTI spatial order is (X, Y, Z); we treat
    # the three spatial axes generically and do not reorder them here.
    X = np.stack(channels, axis=0)

    Y: Optional[np.ndarray] = None
    if label_path is not None:
        Y = nib.load(label_path).get_fdata(dtype=np.float32).astype(np.int64)

    if normalise:
        X = z_score_normalise(X)

    return X, Y


# ── Torch tensor factory ──────────────────────────────────────────────────────
def to_tensor(
    X: np.ndarray,
    device: torch.device,
    pad_multiple: int = 16,
) -> Tuple[torch.Tensor, Tuple[int, ...]]:
    """
    Pad and convert a (C, D, H, W) numpy array to a batched float32 tensor.

    Returns:
        tensor      : shape (1, C, D', H', W') on `device`
        pad_amounts : passed to unpad() after inference
    """
    X, pad_amounts = pad_to_multiple(X, multiple=pad_multiple)
    tensor = torch.from_numpy(X).unsqueeze(0).to(device, non_blocking=True)
    return tensor, pad_amounts