scripts/experiments/check_farzone_gates.py

"""Far-zone hand-off 2: the gates that can only be MEASURED on a real clip.

Steps 2, 3 and 6 of design doc 7 section 5 are guarded by unit tests, but
three of the acceptance numbers live in the footage itself and no stub can
produce them. This driver runs the SHIPPED analyze front end
(``poolcoach_cv.broadcast.analyze_clip`` with the resolutions, the
confidence floor and the height-compensated homography ``scripts/cv_worker.py``
configures) over the probe clips of 28/08/2026 and prints the three numbers:

  G1 (step 1)  shot_02: ``balls_init`` must hold a ball within 0.06 m of
               (1.1447, 2.4676) -- ball 5 wedged in the far right pocket --
               at a confidence at or above OP_CONF. 640 does not produce a
               box for it AT ALL and 960 reads it 0.208; only the 1280
               first-frame pass clears the bar (probe T1).
  G2 (step 2)  shot_05: the cue track through [2.30, 2.80] must have no hole
               longer than 0.1 s and must reach y >= 2.45 -- the shot whose
               0.5 s hole made the ball look like it bounced before the
               cushion (probe T2).
  G3 (step 3)  ``rack_guess.on`` true on shot_09 and shot_10, false on
               shot_02, shot_03 and shot_05.

It CHOOSES nothing and tunes nothing: every threshold above comes from the
design doc, and the script only reports PASS/FAIL against them. Steps that
have not landed yet report "absent" rather than failing.

Inputs are the clips the probe already cut with the worker's own
``cut_clip`` (``out_probe_farzone_20260828\\clips\\``), so these frames are
the frames the analyze job read. Nothing is written anywhere except the JSON
report next to them; the archive is never touched.

Run it through ``check_farzone_gates.bat``.
"""

from __future__ import annotations

import importlib.util
import json
import os
import sys
import time
from pathlib import Path

import numpy as np

APP_ROOT = Path(os.environ.get("POOLCOACH_APP_ROOT",
                               r"D:\poolcoach\poolcoach-app"))
CLIPS = Path(os.environ.get(
    "POOLCOACH_GATE_CLIPS",
    r"D:\poolcoach\out_probe_farzone_20260828\clips"))
OUT = Path(os.environ.get("POOLCOACH_GATE_OUT",
                          r"D:\poolcoach\out_probe_farzone_20260828"))
DEVICE = os.environ.get("POOLCOACH_CV_DEVICE", "cpu")

# The same two inputs every regeneration of the demo has used since run A2b
# (14/08): this VOD's corner pockets in the order the front-end asks for.
CORNERS = [[527.5, 869.7], [1380.1, 864.0], [1232.3, 278.5], [673.0, 272.8]]

for _p in (APP_ROOT / "src", APP_ROOT):
    if str(_p) not in sys.path:
        sys.path.insert(0, str(_p))

# ---- G1: ball 5 of shot_02, measured on the frame (probe T1)
BALL5_XY_M = (1.1447, 2.4676)
BALL5_TOL_M = 0.06        # ~2R: two balls cannot both sit inside it

# ---- G2: the cue against the far cushion on shot_05 (probe T2)
CUE_WINDOW_S = (2.30, 2.80)
CUE_MAX_GAP_S = 0.1
CUE_MIN_Y_M = 2.45
CUE_TRACK_MIN_Y_M = 2.00  # in this window the cue is the only thing that
                          # high; the object balls start at y <= 0.91

# ---- G3: which shots are break racks
RACK_ON = (9, 10)
SHOTS = (2, 3, 5, 9, 10)


def load_worker():
    """``scripts/cv_worker.py`` as a module, loaded BY PATH -- the same way
    ``app/localcv.py`` loads it, because ``scripts/`` is not a package.
    Importing rather than copying is what makes this driver measure the
    SHIPPED configuration instead of a second opinion about it."""
    name = "poolcoach_cv_worker"
    if name in sys.modules:
        return sys.modules[name]
    path = APP_ROOT / "scripts" / "cv_worker.py"
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise ImportError(f"could not load {path}")
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod


def analyze(cw, clip: Path) -> dict:
    """One clip through the shipped analyze front end, with the worker's own
    height-compensated homography patched over ``bc.TableHomography`` exactly
    as ``analyze_once`` patches it."""
    from poolcoach_cv import broadcast as bc

    model = analyze.model
    comp: dict = {}
    orig = bc.TableHomography
    bc.TableHomography = cw.comp_homography_cls(cw.clip_frame_wh(clip),
                                                bc.BALL_R_M, comp)
    first_pass = None
    if hasattr(bc, "FirstFramePass"):
        first_pass = bc.FirstFramePass(model, cw.ANALYZE_FIRST_IMGSZ)
    t0 = time.perf_counter()
    try:
        res = bc.analyze_clip(clip, CORNERS, model, op_conf=cw.OP_CONF,
                              imgsz=getattr(cw, "ANALYZE_IMGSZ", cw.IMGSZ),
                              first_pass=first_pass, device=DEVICE)
    finally:
        bc.TableHomography = orig
    res["_elapsed_s"] = round(time.perf_counter() - t0, 2)
    res["_height_comp"] = dict(comp)
    return res


def gate_ball5(res: dict) -> dict:
    """G1 -- is ball 5 in ``balls_init``, and how confidently?"""
    hits = []
    for b in res.get("balls_init") or []:
        d = float(np.hypot(b["x_m"] - BALL5_XY_M[0],
                           b["y_m"] - BALL5_XY_M[1]))
        if d <= BALL5_TOL_M:
            hits.append({"d_m": round(d, 4), "conf": b.get("conf"),
                         "x_m": b["x_m"], "y_m": b["y_m"],
                         "number": b.get("number")})
    hits.sort(key=lambda h: h["d_m"])
    return {"n_balls_init": len(res.get("balls_init") or []),
            "hits": hits, "best": hits[0] if hits else None}


def gate_cue_gap(res: dict) -> dict:
    """G2 -- the largest hole in the cue track inside the window, and how
    close to the far cushion it gets. ``covered`` marks the frames where the
    cue was actually detected; a hole is the time between two covered ones."""
    t_a, t_b = CUE_WINDOW_S
    pts = [(float(p["t_s"]), float(p["y_m"]))
           for p in res.get("track") or []
           if t_a - 1e-9 <= float(p["t_s"]) <= t_b + 1e-9]
    if not pts:
        return {"n_points": 0, "max_gap_s": None, "max_y_m": None}
    gaps = [round(pts[i + 1][0] - pts[i][0], 4) for i in range(len(pts) - 1)]
    high = [y for _t, y in pts if y >= CUE_TRACK_MIN_Y_M]
    return {"n_points": len(pts),
            "max_gap_s": max(gaps) if gaps else 0.0,
            "max_y_m": round(max(y for _t, y in pts), 4),
            "n_above_2m": len(high),
            "t_first": pts[0][0], "t_last": pts[-1][0]}


def gate_rack(res: dict) -> dict:
    """G3 -- the rack_guess block, or "absent" while step 3 has not landed."""
    rg = res.get("rack_guess")
    if rg is None:
        return {"present": False}
    return {"present": True, "on": bool(rg.get("on")),
            "cx_m": rg.get("cx_m"), "cy_m": rg.get("cy_m"),
            "n_seen": rg.get("n_seen")}


def main() -> int:
    cw = load_worker()
    from ultralytics import YOLO

    weights = cw.resolve_weights()
    print(f"[gates] weights  : {weights}")
    print(f"[gates] imgsz    : main "
          f"{getattr(cw, 'ANALYZE_IMGSZ', cw.IMGSZ)}, first frame "
          f"{getattr(cw, 'ANALYZE_FIRST_IMGSZ', None)}, "
          f"segmentation {cw.IMGSZ}")
    print(f"[gates] conf     : op {cw.OP_CONF}, floor "
          f"{getattr(cw, 'DET_CONF_FLOOR', 'n/a')}")
    print(f"[gates] device   : {DEVICE}", flush=True)
    analyze.model = YOLO(str(weights))

    report: dict = {"weights": str(weights), "device": DEVICE,
                    "op_conf": cw.OP_CONF,
                    "imgsz_main": getattr(cw, "ANALYZE_IMGSZ", cw.IMGSZ),
                    "imgsz_first": getattr(cw, "ANALYZE_FIRST_IMGSZ", None),
                    "det_conf_floor": getattr(cw, "DET_CONF_FLOOR", None),
                    "corners": CORNERS, "shots": {}}
    for idx in SHOTS:
        clip = CLIPS / f"shot_{idx:02d}.mp4"
        if not clip.exists():
            print(f"[gates] shot_{idx:02d}: MISSING {clip}", flush=True)
            continue
        res = analyze(cw, clip)
        blk = {"elapsed_s": res["_elapsed_s"],
               "height_comp_on": bool(res["_height_comp"].get("on")),
               "n_ball_tracks": len(res.get("ball_tracks") or []),
               "rack": gate_rack(res)}
        if idx == 2:
            blk["ball5"] = gate_ball5(res)
        if idx == 5:
            blk["cue_gap"] = gate_cue_gap(res)
        report["shots"][str(idx)] = blk
        print(f"[gates] shot_{idx:02d}: {res['_elapsed_s']}s, "
              f"balls_init {len(res.get('balls_init') or [])}, "
              f"tracks {blk['n_ball_tracks']}, rack {blk['rack']}",
              flush=True)

    # ------------------------------------------------------------- verdicts
    lines = []
    b5 = (report["shots"].get("2") or {}).get("ball5")
    if b5 is None:
        lines.append(("G1 ball 5 in balls_init", "NO CLIP", False))
    else:
        best = b5["best"]
        ok = best is not None and float(best["conf"]) >= cw.OP_CONF
        lines.append((f"G1 ball 5 in balls_init (>= {cw.OP_CONF})",
                      "absent" if best is None
                      else f"conf {best['conf']} at {best['d_m']} m", ok))

    cg = (report["shots"].get("5") or {}).get("cue_gap")
    if cg is None or cg["max_gap_s"] is None:
        lines.append(("G2 cue hole in [2.30, 2.80]", "NO CLIP", False))
    else:
        ok = (cg["max_gap_s"] <= CUE_MAX_GAP_S + 1e-9
              and cg["max_y_m"] >= CUE_MIN_Y_M)
        lines.append((f"G2 cue hole <= {CUE_MAX_GAP_S}s and y >= "
                      f"{CUE_MIN_Y_M}",
                      f"gap {cg['max_gap_s']}s, max y {cg['max_y_m']}", ok))

    racks = {int(k): v["rack"] for k, v in report["shots"].items()}
    if not racks or not any(r.get("present") for r in racks.values()):
        lines.append(("G3 rack_guess", "absent (step 3 not landed)", False))
    else:
        want = {i: (i in RACK_ON) for i in racks}
        got = {i: bool(r.get("on")) for i, r in racks.items()}
        ok = want == got
        lines.append(("G3 rack_guess on 09+10 only",
                      ", ".join(f"{i}:{'on' if got[i] else 'off'}"
                                for i in sorted(got)), ok))

    print()
    for name, measured, ok in lines:
        print(f"[gates] {'PASS' if ok else 'FAIL'}  {name}  ->  {measured}")
    report["verdicts"] = [{"gate": n, "measured": m, "pass": ok}
                          for n, m, ok in lines]

    OUT.mkdir(parents=True, exist_ok=True)
    dst = OUT / "farzone_gates.json"
    dst.write_text(json.dumps(report, indent=2, ensure_ascii=False),
                   encoding="utf-8")
    print(f"\n[gates] report -> {dst}", flush=True)
    return 0


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