src/drdetect/enhance/preprocessing.py

"""Fundus preprocessing primitives.

The pipeline order matters and is not arbitrary:

    crop_from_gray -> circle_crop -> ben_graham -> (CLAHE) -> resize

`crop_from_gray` first, because the black surround otherwise drags down every
subsequent statistic (the blur in `ben_graham`, the CLAHE histogram, the
illumination estimate). `ben_graham` before `resize`, because it is defined
relative to image radius and is cheapest to reason about at native scale.

Reference: Ben Graham's winning solution, Kaggle Diabetic Retinopathy 2015.
"""

from __future__ import annotations

import cv2
import numpy as np

__all__ = [
    "crop_from_gray",
    "circle_crop",
    "ben_graham",
    "apply_clahe",
    "preprocess",
]


def crop_from_gray(image: np.ndarray, tol: int = 7) -> np.ndarray:
    """Trim the uninformative dark border around a fundus photograph.

    Fundus cameras produce a bright circular retina on a black rectangle. The
    black region carries no signal but skews global statistics, so it goes first.

    Args:
        image: HxWx3 uint8 RGB, or HxW uint8 grayscale.
        tol: Intensity at or below which a pixel counts as background.

    Returns:
        The tightest crop containing above-threshold content. Returns the input
        unchanged if the mask is empty (a fully dark frame), rather than raising
        -- an all-dark image is a *quality* failure and is Stage 1's business,
        not this function's.
    """
    if image.ndim == 2:
        mask = image > tol
        if not mask.any():
            return image
        return image[np.ix_(mask.any(1), mask.any(0))]

    gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
    mask = gray > tol
    if not mask.any():
        return image

    rows, cols = mask.any(1), mask.any(0)
    return image[np.ix_(rows, cols)]


def circle_crop(image: np.ndarray) -> np.ndarray:
    """Mask everything outside the inscribed circular field of view.

    Standardises the FOV across cameras so the model cannot learn to exploit
    corner artefacts that happen to correlate with a particular device.
    """
    image = crop_from_gray(image)
    h, w = image.shape[:2]
    cx, cy = w // 2, h // 2
    r = min(cx, cy)

    mask = np.zeros((h, w), dtype=np.uint8)
    cv2.circle(mask, (cx, cy), r, color=1, thickness=-1)

    out = image * mask[..., None] if image.ndim == 3 else image * mask
    return crop_from_gray(out)


def ben_graham(image: np.ndarray, sigma_scale: float = 30.0, weight: float = 4.0) -> np.ndarray:
    """Subtract local average colour (Ben Graham, Kaggle DR 2015).

        out = weight*img - weight*GaussianBlur(img, sigma) + 128

    The blur sigma is tied to *image radius*, not to a fixed pixel count, which
    makes the normalisation scale-invariant -- important because fundus cameras
    output wildly different resolutions.

    This removes the between-camera and between-patient variation in overall
    illumination and pigmentation while preserving the high-frequency detail
    that lesions live in.
    """
    radius = min(image.shape[:2]) / 2.0
    sigma = max(radius / sigma_scale, 1.0)
    blurred = cv2.GaussianBlur(image, (0, 0), sigmaX=sigma)
    out = cv2.addWeighted(image, weight, blurred, -weight, 128)
    return np.clip(out, 0, 255).astype(np.uint8)


def apply_clahe(image: np.ndarray, clip_limit: float = 2.0, tile_grid: int = 8) -> np.ndarray:
    """Contrast-limited adaptive histogram equalisation on the LAB L channel.

    Operating on L rather than per-RGB-channel avoids the colour casts that
    channel-wise CLAHE introduces.

    Note: `clip_limit` is a genuine risk knob. Too high and sensor noise is
    amplified into structures that look like microaneurysms. Ablate it; do not
    assume 2.0 is right for your camera mix.
    """
    lab = cv2.cvtColor(image, cv2.COLOR_RGB2LAB)
    clahe = cv2.createCLAHE(clipLimit=clip_limit, tileGridSize=(tile_grid, tile_grid))
    lab[..., 0] = clahe.apply(lab[..., 0])
    return cv2.cvtColor(lab, cv2.COLOR_LAB2RGB)


def preprocess(
    image: np.ndarray,
    size: int = 512,
    use_ben_graham: bool = True,
    use_clahe: bool = False,
) -> np.ndarray:
    """Full preprocessing chain, resized to `size` x `size`.

    `use_clahe` defaults to False deliberately: it is a hypothesis to be tested
    in the ablation, not a default to be inherited. See docs/01_PROJECT_ANALYSIS.md
    section 6, Stage 2.
    """
    out = circle_crop(image)
    if use_ben_graham:
        out = ben_graham(out)
    if use_clahe:
        out = apply_clahe(out)
    return cv2.resize(out, (size, size), interpolation=cv2.INTER_AREA)