"""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())