Upload folder using huggingface_hub

src/biomarkers.py

@@ -0,0 +1,81 @@
+# -*- coding: utf-8 -*-
+"""
+Interpretable retinal imaging biomarkers of anti-VEGF intolerance.
+
+Computes, from a single color fundus photograph:
+  - vascular density        : fraction of FOV occupied by the vessel mask
+  - vascular skeleton length: normalized length of the vessel skeleton
+  - vascular fractal dimension : box-counting fractal dimension of the vessels
+
+Vessels are enhanced with CLAHE + morphological black-hat on the green channel.
+PIL is used to read images (supports non-ASCII paths).
+"""
+import argparse
+import numpy as np
+import cv2
+from PIL import Image
+from skimage.morphology import skeletonize
+
+
+def _fov_mask(gray):
+    _, m = cv2.threshold(gray, 12, 255, cv2.THRESH_BINARY)
+    return m > 0
+
+
+def _vessel_mask(img_bgr, mask):
+    g = img_bgr[:, :, 1]
+    clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
+    ge = clahe.apply(g)
+    kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (15, 15))
+    bh = cv2.morphologyEx(ge, cv2.MORPH_BLACKHAT, kernel)
+    _, vessel = cv2.threshold(bh, 15, 255, cv2.THRESH_BINARY)
+    return (vessel > 0) & mask
+
+
+def fractal_dimension(binary):
+    """Box-counting fractal dimension of a binary structure.
+
+    D = -slope of log N(eps) vs log eps, where N(eps) is the number of boxes of
+    side eps that intersect the structure.
+    """
+    Z = binary > 0
+    if Z.sum() == 0:
+        return 0.0
+
+    def boxcount(Z, k):
+        S = np.add.reduceat(
+            np.add.reduceat(Z, np.arange(0, Z.shape[0], k), axis=0),
+            np.arange(0, Z.shape[1], k), axis=1)
+        return len(np.where((S > 0) & (S < k * k))[0])
+
+    sizes = 2 ** np.arange(1, 7)
+    counts = [max(boxcount(Z, int(s)), 1) for s in sizes]
+    coeffs = np.polyfit(np.log(sizes), np.log(counts), 1)
+    return float(-coeffs[0])
+
+
+def compute_biomarkers(image_path: str, size: int = 512) -> dict:
+    pil = Image.open(image_path).convert("RGB")
+    img = cv2.cvtColor(np.array(pil), cv2.COLOR_RGB2BGR)
+    img = cv2.resize(img, (size, size))
+    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+    mask = _fov_mask(gray)
+    area = max(int(mask.sum()), 1)
+
+    vessel = _vessel_mask(img, mask)
+    density = float(vessel.sum() / area)
+    skeleton_len = float(skeletonize(vessel).sum() / area)
+    fd = fractal_dimension(vessel)
+    return {
+        "vascular_density": round(density, 4),
+        "vascular_skeleton_length": round(skeleton_len, 4),
+        "vascular_fractal_dimension": round(fd, 4),
+    }
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser(description="Compute retinal vascular biomarkers.")
+    ap.add_argument("--image", required=True)
+    args = ap.parse_args()
+    for k, v in compute_biomarkers(args.image).items():
+        print(f"{k}: {v}")

src/dataset.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+"""
+Fundus image dataset and transforms.
+
+Expects a manifest CSV with columns:
+    image_path,label[,patient_id]
+where label in {0,1}  (0 = tolerant / NPDR, 1 = intolerant / PDR).
+patient_id is optional but recommended for patient-level cross-validation splits.
+
+NO patient data is distributed with this repository; supply your own
+ethically-approved, de-identified manifest and images.
+"""
+import pandas as pd
+from PIL import Image
+from torch.utils.data import Dataset
+from torchvision import transforms
+
+IMAGENET_MEAN = [0.485, 0.456, 0.406]
+IMAGENET_STD = [0.229, 0.224, 0.225]
+
+
+def build_transforms(image_size: int = 224, train: bool = False):
+    norm = transforms.Normalize(IMAGENET_MEAN, IMAGENET_STD)
+    if train:
+        return transforms.Compose([
+            transforms.RandomResizedCrop(image_size, scale=(0.8, 1.0)),
+            transforms.RandomHorizontalFlip(),
+            transforms.RandomRotation(15),
+            transforms.ColorJitter(0.1, 0.1, 0.1),
+            transforms.ToTensor(), norm,
+        ])
+    return transforms.Compose([
+        transforms.Resize((image_size, image_size)),
+        transforms.ToTensor(), norm,
+    ])
+
+
+class FundusDataset(Dataset):
+    def __init__(self, manifest_csv: str, image_size: int = 224, train: bool = False):
+        self.df = pd.read_csv(manifest_csv)
+        assert {"image_path", "label"}.issubset(self.df.columns), \
+            "manifest must contain columns: image_path,label[,patient_id]"
+        self.tf = build_transforms(image_size, train)
+
+    def __len__(self):
+        return len(self.df)
+
+    def __getitem__(self, i):
+        r = self.df.iloc[i]
+        x = self.tf(Image.open(r["image_path"]).convert("RGB"))
+        return x, int(r["label"])

src/gradcam.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+"""Grad-CAM explanation for the DINOv2 (ViT-L/14) intolerance classifier.
+
+ViT tokens are reshaped to a 16x16 grid (224/14) for spatial attribution.
+"""
+import argparse
+import numpy as np
+import torch
+from PIL import Image
+from pytorch_grad_cam import GradCAM
+from pytorch_grad_cam.utils.image import show_cam_on_image
+from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
+from .model import load_dinov2, dinov2_gradcam_target, dinov2_reshape
+from .dataset import build_transforms
+
+
+def reshape_transform(tensor, grid=16):
+    return dinov2_reshape(tensor, grid)
+
+
+def gradcam(image_path: str, weights: str, out_path: str = "cam.png",
+            device: str = None, target_class: int = 1):
+    device = device or ("cuda" if torch.cuda.is_available() else "cpu")
+    model = load_dinov2(weights, device)
+    target_layers = dinov2_gradcam_target(model)
+    cam = GradCAM(model=model, target_layers=target_layers,
+                  reshape_transform=reshape_transform)
+
+    pil = Image.open(image_path).convert("RGB").resize((224, 224))
+    rgb = np.array(pil).astype(np.float32) / 255.0
+    x = build_transforms(224, train=False)(pil).unsqueeze(0).to(device)
+    grayscale = cam(input_tensor=x, targets=[ClassifierOutputTarget(target_class)])[0]
+    vis = show_cam_on_image(rgb, grayscale, use_rgb=True)
+    Image.fromarray(vis).save(out_path)
+    print("saved:", out_path)
+    return vis
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser(description="Grad-CAM for the intolerance classifier.")
+    ap.add_argument("--image", required=True)
+    ap.add_argument("--weights", required=True)
+    ap.add_argument("--out", default="cam.png")
+    args = ap.parse_args()
+    gradcam(args.image, args.weights, args.out)

src/inference.py

@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+"""Single-image (or batch) inference: anti-VEGF intolerance risk score."""
+import argparse
+import torch
+from PIL import Image
+from .model import load_dinov2
+from .dataset import build_transforms
+
+
+def predict(image_path: str, weights: str, device: str = None, threshold: float = 0.5):
+    device = device or ("cuda" if torch.cuda.is_available() else "cpu")
+    model = load_dinov2(weights, device)   # primary backbone: DINOv2 (ViT-L/14)
+    tf = build_transforms(224, train=False)
+    x = tf(Image.open(image_path).convert("RGB")).unsqueeze(0).to(device)
+    with torch.no_grad():
+        prob = torch.softmax(model(x), 1)[0, 1].item()
+    return {
+        "intolerance_risk": round(prob, 4),
+        "prediction": "intolerant" if prob >= threshold else "tolerant",
+        "threshold": threshold,
+    }
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser(description="anti-VEGF intolerance risk from a fundus image.")
+    ap.add_argument("--image", required=True)
+    ap.add_argument("--weights", required=True)
+    ap.add_argument("--threshold", type=float, default=0.5)
+    args = ap.parse_args()
+    out = predict(args.image, args.weights, threshold=args.threshold)
+    for k, v in out.items():
+        print(f"{k}: {v}")

src/model.py

@@ -0,0 +1,97 @@
+# -*- coding: utf-8 -*-
+"""
+Models for anti-VEGF intolerance prediction.
+
+PRIMARY: DINOv2 (ViT-L/14) — a generalist vision foundation model (Meta AI, Apache-2.0),
+fine-tuned for binary classification. Released fine-tuned weights (dino_deploy.pth) are
+Apache-2.0 (free for research and commercial use with attribution). Use `build_dinov2`
+/ `load_dinov2`.
+
+COMPARATOR: RETFound (ViT-L/16) — a retinal-domain foundation model (Zhou et al.,
+Nature 2023, CC BY-NC 4.0). Provided via `FundusClassifier` for the head-to-head
+comparison reported in the paper; RETFound weights are NOT redistributed here.
+"""
+import torch
+import torch.nn as nn
+import timm
+
+
+class FundusClassifier(nn.Module):
+    """ViT-L/16 backbone (RETFound-initialized) + linear head.
+
+    Args:
+        num_classes: 2 (tolerant / intolerant).
+        retfound_weights: path to RETFound .pth, or None to start from timm init.
+        drop_rate: dropout for the head.
+    """
+
+    def __init__(self, num_classes: int = 2, retfound_weights: str | None = None,
+                 backbone: str = "vit_large_patch16_224", drop_rate: float = 0.2):
+        super().__init__()
+        self.backbone = timm.create_model(backbone, pretrained=False,
+                                          num_classes=0, drop_rate=drop_rate)
+        if retfound_weights:
+            self._load_retfound(retfound_weights)
+        d = self.backbone.num_features
+        self.head = nn.Sequential(
+            nn.LayerNorm(d), nn.Dropout(drop_rate), nn.Linear(d, num_classes)
+        )
+
+    def _load_retfound(self, path: str):
+        sd = torch.load(path, map_location="cpu", weights_only=False)
+        sd = sd.get("model", sd)
+        own = self.backbone.state_dict()
+        matched = {k: v for k, v in sd.items()
+                   if not k.startswith("decoder") and k != "mask_token"
+                   and k in own and own[k].shape == v.shape}
+        own.update(matched)
+        self.backbone.load_state_dict(own, strict=False)
+        print(f"[RETFound] loaded {len(matched)}/{len(own)} encoder tensors")
+
+    def forward(self, x, return_feat: bool = False):
+        f = self.backbone(x)
+        logits = self.head(f)
+        return (logits, f) if return_feat else logits
+
+
+def load_finetuned(weights_path: str, device: str = "cuda"):
+    """Load a fully fine-tuned RETFound (comparator) classifier checkpoint."""
+    model = FundusClassifier(num_classes=2, retfound_weights=None)
+    state = torch.load(weights_path, map_location="cpu")
+    model.load_state_dict(state)
+    return model.to(device).eval()
+
+
+# --------------------------------------------------------------------------- #
+# PRIMARY model: DINOv2 (ViT-L/14) generalist vision foundation model
+# --------------------------------------------------------------------------- #
+def build_dinov2(num_classes: int = 2, img_size: int = 224, drop_rate: float = 0.2):
+    """DINOv2 (ViT-L/14) backbone + LayerNorm/Dropout/Linear head, as an nn.Sequential.
+
+    The 224-px input yields a 16x16 = 256 patch-token grid. The returned module's
+    state_dict matches the released `dino_deploy.pth` (keys '0.*' backbone, '1.*' head).
+    """
+    backbone = timm.create_model("vit_large_patch14_dinov2", pretrained=False,
+                                 num_classes=0, img_size=img_size, drop_rate=drop_rate)
+    head = nn.Sequential(nn.LayerNorm(backbone.num_features),
+                         nn.Dropout(drop_rate),
+                         nn.Linear(backbone.num_features, num_classes))
+    return nn.Sequential(backbone, head)
+
+
+def load_dinov2(weights_path: str, device: str = "cuda", img_size: int = 224):
+    """Load the fine-tuned DINOv2 classifier (dino_deploy.pth) for inference."""
+    model = build_dinov2(num_classes=2, img_size=img_size)
+    model.load_state_dict(torch.load(weights_path, map_location="cpu"))
+    return model.to(device).eval()
+
+
+def dinov2_gradcam_target(model):
+    """Grad-CAM target layer for the DINOv2 Sequential model."""
+    return [model[0].blocks[-1].norm1]
+
+
+def dinov2_reshape(tensor, grid: int = 16):
+    """reshape_transform for Grad-CAM on DINOv2 (keep the last grid*grid patch tokens)."""
+    x = tensor[:, -grid * grid:, :]
+    return x.reshape(tensor.size(0), grid, grid, tensor.size(2)).permute(0, 3, 1, 2)

src/train.py

@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+"""
+Fine-tune a vision foundation model for anti-VEGF intolerance (5-fold, patient-level CV).
+
+Primary backbone: DINOv2 (ViT-L/14, Apache-2.0); set `backbone: retfound` in the config
+to fine-tune the RETFound comparator instead. Class-weighted cross-entropy handles the
+~9:1 imbalance; AMP + cosine schedule; early stopping on validation balanced accuracy.
+Supply your own de-identified manifest (see src/dataset.py). NO patient data is included.
+"""
+import argparse, yaml
+import numpy as np
+import pandas as pd
+import torch
+import torch.nn as nn
+from torch.utils.data import DataLoader
+from sklearn.model_selection import StratifiedGroupKFold, StratifiedKFold
+from sklearn.metrics import roc_auc_score, balanced_accuracy_score
+from .model import FundusClassifier, build_dinov2
+from .dataset import FundusDataset, build_transforms
+from PIL import Image
+
+
+class _DS(torch.utils.data.Dataset):
+    def __init__(self, df, train, size=224):
+        self.df = df.reset_index(drop=True); self.tf = build_transforms(size, train)
+    def __len__(self): return len(self.df)
+    def __getitem__(self, i):
+        r = self.df.iloc[i]
+        return self.tf(Image.open(r["image_path"]).convert("RGB")), int(r["label"])
+
+
+def class_weights(labels, device):
+    c = np.bincount(labels, minlength=2).astype(float)
+    w = c.sum() / (2 * np.maximum(c, 1))
+    return torch.tensor(w, dtype=torch.float32, device=device)
+
+
+def train_one_fold(tr, va, cfg, device):
+    tr_ds, va_ds = _DS(tr, True), _DS(va, False)
+    tr_ld = DataLoader(tr_ds, cfg["batch_size"], shuffle=True, num_workers=4, drop_last=True)
+    va_ld = DataLoader(va_ds, cfg["batch_size"], shuffle=False, num_workers=4)
+    if cfg.get("backbone", "dinov2") == "retfound":
+        model = FundusClassifier(2, cfg.get("retfound_weights")).to(device)
+    else:
+        model = build_dinov2(2, img_size=cfg.get("image_size", 224)).to(device)
+    crit = nn.CrossEntropyLoss(weight=class_weights(tr["label"].values, device))
+    opt = torch.optim.AdamW(model.parameters(), lr=cfg["lr"], weight_decay=cfg["weight_decay"])
+    sched = torch.optim.lr_scheduler.CosineAnnealingLR(opt, T_max=cfg["epochs"])
+    scaler = torch.amp.GradScaler("cuda", enabled=device == "cuda")
+    best, bad = -1, 0
+    for ep in range(cfg["epochs"]):
+        model.train()
+        for x, y in tr_ld:
+            x, y = x.to(device), y.to(device); opt.zero_grad()
+            with torch.amp.autocast("cuda", enabled=device == "cuda"):
+                loss = crit(model(x), y)
+            scaler.scale(loss).backward(); scaler.step(opt); scaler.update()
+        sched.step()
+        ys, ps = _eval(model, va_ld, device)
+        auc = roc_auc_score(ys, ps); ba = balanced_accuracy_score(ys, (ps >= 0.5).astype(int))
+        if ba > best: best, bad = ba, 0; torch.save(model.state_dict(), cfg["out_weights"])
+        else:
+            bad += 1
+            if bad >= cfg.get("patience", 10): break
+    return best
+
+
+@torch.no_grad()
+def _eval(model, ld, device):
+    model.eval(); ys, ps = [], []
+    for x, y in ld:
+        p = torch.softmax(model(x.to(device)), 1)[:, 1].cpu().numpy()
+        ps.extend(p); ys.extend(y.numpy())
+    return np.array(ys), np.array(ps)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--config", default="configs/default.yaml")
+    args = ap.parse_args()
+    cfg = yaml.safe_load(open(args.config, encoding="utf-8"))
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    df = pd.read_csv(cfg["manifest"])
+    y = df["label"].values
+    if "patient_id" in df.columns:
+        splitter = StratifiedGroupKFold(cfg["folds"], shuffle=True, random_state=cfg["seed"])
+        folds = splitter.split(df, y, df["patient_id"].values)
+    else:
+        splitter = StratifiedKFold(cfg["folds"], shuffle=True, random_state=cfg["seed"])
+        folds = splitter.split(df, y)
+    aucs = []
+    for k, (tr_idx, va_idx) in enumerate(folds):
+        cfg["out_weights"] = f"weights/fold{k}.pth"
+        best = train_one_fold(df.iloc[tr_idx], df.iloc[va_idx], cfg, device)
+        print(f"fold{k}: best balanced_acc={best:.3f}")
+        aucs.append(best)
+    print(f"mean balanced_acc: {np.mean(aucs):.3f} ± {np.std(aucs):.3f}")
+
+
+if __name__ == "__main__":
+    main()