| """Evaluate the packaged VesselCNN weights on the VesselMNIST3D test |
| split and produce the failure analysis + packaged sample volumes. |
| |
| Outputs (defaults): |
| assets/evaluation_results.json — full test metrics, occupancy |
| analysis, error inventory |
| assets/samples/*.npy — sample test volumes (uint8 28^3) |
| assets/samples/index.json — per-sample label + the prediction |
| measured from the packaged .npy |
| file itself (via vesselcnn.infer) |
| |
| Sample selection rule (fixed, stated here so it cannot be mistaken |
| for cherry-picking): from the test split, in test order — |
| * the first false negative (aneurysm read as vessel), |
| * the first false positive (vessel read as aneurysm), |
| * the correct aneurysm with the highest predicted aneurysm |
| probability, |
| * the correct vessel whose test index is the median of the correct |
| vessels. |
| If a category does not exist in the evaluation (e.g. no false |
| negatives), it is simply absent and the index records the rule. |
| |
| Usage: |
| python scripts/evaluate.py [--weights model.safetensors] |
| [--data /path/to/vesselmnist3d.npz] [--out assets/evaluation_results.json] |
| [--samples-dir assets/samples] |
| """ |
| import argparse |
| import json |
| import os |
| import sys |
|
|
| import numpy as np |
|
|
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) |
|
|
| import torch |
| from sklearn.metrics import (accuracy_score, balanced_accuracy_score, |
| confusion_matrix, precision_score, |
| recall_score, roc_auc_score) |
|
|
| from vesselcnn.infer import LABELS, TRAIN_MEAN, TRAIN_STD, load_model, predict |
| from vesselcnn.model import build_model |
|
|
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| PKG = os.path.dirname(HERE) |
| NUM_CLASSES = 2 |
|
|
|
|
| @torch.no_grad() |
| def evaluate(model, images_u8, batch=64): |
| model.eval() |
| xt = torch.from_numpy(images_u8)[:, None] |
| probs = [] |
| for i in range(0, len(xt), batch): |
| xb = (xt[i:i + batch].float() / 255.0 - TRAIN_MEAN) / TRAIN_STD |
| probs.append(torch.softmax(model(xb), dim=1)) |
| probs = torch.cat(probs).numpy() |
| return probs, probs.argmax(1) |
|
|
|
|
| def full_metrics(y_true, probs, preds): |
| cm = confusion_matrix(y_true, preds, labels=list(range(NUM_CLASSES))) |
| rec = recall_score(y_true, preds, labels=list(range(NUM_CLASSES)), |
| average=None, zero_division=0) |
| prec = precision_score(y_true, preds, labels=list(range(NUM_CLASSES)), |
| average=None, zero_division=0) |
| return { |
| "accuracy": round(float(accuracy_score(y_true, preds)), 4), |
| "auc": round(float(roc_auc_score(y_true, probs[:, 1])), 4), |
| "balanced_accuracy": round(float(balanced_accuracy_score(y_true, preds)), 4), |
| "confusion_matrix": cm.tolist(), |
| "per_class_recall": {LABELS[i]: round(float(rec[i]), 4) for i in range(NUM_CLASSES)}, |
| "per_class_precision": {LABELS[i]: round(float(prec[i]), 4) for i in range(NUM_CLASSES)}, |
| "class_counts": np.bincount(y_true, minlength=NUM_CLASSES).tolist(), |
| "majority_baseline": round(float(np.bincount(y_true).max() / len(y_true)), 4), |
| "n": int(len(y_true)), |
| } |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--weights", default=os.path.join(PKG, "model.safetensors")) |
| ap.add_argument("--data", required=True, help="path to vesselmnist3d.npz") |
| ap.add_argument("--out", default=os.path.join(PKG, "assets", "evaluation_results.json")) |
| ap.add_argument("--samples-dir", default=os.path.join(PKG, "assets", "samples")) |
| args = ap.parse_args() |
|
|
| d = np.load(args.data) |
| x_test = d["test_images"] |
| y_test = d["test_labels"].ravel().astype(np.int64) |
|
|
| model = load_model(args.weights) |
| probs, preds = evaluate(model, x_test) |
| metrics = full_metrics(y_test, probs, preds) |
|
|
| occupancy = (x_test > 0).sum(axis=(1, 2, 3)) |
| correct = preds == y_test |
| quartiles = np.quantile(occupancy, [0.25, 0.5, 0.75]) |
| occ_rows = [] |
| edges = [-1, quartiles[0], quartiles[1], quartiles[2], occupancy.max()] |
| for q in range(4): |
| m = (occupancy > edges[q]) & (occupancy <= edges[q + 1]) |
| row = {"quartile": q + 1, "occupancy_range": [int(edges[q] + 1), int(edges[q + 1])], |
| "n": int(m.sum()), "accuracy": round(float(correct[m].mean()), 4)} |
| pos = m & (y_test == 1) |
| row["aneurysm_n"] = int(pos.sum()) |
| row["aneurysm_recall"] = (round(float(correct[pos].mean()), 4) |
| if pos.sum() else None) |
| occ_rows.append(row) |
|
|
| fn_idx = [int(i) for i in np.where((y_test == 1) & (preds == 0))[0]] |
| fp_idx = [int(i) for i in np.where((y_test == 0) & (preds == 1))[0]] |
|
|
| def err_rec(i): |
| return {"test_index": int(i), "label": LABELS[int(y_test[i])], |
| "predicted": LABELS[int(preds[i])], |
| "prob_aneurysm": round(float(probs[i, 1]), 4), |
| "occupancy_voxels": int(occupancy[i])} |
|
|
| analysis = { |
| "occupancy_quartiles": occ_rows, |
| "false_negatives": {"n": len(fn_idx), "items": [err_rec(i) for i in fn_idx]}, |
| "false_positives": {"n": len(fp_idx), "items": [err_rec(i) for i in fp_idx]}, |
| "mean_prob_of_correct": round(float(probs[np.arange(len(y_test)), y_test][correct].mean()), 4), |
| "occupancy_mean_correct": round(float(occupancy[correct].mean()), 1), |
| "occupancy_mean_incorrect": round(float(occupancy[~correct].mean()), 1), |
| } |
|
|
| record = {"split": "test", **metrics, "failure_analysis": analysis} |
| os.makedirs(os.path.dirname(args.out), exist_ok=True) |
| with open(args.out, "w") as f: |
| json.dump(record, f, indent=2) |
| print(json.dumps(metrics, indent=2)) |
| print(f"wrote {args.out}") |
|
|
| |
| os.makedirs(args.samples_dir, exist_ok=True) |
| picks = {} |
| if fn_idx: |
| picks["false_negative"] = fn_idx[0] |
| if fp_idx: |
| picks["false_positive"] = fp_idx[0] |
| correct_an = [int(i) for i in np.where((y_test == 1) & (preds == 1))[0]] |
| if correct_an: |
| picks["correct_aneurysm_most_confident"] = max(correct_an, key=lambda i: probs[i, 1]) |
| correct_v = [int(i) for i in np.where((y_test == 0) & (preds == 0))[0]] |
| if correct_v: |
| picks["correct_vessel_median"] = correct_v[len(correct_v) // 2] |
| index = {"selection_rule": ( |
| "first false negative, first false positive, most-confident correct " |
| "aneurysm, median correct vessel — all in test order, from the " |
| "packaged-weights evaluation in assets/evaluation_results.json"), |
| "samples": []} |
| for role, i in picks.items(): |
| fname = f"{role}_{i}.npy" |
| np.save(os.path.join(args.samples_dir, fname), x_test[i]) |
| r = predict(model, os.path.join(args.samples_dir, fname)) |
| index["samples"].append({ |
| "file": fname, "role": role, "test_index": int(i), |
| "label": LABELS[int(y_test[i])], |
| "predicted": r["label"], |
| "probabilities": {k: round(v, 4) for k, v in r["probabilities"].items()}, |
| "occupancy_voxels": int(occupancy[i]), |
| }) |
| print(f"sample {fname}: label={LABELS[int(y_test[i])]} predicted={r['label']}") |
| with open(os.path.join(args.samples_dir, "index.json"), "w") as f: |
| json.dump(index, f, indent=2) |
| print(f"wrote {os.path.join(args.samples_dir, 'index.json')}") |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
