scripts/cv/eval_deploy_pr.py

"""P/R/F1 cua PREDICT-pipeline tai conf VAN HANH — "so luc deploy" (ban giao 19, 05/08/2026).

Do dung nhung gi CV worker deploy (scripts/cv_worker.py) lam voi MOT anh,
tren 49 anh val pix2pockets:

  1. resize max-dim 1280 INTER_AREA (chep nguyen block cua ``scan_once`` —
     anh val 1920x1080 va 3360x2100 deu vuot tran nen buoc nay CO tac dung),
  2. ``model.predict(conf=<opconf.json>, imgsz=640)`` — NMS
     ``multi_label=False`` (chi class argmax), dung pipeline worker goi,

roi merge any-ball (Black+Cue+Solid+Striped -> "ball"; Dot LOAI khoi ca
pred lan GT — khong phai bi) va match GREEDY voi GT: duyet pred theo conf
GIAM DAN, moi pred lay GT chua-match co IoU cao nhat >= 0.5; pred khong
match = FP, GT thua = FN -> precision / recall / F1 toan split.

MATCHER TU VIET LA DUNG O DAY (khac ban giao 17): lan do can identity voi
val-pipeline nen matcher port bi loai; lan nay muc tieu la do THUOC DEPLOY
— khong co reference nao de identity, chi can khai phuong phap ro rang.

HAI CAI THUOC (bai hoc ban giao 17): so o day la P/R/F1 tai MOT diem conf
tren PREDICT-pipeline. KHONG phai AP, KHONG duoc so voi any-ball AP50
0.9383 (val-pipeline, NMS multi_label=True). Cap so hop le duy nhat:
P/R/F1@conf cua val-pipeline (opconf.json) vs P/R/F1@conf o day — ky vong
predict-pipeline THAP hon (multi_label=False mat cac box "duoc cuu"; Cue
tung lech -0.20 AP giua hai thuoc).

KHONG tai lap cac khau SAU predict cua worker (homography, kep mep, dedupe
tam ban, cue-dung-1): chung can corners + toa do ban, con GT o day la bbox
pixel — do o tang bbox, TRUOC homography.

Guard (BRIEF "Neu bi"): P hoac R < 0.5 -> nghi sai matcher/resize, DUNG
khong ghi so. Neo them: tong GT ball tu file nhan phai khop counters
``gt_kept`` cua opconf.json (563) — lech la parse nhan sai, DUNG.

Chay tren venv CV (can ultralytics; GPU nhu worker):

    python scripts/cv/eval_deploy_pr.py --artifact-dir "D:/Khoa luan/cv_full_20260805"

Console ASCII-only (bay cp1252). Ket qua ghi ``opconf_predict.json`` vao
artifact dir, canh ``opconf.json`` de dat hai thuoc canh nhau.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]  # poolcoach-rl/
sys.path.insert(0, str(ROOT / "src"))

DATA_YAML = ROOT / "datasets" / "pix2pockets" / "yolo" / "data.yaml"

BALL_NAMES = ("Black", "Cue", "Solid", "Striped")
EXCLUDE_NAME = "Dot"
IMGSZ = 640      # = cv_worker.IMGSZ — cung imgsz train/val/worker
MAX_DIM = 1280   # = cv_worker.MAX_DIM
IOU_THR = 0.5


def resize_like_worker(img, max_dim: int = MAX_DIM):
    """Chep NGUYEN block resize cua cv_worker.scan_once — doi o do thi doi o day."""
    import cv2

    h, w = img.shape[:2]
    scale = 1.0
    if max(h, w) > max_dim:
        scale = max_dim / max(h, w)
        img = cv2.resize(img, (round(w * scale), round(h * scale)),
                         interpolation=cv2.INTER_AREA)
    return img, scale


def load_gt_boxes(lbl_path: Path, w: int, h: int,
                  ball_ids: set[int]) -> tuple[list[list[float]], int]:
    """Nhan YOLO txt (class cx cy bw bh, normalized) -> list xyxy pixel tren
    anh DA resize (cung he voi pred). Tra (boxes_ball, n_dot_dropped)."""
    boxes: list[list[float]] = []
    n_dot = 0
    if not lbl_path.exists():
        return boxes, n_dot
    for line in lbl_path.read_text(encoding="utf-8").splitlines():
        parts = line.split()
        if not parts:
            continue
        cid = int(parts[0])
        if cid not in ball_ids:
            n_dot += 1
            continue
        cx, cy, bw, bh = (float(v) for v in parts[1:5])
        boxes.append([(cx - bw / 2) * w, (cy - bh / 2) * h,
                      (cx + bw / 2) * w, (cy + bh / 2) * h])
    return boxes, n_dot


def iou_xyxy(a: list[float], b: list[float]) -> float:
    ix1, iy1 = max(a[0], b[0]), max(a[1], b[1])
    ix2, iy2 = min(a[2], b[2]), min(a[3], b[3])
    iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1)
    inter = iw * ih
    if inter <= 0.0:
        return 0.0
    area_a = (a[2] - a[0]) * (a[3] - a[1])
    area_b = (b[2] - b[0]) * (b[3] - b[1])
    return inter / (area_a + area_b - inter)


def greedy_match(preds: list[tuple[float, list[float]]],
                 gts: list[list[float]],
                 thr: float = IOU_THR) -> tuple[int, int, int]:
    """Match 1 anh: pred theo conf GIAM DAN, moi pred lay GT chua-match co
    IoU cao nhat >= thr. Tra (tp, fp, fn). Mot GT match toi da MOT lan —
    double-detection cung mot bi thanh FP, trung thuc voi detector."""
    used = [False] * len(gts)
    tp = fp = 0
    for _conf, pb in sorted(preds, key=lambda t: -t[0]):
        best_i, best_iou = -1, thr
        for i, gb in enumerate(gts):
            if used[i]:
                continue
            v = iou_xyxy(pb, gb)
            if v >= best_iou:
                best_i, best_iou = i, v
        if best_i >= 0:
            used[best_i] = True
            tp += 1
        else:
            fp += 1
    fn = used.count(False)
    return tp, fp, fn


def main() -> None:
    ap = argparse.ArgumentParser(
        description="P/R/F1 predict-pipeline (deploy ruler) at operating conf")
    ap.add_argument("--artifact-dir", type=Path, required=True,
                    help="chua opconf.json (doc conf + weights); ghi opconf_predict.json")
    ap.add_argument("--weights", type=Path, default=None,
                    help="mac dinh: truong weights cua opconf.json")
    ap.add_argument("--conf", type=float, default=None,
                    help="mac dinh: truong conf cua opconf.json")
    ap.add_argument("--device", default=None,
                    help="mac dinh: khong truyen (model tu chon, nhu worker)")
    args = ap.parse_args()

    import torch
    import yaml
    import cv2
    from ultralytics import YOLO

    opconf = json.loads((args.artifact_dir / "opconf.json").read_text(encoding="utf-8"))
    conf = args.conf if args.conf is not None else float(opconf["conf"])
    weights = args.weights or Path(opconf["weights"])
    predict_kw = {"device": args.device} if args.device else {}
    print(f"[env] torch={torch.__version__} cuda_available={torch.cuda.is_available()}"
          f" conf={conf:g} weights={weights}")

    data = yaml.safe_load(DATA_YAML.read_text(encoding="utf-8"))
    names = {i: n for i, n in enumerate(data["names"])}
    name_to_id = {n: i for i, n in names.items()}
    missing = [n for n in (*BALL_NAMES, EXCLUDE_NAME) if n not in name_to_id]
    if missing:
        sys.exit(f"[FAIL] data.yaml thieu class {missing}")
    ball_ids = {name_to_id[n] for n in BALL_NAMES}
    img_dir = Path(data["path"]) / data["val"]
    lbl_dir = Path(str(img_dir).replace("images", "labels"))
    imgs = sorted(img_dir.glob("*.jpg"))
    if not imgs:
        sys.exit(f"[FAIL] khong thay anh val o {img_dir}")

    model = YOLO(str(weights))
    if dict(model.names) != names:
        sys.exit(f"[ERROR] names checkpoint {model.names} khac data.yaml {names}")

    tot = {"tp": 0, "fp": 0, "fn": 0, "gt_ball": 0, "gt_dot": 0,
           "pred_ball": 0, "pred_dot": 0}
    for p in imgs:
        img = cv2.imread(str(p))
        if img is None:
            sys.exit(f"[FAIL] khong doc duoc anh {p}")
        img, _scale = resize_like_worker(img)
        h, w = img.shape[:2]
        res = model.predict(img, conf=conf, imgsz=IMGSZ, verbose=False,
                            **predict_kw)[0]
        preds: list[tuple[float, list[float]]] = []
        for b in res.boxes:
            if names[int(b.cls)] not in BALL_NAMES:
                tot["pred_dot"] += 1
                continue
            preds.append((float(b.conf), [float(v) for v in b.xyxy[0]]))
        gts, n_dot = load_gt_boxes(
            lbl_dir / (p.stem + ".txt"), w, h, ball_ids)
        tp, fp, fn = greedy_match(preds, gts)
        tot["tp"] += tp
        tot["fp"] += fp
        tot["fn"] += fn
        tot["gt_ball"] += len(gts)
        tot["gt_dot"] += n_dot
        tot["pred_ball"] += len(preds)

    # --- neo: tong GT ball tu file nhan phai khop validator da nghiem thu ---
    ref_gt = opconf.get("counters", {}).get("gt_kept")
    if ref_gt is not None and tot["gt_ball"] != ref_gt:
        sys.exit(f"[FAIL] GT ball tu file nhan = {tot['gt_ball']} khac "
                 f"counters.gt_kept = {ref_gt} cua opconf.json - parse nhan "
                 f"sai, DUNG khong ghi so.")

    prec = tot["tp"] / (tot["tp"] + tot["fp"]) if tot["tp"] + tot["fp"] else 0.0
    rec = tot["tp"] / (tot["tp"] + tot["fn"]) if tot["tp"] + tot["fn"] else 0.0
    f1 = 2 * prec * rec / (prec + rec) if prec + rec else 0.0

    print(f"[deploy-pr] {len(imgs)} anh val; GT ball {tot['gt_ball']} "
          f"(Dot loai {tot['gt_dot']}); pred ball {tot['pred_ball']} "
          f"(pred Dot bo {tot['pred_dot']})")
    print(f"[deploy-pr] TP {tot['tp']}  FP {tot['fp']}  FN {tot['fn']}")
    print(f"[deploy-pr] conf {conf:g}: P {prec:.4f}  R {rec:.4f}  F1 {f1:.4f}"
          f"  (predict-pipeline)")
    print(f"[deploy-pr] val-pipeline cung conf (opconf.json): "
          f"P {opconf['precision']:.4f}  R {opconf['recall']:.4f}  "
          f"F1 {opconf['f1']:.4f}  - HAI CAI THUOC, ky vong predict thap hon")

    # BRIEF "Neu bi": so vo ly la dau hieu sai matcher/resize -> DUNG, bao lai
    if prec < 0.5 or rec < 0.5:
        sys.exit(f"[FAIL] P {prec:.4f} / R {rec:.4f} < 0.5 - nghi sai "
                 f"matcher/resize, DUNG khong ghi opconf_predict.json, "
                 f"bao lai Cowork (BRIEF).")

    out = {
        "date": "2026-08-05",
        "weights": str(weights),
        "split": "val",
        "conf": conf,
        "method": (
            "worker predict-pipeline: resize max-dim 1280 INTER_AREA (chep "
            "block scan_once cua cv_worker.py) + model.predict imgsz=640 "
            "NMS multi_label=False tai conf van hanh; any-ball merge "
            "(Black+Cue+Solid+Striped -> ball), Dot loai ca pred lan GT; "
            "greedy match theo conf giam dan, IoU >= 0.5, moi GT match "
            "toi da 1 lan. Day la P/R/F1 tai MOT diem conf - KHONG phai "
            "AP, KHONG so voi anyball AP50 val-pipeline (hai cai thuoc, "
            "ban giao 17). Khong tai lap cac khau sau predict cua worker "
            "(homography, dedupe tam ban, cue-dung-1)."),
        "predict_pipeline": {"precision": round(prec, 4),
                             "recall": round(rec, 4),
                             "f1": round(f1, 4),
                             "tp": tot["tp"], "fp": tot["fp"], "fn": tot["fn"]},
        "val_pipeline_ref": {"precision": opconf["precision"],
                             "recall": opconf["recall"],
                             "f1": opconf["f1"],
                             "source": "opconf.json (cung conf, val-pipeline)"},
        "counters": {"n_images": len(imgs),
                     "gt_ball": tot["gt_ball"],
                     "gt_dot_excluded": tot["gt_dot"],
                     "pred_ball": tot["pred_ball"],
                     "pred_dot_dropped": tot["pred_dot"]},
        "env": {"torch": torch.__version__,
                "cuda_available": torch.cuda.is_available(),
                "device_arg": args.device},
    }
    out_path = args.artifact_dir / "opconf_predict.json"
    out_path.write_text(json.dumps(out, indent=2), encoding="utf-8")
    print(f"[artifact] -> {out_path}")


if __name__ == "__main__":
    main()