ml/scripts/eval_scene_gate.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
eval_scene_gate.py — Offline-Messung des Phase-A-Szenen-Gates (Stufe A0).

Misst die zwei Zahlen, die über die Live-Schaltung entscheiden
(docs/PHASE_A_SCENE_GATE.md §5):

  * FP-Kill-Rate  – Anteil der NICHT-Müll-Bilder (Natur/leerer Boden/Tier/Möbel),
                    bei denen das Gate korrekt KEIN Müll meldet. Hoch = gut.
  * TP-Retention  – Anteil der ECHTEN Müll-Bilder, die das Gate BEHÄLT
                    (nicht fälschlich unterdrückt). Muss ~1.0 bleiben — ein
                    False Negative = ehrlicher Nutzer bekommt 0 TC.

Eine Szene gilt bei Schwelle τ als "Müll", wenn waste_likelihood >= τ.
Wir sweepen τ und suchen den Betriebspunkt, der das Gate-Kriterium erfüllt
(Default: FP-Kill >= 0.60 BEI TP-Retention >= 0.99).

Nutzung:
    python -m ml.scripts.eval_scene_gate \
        --positives data/scene_eval/waste \
        --negatives data/scene_eval/non_waste \
        --out artifacts/summary_scene_gate.json

Ergebnis: JSON neben summary_eval.json (gleiche Konvention wie evaluate.py),
fließt als Snapshot in die Chronik.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Dict, List

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


def list_images(d: Path, limit: int = 0) -> List[Path]:
    if not d or not d.exists():
        return []
    imgs = sorted(p for p in d.rglob("*") if p.suffix.lower() in IMG_EXT)
    return imgs[:limit] if limit else imgs


def score_folder(gate, imgs: List[Path]) -> List[float]:
    """waste_likelihood je Bild."""
    out: List[float] = []
    for i, p in enumerate(imgs, 1):
        try:
            scene = gate.analyze(str(p))
            out.append(float(scene["waste_likelihood"]))
        except Exception as e:  # ein kaputtes Bild darf den Lauf nicht killen
            print(f"[WARN] {p.name}: {type(e).__name__}: {str(e)[:120]}", file=sys.stderr)
            out.append(0.0)
        if i % 25 == 0:
            print(f"  … {i}/{len(imgs)}", file=sys.stderr)
    return out


def sweep(pos: List[float], neg: List[float],
          min_fp_kill: float, min_tp_retention: float) -> Dict:
    """τ-Sweep; findet den besten zulässigen Betriebspunkt."""
    thresholds = [round(t / 100, 2) for t in range(0, 101, 2)]
    curve = []
    best = None
    for tau in thresholds:
        tp_ret = sum(1 for s in pos if s >= tau) / len(pos) if pos else None
        fp_kill = sum(1 for s in neg if s < tau) / len(neg) if neg else None
        point = {"tau": tau, "tp_retention": tp_ret, "fp_kill_rate": fp_kill}
        curve.append(point)
        # zulässig = TP-Retention-Schmerzgrenze gehalten; darunter maximiere FP-Kill
        if (tp_ret is not None and fp_kill is not None
                and tp_ret >= min_tp_retention):
            if best is None or fp_kill > best["fp_kill_rate"]:
                best = point
    passed = bool(best is not None and best["fp_kill_rate"] >= min_fp_kill)
    return {"curve": curve, "recommended_operating_point": best, "gate_passed": passed}


def main() -> int:
    ap = argparse.ArgumentParser(description="Phase-A Szenen-Gate — Offline-Eval (A0)")
    ap.add_argument("--positives", type=Path, required=True, help="Ordner mit echten Müll-Bildern")
    ap.add_argument("--negatives", type=Path, required=True, help="Ordner mit Nicht-Müll/leer-Bildern")
    ap.add_argument("--out", type=Path, default=Path("artifacts/summary_scene_gate.json"))
    ap.add_argument("--weights", default="yolov8s-worldv2.pt")
    ap.add_argument("--conf", type=float, default=0.10)
    ap.add_argument("--limit", type=int, default=0, help="max Bilder je Ordner (0=alle)")
    ap.add_argument("--min-fp-kill", type=float, default=0.60)
    ap.add_argument("--min-tp-retention", type=float, default=0.99)
    args = ap.parse_args()

    pos_imgs = list_images(args.positives, args.limit)
    neg_imgs = list_images(args.negatives, args.limit)
    if not pos_imgs or not neg_imgs:
        print(f"[ERROR] Brauche Bilder in beiden Ordnern. "
              f"positives={len(pos_imgs)} negatives={len(neg_imgs)}", file=sys.stderr)
        return 2

    print(f"Positives: {len(pos_imgs)} · Negatives: {len(neg_imgs)} · weights={args.weights}",
          file=sys.stderr)

    # Modell erst hier laden (schwer) — so bleibt --help / Argparse leichtgewichtig.
    from ml.scene_gate import SceneGate, GATE_VERSION
    gate = SceneGate(weights=args.weights, conf_thr=args.conf)

    print("Scoring positives …", file=sys.stderr)
    pos = score_folder(gate, pos_imgs)
    print("Scoring negatives …", file=sys.stderr)
    neg = score_folder(gate, neg_imgs)

    result = sweep(pos, neg, args.min_fp_kill, args.min_tp_retention)
    summary = {
        "gate_version": GATE_VERSION,
        "weights": args.weights,
        "conf_thr": args.conf,
        "n_positives": len(pos_imgs),
        "n_negatives": len(neg_imgs),
        "criterion": {"min_fp_kill": args.min_fp_kill, "min_tp_retention": args.min_tp_retention},
        **result,
    }
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")

    op = result["recommended_operating_point"]
    print("\n===== Phase-A Szenen-Gate — Ergebnis =====")
    if op:
        print(f"  Betriebspunkt τ={op['tau']}: "
              f"TP-Retention={op['tp_retention']:.3f} · FP-Kill={op['fp_kill_rate']:.3f}")
    print(f"  Gate-Kriterium erfüllt: {'JA ✅' if result['gate_passed'] else 'NEIN ❌'}")
    print(f"  -> {args.out}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())