File size: 7,847 Bytes
1ea7ba6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | """
Cross-source evaluation harness for SpiceNet-Bench.
Splits the unified benchmark test set into source-specific subsets and reports
per-source accuracy. This is the core evaluation that exposes whether a model
has memorized source-specific shortcuts (uniform white background of Mendeley
vs. "in-the-wild" SpiceSpectrum).
For a given trained model checkpoint, produces:
* Full unified test accuracy
* SpiceSpectrum-only test accuracy
* Mendeley-only test accuracy
* Per-class breakdown
* Drop from unified -> single-source-only (the "shortcut tax")
Usage:
python eval_cross_source.py --ckpt outputs/checkpoints/unified/best.pth
python eval_cross_source.py --ckpt outputs/checkpoints/granuformer/best.pth --arch granuformer
"""
import sys, os
_base = "/mnt/d/SpiceNet" if os.path.exists("/mnt/d/SpiceNet") else "D:/SpiceNet"
sys.path.insert(0, _base)
import argparse
import json
from collections import defaultdict
from pathlib import Path
import numpy as np
import torch
import torch.nn as nn
import config
from src.dataset import (
SpiceDataset, get_val_transform, load_manifest_splits, _translate_path,
)
from torch.utils.data import DataLoader
MANIFEST = Path(_base) / "outputs" / "unified_benchmark.json"
def _source_of(path: str) -> str:
p = path.replace("\\", "/").lower()
if "/spice_spectrum/" in p:
return "spice_spectrum"
if "/indian_spices/" in p:
return "indian"
return "unknown"
def _build_filtered_loader(paths, labels, multimodal: bool, batch_size: int = 64):
ds = SpiceDataset(paths, labels, get_val_transform(), multimodal=multimodal)
return DataLoader(ds, batch_size=batch_size, num_workers=2, pin_memory=True)
def _load_model(ckpt_path: str, arch: str, num_classes: int, device):
if arch == "spicefusion":
from src.model import SpiceFusionNet
model = SpiceFusionNet(num_classes=num_classes)
ck = torch.load(ckpt_path, map_location=device)
# The trainer.save_checkpoint format keys are "model_state"
state = ck.get("model_state", ck)
model.load_state_dict(state)
return model.to(device), True # multimodal
elif arch == "granuformer":
from src.gsa import GranuFormer
model = GranuFormer(num_classes=num_classes)
ck = torch.load(ckpt_path, map_location=device)
state = ck.get("model_state", ck)
model.load_state_dict(state)
return model.to(device), False # image-only
else:
raise ValueError(f"Unknown arch: {arch}")
@torch.no_grad()
def _predict(model, loader, device, multimodal: bool):
model.eval()
preds, gts = [], []
for imgs, tex, col, labels in loader:
imgs = imgs.to(device, non_blocking=True)
labels = labels.to(device, non_blocking=True)
if multimodal:
tex, col = tex.to(device), col.to(device)
logits, _ = model.forward_fusion(imgs, tex, col)
else:
logits = model(imgs)
preds.extend(logits.argmax(1).cpu().tolist())
gts.extend(labels.cpu().tolist())
return np.asarray(gts), np.asarray(preds)
def _accuracy(y_true: np.ndarray, y_pred: np.ndarray) -> float:
if len(y_true) == 0:
return float("nan")
return float((y_true == y_pred).mean())
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--ckpt", required=True)
parser.add_argument("--arch", default="spicefusion",
choices=["spicefusion", "granuformer"])
parser.add_argument("--batch", type=int, default=64)
parser.add_argument("--out_suffix", default=None,
help="Suffix for output JSON filename")
args = parser.parse_args()
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
splits, classes = load_manifest_splits(MANIFEST)
NUM_CLASSES = len(classes)
print(f"Classes: {NUM_CLASSES}")
# Patch config so SpiceFusionNet sees correct class count
config.NUM_CLASSES = NUM_CLASSES
config.CLASSES = classes
model, multimodal = _load_model(args.ckpt, args.arch, NUM_CLASSES, device)
n_params = sum(p.numel() for p in model.parameters())
print(f"Loaded {args.arch} ({n_params:,} params) from {args.ckpt}")
print(f"Mode: {'multimodal' if multimodal else 'image-only'}")
# ── Build three filtered evaluation sets ─────────────────────────────
paths, labels = splits["test"]
src_paths = defaultdict(list)
src_labels = defaultdict(list)
for p, y in zip(paths, labels):
s = _source_of(p)
src_paths[s].append(p)
src_labels[s].append(y)
src_paths["all"].append(p)
src_labels["all"].append(y)
print(f"\nTest split by source:")
for s in ("all", "spice_spectrum", "indian"):
print(f" {s:18s} n={len(src_paths[s])}")
# ── Evaluate each ────────────────────────────────────────────────────
results = {"checkpoint": args.ckpt, "arch": args.arch,
"n_params": n_params, "by_source": {}}
for src in ("all", "spice_spectrum", "indian"):
if not src_paths[src]:
continue
loader = _build_filtered_loader(src_paths[src], src_labels[src],
multimodal=multimodal, batch_size=args.batch)
y_true, y_pred = _predict(model, loader, device, multimodal)
acc = _accuracy(y_true, y_pred)
# Per-class
per_class = {}
for c_idx, c_name in enumerate(classes):
mask = (y_true == c_idx)
if mask.sum() > 0:
per_class[c_name] = {
"n": int(mask.sum()),
"acc": float((y_pred[mask] == c_idx).mean()),
}
results["by_source"][src] = {
"n_total": len(y_true),
"accuracy": acc,
"per_class": per_class,
}
print(f"\n{src}: acc = {acc:.4f} (n={len(y_true)})")
# ── Compute the "shortcut tax" ────────────────────────────────────────
if all(s in results["by_source"] for s in ("all", "spice_spectrum", "indian")):
ss_acc = results["by_source"]["spice_spectrum"]["accuracy"]
in_acc = results["by_source"]["indian"]["accuracy"]
all_acc = results["by_source"]["all"]["accuracy"]
results["analysis"] = {
"overall_test_acc": all_acc,
"spice_spectrum_test_acc": ss_acc,
"indian_test_acc": in_acc,
"source_gap_pp": round(abs(ss_acc - in_acc) * 100, 2),
"weaker_source": "spice_spectrum" if ss_acc < in_acc else "indian",
}
print("\n──── Cross-source summary ────")
print(f" Overall : {all_acc:.4f}")
print(f" SpiceSpectrum: {ss_acc:.4f}")
print(f" Indian : {in_acc:.4f}")
print(f" Source gap : {results['analysis']['source_gap_pp']:.2f} pp "
f"(weaker: {results['analysis']['weaker_source']})")
# ── Save ─────────────────────────────────────────────────────────────
suffix = args.out_suffix or args.arch
out = config.OUTPUT_DIR / f"cross_source_{suffix}.json"
with open(out, "w") as f:
json.dump(results, f, indent=2)
print(f"\nSaved: {out}")
if __name__ == "__main__":
main()
|