scripts/eval_synthetic_cd.py

"""Eval run_detection vs auto-generated GT on the synthetic before/after/mask dataset.

Generated by gen_synthetic.py (Priyanka, dataset-prep track): synthetic objects
(roofs/vehicles/vegetation) are composited onto real "before" tiles to make an
"after" image, with the exact pasted region saved as the GT mask - no manual
labeling involved. This script is the training-track's counterpart: point it at
the dataset once a fine-tuned model is available and it reports F1/precision/
recall/IoU the same way eval_drone_gt_packs.py does for the hand-labeled packs.

Usage:
  python scripts/eval_synthetic_cd.py --tag baseline
  python scripts/eval_synthetic_cd.py --tag baseline --limit 200
  python scripts/eval_synthetic_cd.py --dataset-dir data/some_other_triplet_set --tag holdout
"""
from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path

import numpy as np
from PIL import Image

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

try:
    from dotenv import load_dotenv
    load_dotenv(ROOT / ".env")
except ImportError:
    pass

DEFAULT_DATASET_DIR = Path(r"C:\Users\Priyanka\Downloads\Synthetic_CD_dataset")


def _metrics(pred: np.ndarray, gt: np.ndarray) -> dict:
    p = pred.astype(bool).ravel()
    g = gt.astype(bool).ravel()
    tp = int(np.logical_and(p, g).sum())
    fp = int(np.logical_and(p, ~g).sum())
    fn = int(np.logical_and(~p, g).sum())
    prec = tp / (tp + fp) if (tp + fp) else 0.0
    rec = tp / (tp + fn) if (tp + fn) else 0.0
    f1 = (2 * prec * rec / (prec + rec)) if (prec + rec) else 0.0
    iou = tp / (tp + fp + fn) if (tp + fp + fn) else 0.0
    return {
        "f1": round(f1, 4),
        "precision": round(prec, 4),
        "recall": round(rec, 4),
        "iou": round(iou, 4),
        "tp": tp,
        "fp": fp,
        "fn": fn,
        "pred_change_pct": round(100.0 * float(p.mean()), 4),
        "gt_change_pct": round(100.0 * float(g.mean()), 4),
    }


def eval_one(tile_id: str, dataset_dir: Path, method: str, enable_registration: bool = True) -> dict:
    from app.detection_engine import run_detection

    before_p = dataset_dir / "before" / f"{tile_id}.png"
    after_p = dataset_dir / "after" / f"{tile_id}.png"
    gt_p = dataset_dir / "mask" / f"{tile_id}.png"
    if not (before_p.is_file() and after_p.is_file() and gt_p.is_file()):
        return {"tile_id": tile_id, "error": "missing files"}

    before_pil = Image.open(before_p).convert("RGB")
    after_pil = Image.open(after_p).convert("RGB")
    if after_pil.size != before_pil.size:
        after_pil = after_pil.resize(before_pil.size, Image.Resampling.LANCZOS)

    t0 = time.time()
    change_mask, _vis, stats, regions = run_detection(
        before_pil,
        after_pil,
        method=method,
        enable_registration=enable_registration,
        enable_normalization=True,
        detection_sensitivity=0.5,
        min_region_area=150,
        max_size=max(before_pil.size),
        before_path=None,
        after_path=None,
    )
    elapsed = time.time() - t0

    gt = np.array(Image.open(gt_p).convert("L")) > 127
    pred = np.asarray(change_mask)
    if pred.ndim == 3:
        pred = pred[..., 0]
    pred = pred > 127
    if pred.shape != gt.shape:
        gt_img = Image.fromarray((gt.astype(np.uint8) * 255))
        gt_img = gt_img.resize((pred.shape[1], pred.shape[0]), Image.Resampling.NEAREST)
        gt = np.array(gt_img) > 127

    m = _metrics(pred, gt)
    return {
        "tile_id": tile_id,
        "elapsed_s": round(elapsed, 2),
        "n_regions": len(regions or []),
        "report_change_pct": round(float(stats.get("change_percentage") or 0.0), 4),
        **m,
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--tag", default="eval")
    ap.add_argument("--dataset-dir", default=str(DEFAULT_DATASET_DIR))
    ap.add_argument("--out-dir", default="data/synthetic_cd_eval")
    ap.add_argument("--limit", type=int, default=None, help="Evaluate only the first N tiles (quick smoke test)")
    ap.add_argument("--method", default="AI-Based Deep Learning",
                     help="Detection method passed to run_detection, e.g. 'Feature-Based' "
                          "(CPU-only, no torch) or 'AI-Based Deep Learning' (needs GPU/torch)")
    ap.add_argument("--no-register", action="store_true")
    args = ap.parse_args()

    dataset_dir = Path(args.dataset_dir)
    before_dir = dataset_dir / "before"
    if not before_dir.is_dir():
        print(f"ERROR: {before_dir} not found", flush=True)
        sys.exit(1)

    tile_ids = sorted(p.stem for p in before_dir.glob("*.png"))
    if args.limit:
        tile_ids = tile_ids[: args.limit]
    print(f"Evaluating {len(tile_ids)} tiles from {dataset_dir}", flush=True)

    out_dir = ROOT / args.out_dir
    out_dir.mkdir(parents=True, exist_ok=True)

    rows = []
    for i, tid in enumerate(tile_ids):
        row = eval_one(tid, dataset_dir, method=args.method, enable_registration=not args.no_register)
        rows.append(row)
        if "error" in row:
            print(f"  [{i+1}/{len(tile_ids)}] {tid}: ERROR {row['error']}", flush=True)
        else:
            print(
                f"  [{i+1}/{len(tile_ids)}] {tid}: F1={row['f1']:.3f} P={row['precision']:.3f} "
                f"R={row['recall']:.3f} IoU={row['iou']:.3f}",
                flush=True,
            )

    ok = [r for r in rows if "f1" in r]
    summary = {
        "tag": args.tag,
        "dataset_dir": str(dataset_dir),
        "n_tiles": len(tile_ids),
        "n_ok": len(ok),
        "created_unix": time.time(),
        "mean_f1": round(float(np.mean([r["f1"] for r in ok])), 4) if ok else 0.0,
        "mean_precision": round(float(np.mean([r["precision"] for r in ok])), 4) if ok else 0.0,
        "mean_recall": round(float(np.mean([r["recall"] for r in ok])), 4) if ok else 0.0,
        "mean_iou": round(float(np.mean([r["iou"] for r in ok])), 4) if ok else 0.0,
        "tiles": rows,
    }
    out = out_dir / f"{args.tag}.json"
    out.write_text(json.dumps(summary, indent=2), encoding="utf-8")
    print(
        f"\nSaved {out} | mean_F1={summary['mean_f1']:.4f} "
        f"P={summary['mean_precision']:.3f} R={summary['mean_recall']:.3f} IoU={summary['mean_iou']:.3f}",
        flush=True,
    )


if __name__ == "__main__":
    main()