ml/scripts/test_external.py

#!/usr/bin/env python3
"""
Test a trained CropIntel model on EXTERNAL images (outside the training datasets).

This is the real-world readiness check. In-dataset test accuracy overstates field
performance; this script tells you how the model behaves on images it has never
seen from a different distribution.

Two modes:
  1. LABELED EVAL — point at a directory with one subfolder per true class
     (subfolder names are matched to the model's class names, case/space/underscore
     insensitive). Produces a confusion matrix, per-class recall, accuracy, and an
     out-of-distribution (OOD) confidence report.
  2. UNLABELED PREDICT — point at a directory of loose images (or a single image).
     Produces top-2 predictions + confidence + a "below threshold / uncertain" flag.

Usage:
  python -m ml.scripts.test_external --crop corn --path /some/photo.jpg
  python -m ml.scripts.test_external --crop corn --path /folder/of/images
  python -m ml.scripts.test_external --crop corn --path /labeled_root   # subfolders=classes
  python -m ml.scripts.test_external --crop rice --path ./imgs --backend tflite

Labeled layout example:
  labeled_root/
    Healthy/        img1.jpg ...
    Common Rust/    ...
    Blight/         ...
    Gray Leaf Spot/ ...
"""
import argparse
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

os.environ.setdefault("TF_CPP_MIN_LOG_LEVEL", "2")

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))

import numpy as np  # noqa: E402
from PIL import Image  # noqa: E402

from ml.config import CROPS, CONFIDENCE_THRESHOLD  # noqa: E402

IMG_EXTS = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}

# Acceptance gate for the "production-ready" stamp (see docs/DEPLOYMENT.md):
# a model passes when external accuracy meets GATE_MIN_ACCURACY and every
# evaluated class's recall meets GATE_MIN_CLASS_RECALL.
GATE_MIN_ACCURACY = 0.85
GATE_MIN_CLASS_RECALL = 0.60


def _norm(s: str) -> str:
    """Normalize a class/folder name for matching."""
    return s.lower().replace("_", " ").replace("-", " ").strip()


def _list_images(folder: Path):
    return sorted(p for p in folder.rglob("*")
                  if p.is_file() and p.suffix.lower() in IMG_EXTS)


def _load_predictor(crop: str, backend: str, version: str = None):
    if backend == "keras":
        from ml.inference.keras_predictor import KerasPredictor
        return KerasPredictor(crop, version=version)
    else:
        from ml.inference.tflite_predictor import TFLitePredictor
        return TFLitePredictor(crop, version=version)


def _predict_one(predictor, img_path: Path):
    """Return (sorted_all_predictions, top_disease, top_conf) via the production path."""
    image = Image.open(img_path)
    result = predictor.predict(image)
    return result["all_predictions"], result["disease"], result["confidence"]


def _print_confusion(cm, class_names):
    col_w = max(14, max(len(n) for n in class_names) + 2)
    corner = "true\\pred"
    print("\n" + f"{corner:>{col_w}}" + "".join(f"{n[:col_w-1]:>{col_w}}" for n in class_names))
    for i, row in enumerate(cm):
        print(f"{class_names[i][:col_w-1]:>{col_w}}" + "".join(f"{v:>{col_w}}" for v in row))


def run_labeled(predictor, root: Path):
    """Eval against subfolders named by true class."""
    class_names = predictor.class_names
    norm_to_idx = {_norm(c): i for i, c in enumerate(class_names)}

    subdirs = [d for d in root.iterdir() if d.is_dir()]
    matched = [(d, norm_to_idx[_norm(d.name)]) for d in subdirs if _norm(d.name) in norm_to_idx]
    unmatched = [d.name for d in subdirs if _norm(d.name) not in norm_to_idx]
    if not matched:
        print(f"  No subfolders matched model classes {class_names}.")
        print(f"  Found subfolders: {[d.name for d in subdirs]}")
        print("  (Falling back to unlabeled prediction.)")
        return run_unlabeled(predictor, root, recursive=True)

    if unmatched:
        print(f"  [note] ignoring subfolders not matching a class: {unmatched}")

    n = len(class_names)
    cm = np.zeros((n, n), dtype=int)
    confs, below = [], 0
    total = 0
    for folder, true_idx in matched:
        for img in _list_images(folder):
            try:
                allp, top, conf = _predict_one(predictor, img)
            except Exception as e:
                print(f"  [skip] {img.name}: {e}")
                continue
            pred_idx = class_names.index(top)
            cm[true_idx][pred_idx] += 1
            confs.append(conf)
            below += int(conf < CONFIDENCE_THRESHOLD)
            total += 1

    if total == 0:
        print("  No readable images found.")
        return None

    _print_confusion(cm, class_names)
    per_class = {}
    print("\nPer-class recall (true class correctly predicted):")
    for i, name in enumerate(class_names):
        tot = cm[i].sum()
        rec = cm[i, i] / tot if tot else 0.0
        marker = "   <-- WEAK (<0.6)" if (tot and rec < 0.6) else ""
        print(f"  {name:<26} {cm[i,i]:>4}/{tot:<4} = {rec:6.1%}{marker}")
        if tot:
            per_class[name] = {"correct": int(cm[i, i]), "total": int(tot),
                               "recall": round(rec, 4)}
    acc = cm.trace() / cm.sum()
    confs = np.array(confs)
    print(f"\nOverall external accuracy : {acc:.1%}  ({cm.trace()}/{cm.sum()})")
    print(f"Mean confidence           : {confs.mean():.1%}")
    print(f"Below threshold ({CONFIDENCE_THRESHOLD:.2f})    : {below}/{total} = {below/total:.1%}")

    gate_passed = bool(acc >= GATE_MIN_ACCURACY
                       and all(c["recall"] >= GATE_MIN_CLASS_RECALL for c in per_class.values()))
    verdict = "PASS" if gate_passed else "FAIL"
    print(f"\nProduction gate (acc>={GATE_MIN_ACCURACY:.0%}, "
          f"class recall>={GATE_MIN_CLASS_RECALL:.0%}): {verdict}")
    print("\nReading the result:")
    print("  * accuracy here ~ in-dataset test acc  -> generalizes well")
    print("  * accuracy here << test acc            -> domain shift / shortcut learning")
    print("  * high accuracy BUT low mean confidence-> shaky; rely on threshold + top-2")

    return {
        "external_accuracy": round(float(acc), 4),
        "total_images": int(total),
        "correct": int(cm.trace()),
        "per_class": per_class,
        "mean_confidence": round(float(confs.mean()), 4),
        "below_threshold_rate": round(below / total, 4),
        "confidence_threshold": CONFIDENCE_THRESHOLD,
        "confusion_matrix": cm.tolist(),
        "class_names": class_names,
        "gate": {
            "passed": gate_passed,
            "min_accuracy": GATE_MIN_ACCURACY,
            "min_class_recall": GATE_MIN_CLASS_RECALL,
        },
    }


def run_unlabeled(predictor, path: Path, recursive: bool = False):
    class_names = predictor.class_names
    if path.is_file():
        images = [path]
    elif recursive:
        images = sorted(p for p in path.rglob("*") if p.suffix.lower() in IMG_EXTS)
    else:
        images = _list_images(path)
    if not images:
        print(f"  No images found at {path}")
        return

    confs, below = [], 0
    print(f"\n{'image':<40}{'prediction':<22}{'conf':<8}{'2nd guess':<22}{'flag'}")
    print("-" * 100)
    for img in images:
        try:
            allp, top, conf = _predict_one(predictor, img)
        except Exception as e:
            print(f"  [skip] {img.name}: {e}")
            continue
        second = allp[1] if len(allp) > 1 else {"disease": "-", "confidence": 0.0}
        flag = "UNCERTAIN" if conf < CONFIDENCE_THRESHOLD else ""
        confs.append(conf)
        below += int(conf < CONFIDENCE_THRESHOLD)
        print(f"{img.name[:38]:<40}{top:<22}{conf:<8.1%}"
              f"{second['disease']+' '+format(second['confidence'],'.0%'):<22}{flag}")
    if confs:
        confs = np.array(confs)
        print("-" * 100)
        print(f"images={len(confs)}  mean_conf={confs.mean():.1%}  "
              f"uncertain(<{CONFIDENCE_THRESHOLD:.2f})={below} ({below/len(confs):.0%})")
        print("Tip: many UNCERTAIN flags on real photos = the model is out of its "
              "training distribution; collect field images to retrain/augment.")


def main():
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--crop", required=True, choices=list(CROPS.keys()))
    ap.add_argument("--path", required=True, help="image file, folder of images, or labeled root")
    ap.add_argument("--backend", default="keras", choices=["keras", "tflite"],
                    help="keras = full model (most accurate); tflite = mobile model")
    ap.add_argument("--version", default=None, help="model version (default: latest)")
    ap.add_argument("--save-json", action="store_true",
                    help="write external_eval.json into the model version dir "
                         "(labeled mode only); used by the promotion gate")
    args = ap.parse_args()

    path = Path(args.path).expanduser()
    if not path.exists():
        print(f"Path not found: {path}")
        return 2

    print(f"\nLoading {args.crop} model ({args.backend}) ...")
    predictor = _load_predictor(args.crop, args.backend, args.version)
    print(f"  version : {predictor.version}")
    print(f"  classes : {predictor.class_names}")
    print(f"  threshold: {CONFIDENCE_THRESHOLD}")

    # Decide mode: labeled (subfolders match classes) vs unlabeled
    results = None
    if path.is_dir():
        subdirs = [d for d in path.iterdir() if d.is_dir()]
        norm_classes = {_norm(c) for c in predictor.class_names}
        if any(_norm(d.name) in norm_classes for d in subdirs):
            print("\nMode: LABELED EVAL (subfolders matched to classes)")
            results = run_labeled(predictor, path)
        else:
            print("\nMode: UNLABELED PREDICT (loose images)")
            run_unlabeled(predictor, path, recursive=bool(subdirs))
    else:
        print("\nMode: SINGLE IMAGE")
        run_unlabeled(predictor, path)

    if args.save_json:
        if results is None:
            print("\n[save-json] nothing to save (labeled eval did not run)")
        else:
            out_path = predictor.model_dir / predictor.version / "external_eval.json"
            payload = {
                "crop": args.crop,
                "model_version": predictor.version,
                "backend": args.backend,
                "eval_path": str(path),
                "evaluated_at": datetime.now(timezone.utc).isoformat(),
                **results,
            }
            with open(out_path, "w") as f:
                json.dump(payload, f, indent=2)
            print(f"\n[save-json] wrote {out_path}")
            from ml.utils.evaluation import update_metrics_with_external
            if update_metrics_with_external(args.crop, predictor.version):
                print(f"[save-json] updated metrics.json with external_accuracy")
    return 0


if __name__ == "__main__":
    sys.exit(main())