"""Far-zone detection probe -- design doc 7_FarZone_Detection_BallID.md section 3.1, step P of BRIEF 28/08/2026. Called by ``probe_farzone_det.bat``, which owns the environment. WHY THIS RUN HAPPENS. Three of the errors Danh reported on the packaged demo (`a1cd87e8`) share one root: the detector loses balls at the far end of the table. The camera decoded from that demo's own homography (h = 2.363 m, f = 2155 px) puts a ball at 37 px across on the near rail and 22 px on the far rail of a 1920x1080 frame; YOLO at ``imgsz=640`` shrinks the frame 3x, so a far ball is about 7 px. This script MEASURES that -- it does not fix anything and it does not choose a configuration. The decision rule lives in section 3.1 of the design doc and Cowork applies it after reading these numbers. WHAT IT MEASURES. Four detector configurations full640 full frame, imgsz 640 (the shipped baseline) full960 full frame, imgsz 960 full1280 full frame, imgsz 1280 band full frame at 640 PLUS a second pass over the far half of the table cropped at NATIVE resolution, the two merged and de-duplicated with ``BALL_DEDUP_M`` over five target situations taken from the demo bundle, all at ``conf=0.05`` (the floor of design section 3.2, so a detection weaker than the operating point still shows up as a number instead of vanishing): T1 shot_02 frame 0 -- ball 5, wedged in the mouth of the FAR RIGHT corner pocket, missing from ``balls_init`` in the demo T2 shot_05 t in [2.30, 2.80] -- the cue ball against the far cushion, the 0.5 s hole in the demo's cue track T3 shot_03 t in [1.30, 1.70] -- the object ball running the last 23.7 cm into the far right pocket, where the demo's track dies T4 clips 02 and 05, every frame -- detections per frame in the far band (y_m > 1.9): the BACKGROUND number the three targets have to be read against T5 shot_09 + shot_10 frame 0 -- failure mode 2: the rack seen down its own axis, balls hiding behind balls. Not a far-zone case (the rack sits at y ~ 0.52-0.67, balls ~35 px); it is here to measure the CEILING of what NMS can return from a pile, by sweeping ``iou`` {0.5, 0.7, 0.9} x ``imgsz`` {640, 1280} For every target detection the confidence is logged TWICE: once over all detections and once over only those that survive the ``IN_TOL_M`` table bounds gate. That is the whole point of the split -- a ball in the mouth of a pocket projects to the edge of the table, and "the gate ate it" had to be ruled out as an explanation before the resolution one is believed. HOW IT STAYS HONEST. - It runs the SHIPPED chain, not a copy: the clips are cut by ``cv_worker.cut_clip`` with the same ffmpeg command and the same boundaries the demo used, and the box centres are mapped by the same height-compensated homography (``cv_worker.comp_homography_cls``) that the analyze job builds. A REPRODUCTION CHECK confirms it: full640 re-run at the operating confidence rebuilds ``balls_init`` of shot_02 exactly as the demo bundle carries it, ball 5 missing included. If that check fails, nothing below can be trusted, so it is reported first. - The ball-5 ground truth is a MEASURED pixel position, not a detection: (1181.5, 280.0) is the centroid of the purple pixels of that ball on frame 0 (OpenCV hue 120-160, S >= 60, V >= 60 over the window x 1120-1250, y 240-330), whose visible bounding box is 28 x 22 px -- the 22 px the design doc predicts for the far rail. Through the compensated homography that lands at table (1.145, 2.475), which is INSIDE the table and therefore not a candidate for being cut by the bounds gate. - The COST column comes from its own benchmark, not from the measurement passes. The passes run over half an hour on a machine doing other things, and their averages are incoherent as a ratio (960 measured cheaper than 640 on one clip). ``cost_benchmark`` decodes one set of frames, warms every configuration at its own ``imgsz``, times them back to back and keeps the best of three. Decision rule (c) reads that column; the contaminated per-clip averages are kept beside it as a witness. - A detection PAST the far rail projects to a table y of 4 m and more, which is precisely what the ``IN_TOL_M`` gate exists to throw away. Distances to a pocket are therefore measured from the last IN-BOUNDS detection, with the out-of-bounds one reported separately rather than silently mixed in. - The archive is read-only. ``D:\\Khoa luan`` is fingerprinted (every file's size and mtime) before and after the run and the two are compared; a single changed byte is a red gate. Output in ``D:\\poolcoach\\out_probe_farzone_20260828\\``: ``probe_farzone.json`` (every number, machine readable) and ``probe_farzone.md`` (the tables a human reads). The cut clips stay in ``clips\\`` of that same directory -- outside git, on the same machine as the archive VOD they come from. """ from __future__ import annotations import importlib.util import json import math import os import statistics 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")) VOD = Path(os.environ["POOLCOACH_PROBE_VOD"]) OUT = Path(os.environ["POOLCOACH_PROBE_OUT"]) BUNDLE = Path(os.environ.get( "POOLCOACH_PROBE_BUNDLE", APP_ROOT / "app" / "demo_analyzer" / "a1cd87e813de422a868ae8eedf9c4969")) DEVICE = os.environ.get("POOLCOACH_CV_DEVICE", "cpu") ARCHIVE_ROOT = Path(os.environ.get("POOLCOACH_ARCHIVE_ROOT", r"D:\Khoa luan")) # The 4 corner pockets of run A2b (14/08) and every regeneration since, in # the order the front-end asks for them -- the same two inputs (VOD + # corners) the demo bundle was produced from. 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)) # --------------------------------------------------------------- constants DET_CONF_FLOOR = 0.05 # design section 3.2: the floor every configuration # predicts at here, so a detection below the # operating point is a NUMBER instead of a silence NMS_IOU = 0.7 # the ultralytics default, stated explicitly so the # rack sweep has a named centre to sweep around FAR_Y_M = 1.9 # "the far band" for the background count (T4) -- # design section 3.1 # T1 -- ball 5 on shot_02 frame 0. A MEASURED pixel centroid (see the module # docstring), not a detection, so a configuration that misses the ball still # has something to be measured against. GT_BALL5_PX = (1181.5, 280.0) TARGET_MATCH_M = 0.06 # ~2R: a detection this close to the ground truth # IS that ball. Two balls cannot both be inside it # (centre to centre at contact is 2R = 57 mm) # T2 -- the cue against the far cushion. The demo's cue track jumps # 2.300 -> 2.800 with the ball at y ~ 2.28-2.31 and the cushion at y ~ 2.51. CUE_WINDOW_S = (2.30, 2.80) CUE_MIN_Y_M = 2.00 # in this window the cue is the ONLY thing above # y = 2.0 (shot_05 starts with the 8 at y = 0.91 # and the 1 at y = 0.58) # T3 -- shot_03's object ball over the last 23.7 cm into the far right # pocket. The demo's track ends at t = 1.300, (1.217, 2.309), still moving # at ~1.35 m/s, so the ball reaches the pocket around t = 1.47. S03_WINDOW_S = (1.30, 1.70) S03_MIN_Y_M = 2.25 S03_MIN_X_M = 1.00 # the corridor into the far right pocket, wide # enough for the position noise and narrow enough # that no other ball of that shot can enter it POCKET_FAR_RIGHT = (1.27, 2.54) # (TABLE_W_M, TABLE_L_M) -- the pocket # shot_03's ball is running at, and the one ball 5 # of shot_02 is wedged in # The far-band second pass of the "band" configuration. BAND_Y0_M = 1.20 # crop the table from just short of half way to the # far rail -- slightly more than "the far half" so # a ball straddling y = 1.27 is whole in the crop BAND_MARGIN_PX = 40 # pixel margin around the projected region, so a # ball ON the far rail is not cut in half BAND_STRIDE = 32 # YOLO's stride: the band's imgsz is rounded UP to # a multiple of it, which keeps the crop at native # scale or larger -- never downscaled, since not # downscaling is the entire point of the pass # T5 -- the rack. RACK_R_M = 0.15 # design section 3.4 ``RACK_R_M``: the radius # around the cluster centre that counts as "in the # rack". A 9-ball diamond spans ~2 ball diameters # from centre to tip (~0.12 m), so 0.15 holds the # whole rack and nothing else of those two frames RACK_IOU = (0.5, 0.7, 0.9) RACK_IMGSZ = (640, 1280) CONFIGS = [ {"name": "full640", "imgsz": 640, "band": False}, {"name": "full960", "imgsz": 960, "band": False}, {"name": "full1280", "imgsz": 1280, "band": False}, {"name": "band", "imgsz": 640, "band": True}, ] CLIPS_NEEDED = (2, 3, 5, 9, 10) # ----------------------------------------------------------------- helpers 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. Only four things are used from it (``cut_clip``, ``clip_frame_wh``, ``comp_homography_cls``, ``OP_CONF``); importing rather than copying is what makes this probe measure the SHIPPED chain.""" 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 fingerprint(root: Path) -> dict: """Every file under ``root`` as name -> (size, mtime_ns). The archive is READ ONLY: this is taken before and after the run and compared, so "nothing was written to D:\\Khoa luan" is a measurement rather than a promise.""" out = {} for p in sorted(root.rglob("*")): try: st = p.stat() except OSError: continue if p.is_file(): out[str(p.relative_to(root))] = [st.st_size, st.st_mtime_ns] return out def cut_clips(cw, shots: list[dict]) -> dict[int, Path]: """The per-shot clips of the five target shots, cut by the worker's own ``cut_clip`` from the boundaries the demo manifest carries -- the same ffmpeg command, the same seek, the same re-encode, so the frames the probe reads are the frames the analyze job read.""" clips_dir = OUT / "clips" clips_dir.mkdir(parents=True, exist_ok=True) out: dict[int, Path] = {} for s in shots: idx = int(s["idx"]) if idx not in CLIPS_NEEDED: continue dst = clips_dir / f"shot_{idx:02d}.mp4" t0 = float(s["t_start_s"]) dur = float(s["t_end_s"]) - t0 err = cw.cut_clip(VOD, t0, dur, dst) if err is not None: raise RuntimeError(f"shot {idx}: {err}") out[idx] = dst print(f"[probe] cut shot_{idx:02d} t0={t0:.3f} dur={dur:.3f} " f"-> {dst.name} ({dst.stat().st_size / 1e6:.1f} MB)", flush=True) return out def homography_for(cw, clip_path: Path): """The clip's height-compensated homography plus its ``height_comp`` metadata -- rebuilt exactly as ``analyze_once`` builds it (the decomposition is deterministic, so the same clip gives the same map).""" from poolcoach_cv import broadcast as bc from poolcoach_cv.homography import orient_corners sink: dict = {} cls = cw.comp_homography_cls(cw.clip_frame_wh(clip_path), bc.BALL_R_M, sink) th = cls(orient_corners(np.asarray(CORNERS, dtype=np.float64)), table_w=bc.TABLE_W_M, table_l=bc.TABLE_L_M) return th, dict(sink) def band_window(th, frame_w: int, frame_h: int) -> dict: """The pixel window of the far half of the table, plus the ``imgsz`` the band pass runs at: the crop's longest side rounded UP to a multiple of the stride, so ultralytics never SHRINKS the crop.""" from poolcoach_cv import broadcast as bc w, ln = bc.TABLE_W_M, bc.TABLE_L_M pts = [th.table_to_px(p) for p in ((0.0, BAND_Y0_M), (w, BAND_Y0_M), (w, ln), (0.0, ln))] xs = [float(p[0]) for p in pts] ys = [float(p[1]) for p in pts] x0 = max(0, int(math.floor(min(xs))) - BAND_MARGIN_PX) x1 = min(frame_w, int(math.ceil(max(xs))) + BAND_MARGIN_PX) y0 = max(0, int(math.floor(min(ys))) - BAND_MARGIN_PX) y1 = min(frame_h, int(math.ceil(max(ys))) + BAND_MARGIN_PX) side = max(x1 - x0, y1 - y0) imgsz = int(math.ceil(side / BAND_STRIDE) * BAND_STRIDE) return {"box": [x0, y0, x1, y1], "w": x1 - x0, "h": y1 - y0, "imgsz": imgsz, "scale": round(imgsz / max(side, 1), 3)} def boxes_to_dets(res, th, offset=(0, 0), source="full"): """One ultralytics result -> the detections this probe records. ``in_tol`` is the ``IN_TOL_M`` table bounds gate of ``analyze_clip``: kept as a FLAG instead of a filter, which is what lets the report show the confidence before and after the gate in separate columns. """ from poolcoach_cv import broadcast as bc ox, oy = offset out = [] names_map = res.names for b in res.boxes: cls = names_map[int(b.cls)] if cls not in bc.BALL_CLASSES: continue bx1, by1, bx2, by2 = (float(v) for v in b.xyxy[0]) bx1, bx2 = bx1 + ox, bx2 + ox by1, by2 = by1 + oy, by2 + oy cx, cy = (bx1 + bx2) / 2.0, (by1 + by2) / 2.0 tx, ty = th.px_to_table((cx, cy)) in_tol = (-bc.IN_TOL_M <= tx <= bc.TABLE_W_M + bc.IN_TOL_M and -bc.IN_TOL_M <= ty <= bc.TABLE_L_M + bc.IN_TOL_M) # conf is kept at FULL precision: the reproduction check rounds it to # 3 exactly once, the way ``balls_init`` does. Rounding here first # and again there turns 0.89649 into 0.897 where the pipeline says # 0.896, and the check would fail on arithmetic instead of on optics. out.append({"cls": cls, "conf": float(b.conf), "x_m": round(float(tx), 4), "y_m": round(float(ty), 4), "px": [round(cx, 1), round(cy, 1)], "w_px": round(bx2 - bx1, 1), "h_px": round(by2 - by1, 1), "in_tol": bool(in_tol), "source": source}) return out def merge_band(main: list[dict], band: list[dict]) -> list[dict]: """Main-pass detections plus the band ones that are not already there -- ``BALL_DEDUP_M`` on the TABLE plane, the same rule ``balls_init`` uses for double detections.""" from poolcoach_cv import broadcast as bc out = list(main) for d in band: if any(math.hypot(d["x_m"] - k["x_m"], d["y_m"] - k["y_m"]) < bc.BALL_DEDUP_M for k in out): continue out.append(d) return out def run_config(cw, clip_path: Path, model, cfg: dict, conf: float = None ) -> dict: """One clip x one configuration -> per-frame detections plus timings. Decoding follows ``analyze_clip``: real container PTS, an ``img_diff`` column over the table region, batches of ``DETECT_BATCH``. Duplicate frames are FLAGGED (``find_dup_frames``), not dropped, so a rate can be reported both ways -- the pipeline's derivatives only ever see the real frames. """ import cv2 from poolcoach_cv import broadcast as bc conf = DET_CONF_FLOOR if conf is None else conf th, comp = homography_for(cw, clip_path) ci = th.corners_px.astype(int) x0, x1 = ci[:, 0].min(), ci[:, 0].max() y0, y1 = ci[:, 1].min(), ci[:, 1].max() cap = cv2.VideoCapture(str(clip_path)) if not cap.isOpened(): raise RuntimeError(f"cannot open {clip_path}") frames: list[dict] = [] names: list[str] = [] diffs: list[float] = [] t_predict = 0.0 t_band = 0.0 n_band_batches = 0 band = None batch_imgs: list = [] batch_idx: list[int] = [] prev_crop = None def flush(): nonlocal t_predict, t_band, n_band_batches if not batch_imgs: return t = time.perf_counter() results = model.predict(batch_imgs, conf=conf, imgsz=cfg["imgsz"], iou=NMS_IOU, verbose=False, device=DEVICE) t_predict += time.perf_counter() - t for i, res in zip(batch_idx, results): frames[i]["dets"] = boxes_to_dets(res, th) if cfg["band"]: bx0, by0, bx1, by1 = band["box"] # .copy(): a numpy VIEW of a frame is not contiguous, and the # pre-transform of ultralytics wants a real array crops = [im[by0:by1, bx0:bx1].copy() for im in batch_imgs] t = time.perf_counter() bres = model.predict(crops, conf=conf, imgsz=band["imgsz"], iou=NMS_IOU, verbose=False, device=DEVICE) t_band += time.perf_counter() - t n_band_batches += 1 for i, res in zip(batch_idx, bres): frames[i]["dets"] = merge_band( frames[i]["dets"], boxes_to_dets(res, th, offset=(bx0, by0), source="band")) batch_imgs.clear() batch_idx.clear() try: while True: if not cap.grab(): break t_s = cap.get(cv2.CAP_PROP_POS_MSEC) / 1000.0 ok, frame = cap.retrieve() if not ok: break if band is None and cfg["band"]: band = band_window(th, frame.shape[1], frame.shape[0]) name = f"{len(frames):05d}" crop = frame[max(0, y0):max(1, y1), max(0, x0):max(1, x1)] gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY).astype(np.int16) diffs.append(float(np.abs(gray - prev_crop).mean()) if prev_crop is not None else -1.0) prev_crop = gray names.append(name) frames.append({"name": name, "t_s": round(t_s, 3), "dets": []}) batch_imgs.append(frame) batch_idx.append(len(frames) - 1) if len(batch_imgs) >= bc.DETECT_BATCH: flush() flush() finally: cap.release() dups = bc.find_dup_frames(names, diffs) for fr, d in zip(frames, diffs): fr["img_diff"] = round(d, 4) fr["dup"] = fr["name"] in dups n = max(len(frames), 1) return {"frames": frames, "n_frames": len(frames), "n_dup": len(dups), "band": band, "height_comp": comp, "ms_per_frame": round(1000.0 * (t_predict + t_band) / n, 1), "ms_per_frame_main": round(1000.0 * t_predict / n, 1), "ms_per_frame_band": round(1000.0 * t_band / n, 1)} # ----------------------------------------------------------------- metrics def _r4(v): """Round for the report, never for the arithmetic -- confidences are carried at full precision everywhere else (see ``boxes_to_dets``).""" return None if v is None else round(float(v), 4) def _confs(dets: list[dict]) -> dict: """min / median / max of a list of detections, or nulls when empty.""" if not dets: return {"n": 0, "min": None, "med": None, "max": None} cs = sorted(d["conf"] for d in dets) return {"n": len(cs), "min": round(cs[0], 4), "med": round(statistics.median(cs), 4), "max": round(cs[-1], 4)} def target_t1(run: dict, th) -> dict: """T1 -- ball 5 on shot_02 frame 0: the detections within ``TARGET_MATCH_M`` of the measured ground truth, split by the bounds gate.""" gx, gy = th.px_to_table(GT_BALL5_PX) fr = run["frames"][0] hits = [d for d in fr["dets"] if math.hypot(d["x_m"] - gx, d["y_m"] - gy) <= TARGET_MATCH_M] inside = [d for d in hits if d["in_tol"]] best = max(hits, key=lambda d: d["conf"], default=None) best_in = max(inside, key=lambda d: d["conf"], default=None) return {"gt_px": list(GT_BALL5_PX), "gt_table": [round(float(gx), 4), round(float(gy), 4)], "n_hits": len(hits), "n_hits_in_tol": len(inside), "best_conf": round(best["conf"], 4) if best else None, "best_conf_in_tol": round(best_in["conf"], 4) if best_in else None, "best_det": best, "hits": hits, "n_dets_frame": len(fr["dets"])} def _window(run: dict, lo: float, hi: float, real_only: bool = False): return [f for f in run["frames"] if lo <= f["t_s"] <= hi and not (real_only and f["dup"])] def target_t2(run: dict) -> dict: """T2 -- the cue against the far cushion of shot_05: the fraction of frames in the window carrying a Cue detection above y = 2.0, over all frames and over the REAL (non-duplicate) ones.""" out = {} for label, real in (("all_frames", False), ("real_frames", True)): win = _window(run, *CUE_WINDOW_S, real_only=real) seen, seen_in, confs, confs_in = 0, 0, [], [] for f in win: hits = [d for d in f["dets"] if d["cls"] == "Cue" and d["y_m"] >= CUE_MIN_Y_M] inside = [d for d in hits if d["in_tol"]] if hits: seen += 1 confs.append(max(d["conf"] for d in hits)) if inside: seen_in += 1 confs_in.append(max(d["conf"] for d in inside)) n = len(win) out[label] = { "n_frames": n, "n_seen": seen, "n_seen_in_tol": seen_in, "frac_seen": round(seen / n, 3) if n else None, "frac_seen_in_tol": round(seen_in / n, 3) if n else None, "conf": _confs([{"conf": c} for c in confs]), "conf_in_tol": _confs([{"conf": c} for c in confs_in])} win = _window(run, *CUE_WINDOW_S) out["per_frame"] = [ {"t_s": f["t_s"], "dup": f["dup"], "best": _r4(max((d["conf"] for d in f["dets"] if d["cls"] == "Cue" and d["y_m"] >= CUE_MIN_Y_M), default=None)), "best_in_tol": _r4(max((d["conf"] for d in f["dets"] if d["cls"] == "Cue" and d["y_m"] >= CUE_MIN_Y_M and d["in_tol"]), default=None))} for f in win] return out def target_t3(run: dict) -> dict: """T3 -- shot_03's object ball over the last stretch into the far right pocket: which frames of the window still carry it, and how close to the pocket the last one sits.""" win = _window(run, *S03_WINDOW_S) rows, last, last_any = [], None, None for f in win: hits = [d for d in f["dets"] if d["cls"] != "Cue" and d["y_m"] >= S03_MIN_Y_M and d["x_m"] >= S03_MIN_X_M] best = max(hits, key=lambda d: d["conf"], default=None) best_in = max([d for d in hits if d["in_tol"]], key=lambda d: d["conf"], default=None) if best is not None: last_any = best if best_in is not None: last = best_in rows.append({"t_s": f["t_s"], "dup": f["dup"], "best": _r4(best["conf"]) if best else None, "best_in_tol": _r4(best_in["conf"]) if best_in else None, "xy": [best["x_m"], best["y_m"]] if best else None, "in_tol": bool(best_in) if best else None}) seen = [r for r in rows if r["best"] is not None] seen_in = [r for r in rows if r["best_in_tol"] is not None] # The distance to the pocket is measured from the last IN-BOUNDS point. # A box past the far rail projects to a table y of 4 m and more (the # bounds gate is exactly what the pipeline uses to throw those away), so # taking it as "the last point" would report a gap of 1.7 m for a ball # that in fact reached the pocket. Both are reported; only the in-bounds # one is a distance. gap = (round(math.hypot(last["x_m"] - POCKET_FAR_RIGHT[0], last["y_m"] - POCKET_FAR_RIGHT[1]), 4) if last else None) return {"n_frames": len(rows), "n_seen": len(seen), "n_seen_in_tol": len(seen_in), "frac_seen": round(len(seen) / len(rows), 3) if rows else None, "frac_seen_in_tol": (round(len(seen_in) / len(rows), 3) if rows else None), "last_xy": [last["x_m"], last["y_m"]] if last else None, "last_xy_any": ([last_any["x_m"], last_any["y_m"]] if last_any else None), "last_gap_to_pocket_m": gap, "per_frame": rows} def target_t4(run: dict) -> dict: """T4 -- the background: detections per frame above ``FAR_Y_M``, at the floor and at the operating point.""" op = OP_CONF per, per_op = [], [] for f in run["frames"]: far = [d for d in f["dets"] if d["y_m"] > FAR_Y_M and d["in_tol"]] per.append(len(far)) per_op.append(sum(1 for d in far if d["conf"] >= op)) n = max(len(per), 1) return {"n_frames": len(per), "per_frame_mean": round(sum(per) / n, 3), "per_frame_mean_op": round(sum(per_op) / n, 3), "max": max(per) if per else 0, "n_frames_zero": sum(1 for v in per if v == 0), "series": per, "series_op": per_op} def rack_centre(shot: dict) -> tuple[float, float]: """The centre of the rack of one demo shot -- the mean of the non-cue entries of its ``balls_init``. Read from the bundle rather than written down here, so the number cannot drift away from the demo.""" pts = [(b["x_m"], b["y_m"]) for b in shot.get("balls_init") or [] if b.get("type") != "cue"] return (sum(p[0] for p in pts) / len(pts), sum(p[1] for p in pts) / len(pts)) def rack_sweep(cw, clip_path: Path, model, centre) -> list[dict]: """T5 -- ``imgsz`` x ``iou`` over the FIRST frame only: how many balls of the pile come back, raw and after the ``BALL_DEDUP_M`` merge.""" import cv2 from poolcoach_cv import broadcast as bc th, _ = homography_for(cw, clip_path) cap = cv2.VideoCapture(str(clip_path)) try: ok, frame = cap.read() finally: cap.release() if not ok: raise RuntimeError(f"cannot read the first frame of {clip_path}") rows = [] for imgsz in RACK_IMGSZ: for iou in RACK_IOU: t = time.perf_counter() res = model.predict([frame], conf=DET_CONF_FLOOR, imgsz=imgsz, iou=iou, verbose=False, device=DEVICE)[0] ms = round(1000.0 * (time.perf_counter() - t), 1) dets = [d for d in boxes_to_dets(res, th) if d["in_tol"] and d["cls"] != "Cue" and math.hypot(d["x_m"] - centre[0], d["y_m"] - centre[1]) <= RACK_R_M] dets.sort(key=lambda d: -d["conf"]) merged: list[dict] = [] for d in dets: if any(math.hypot(d["x_m"] - k["x_m"], d["y_m"] - k["y_m"]) < bc.BALL_DEDUP_M for k in merged): continue merged.append(d) rows.append({"imgsz": imgsz, "iou": iou, "ms": ms, "n_raw": len(dets), "n_dedup": len(merged), "n_dedup_op": sum(1 for d in merged if d["conf"] >= OP_CONF), "confs": [_r4(d["conf"]) for d in merged], "dets": merged}) return rows BENCH_FRAMES = 64 # frames of shot_02 the cost benchmark runs on BENCH_REPEATS = 3 # ... repeated this often, MINIMUM reported def cost_benchmark(cw, clip_path: Path, model) -> dict: """CPU milliseconds per frame of each configuration, measured on its own. The per-clip averages of the measurement passes are contaminated: they are collected over half an hour on a machine that is doing other things, and the RATIO between configurations -- which is the whole of decision rule (c) in section 3.1 -- came out incoherent that way (960 measured cheaper than 640 on one clip and 1.6x dearer on another). This is the honest version: one set of frames decoded once, every configuration warmed at its own ``imgsz``, all of them timed back to back, and the MINIMUM of the repeats reported -- the minimum is the pass least disturbed by whatever else the machine was doing. """ import cv2 from poolcoach_cv import broadcast as bc th, _ = homography_for(cw, clip_path) cap = cv2.VideoCapture(str(clip_path)) frames = [] try: while len(frames) < BENCH_FRAMES: ok, frame = cap.read() if not ok: break frames.append(frame) finally: cap.release() band = band_window(th, frames[0].shape[1], frames[0].shape[0]) batches = [frames[i:i + bc.DETECT_BATCH] for i in range(0, len(frames), bc.DETECT_BATCH)] def one_pass(cfg: dict) -> tuple[float, float]: t_main = t_band = 0.0 for batch in batches: t = time.perf_counter() model.predict(batch, conf=DET_CONF_FLOOR, imgsz=cfg["imgsz"], iou=NMS_IOU, verbose=False, device=DEVICE) t_main += time.perf_counter() - t if not cfg["band"]: continue bx0, by0, bx1, by1 = band["box"] crops = [im[by0:by1, bx0:bx1].copy() for im in batch] t = time.perf_counter() model.predict(crops, conf=DET_CONF_FLOOR, imgsz=band["imgsz"], iou=NMS_IOU, verbose=False, device=DEVICE) t_band += time.perf_counter() - t n = len(frames) return 1000.0 * t_main / n, 1000.0 * t_band / n out: dict[str, dict] = {} for cfg in CONFIGS: one_pass(cfg) # warm at this imgsz runs = [one_pass(cfg) for _ in range(BENCH_REPEATS)] best = min(runs, key=lambda r: r[0] + r[1]) out[cfg["name"]] = { "n_frames": len(frames), "repeats": BENCH_REPEATS, "ms_main": round(best[0], 1), "ms_band": round(best[1], 1), "ms_per_frame": round(best[0] + best[1], 1), "all_repeats": [round(a + b, 1) for a, b in runs]} print(f"[probe] cost {cfg['name']:<10} " f"{out[cfg['name']]['ms_per_frame']:>7.1f} ms/frame " f"(repeats {out[cfg['name']]['all_repeats']})", flush=True) return out def reproduction_check(cw, clip_path: Path, model, demo_shot: dict) -> dict: """full640 at the OPERATING confidence on shot_02 frame 0 -> the ``balls_init`` the pipeline would build -> compared with the one the demo bundle carries. This is the probe's own negative control on itself: if the clip, the homography and the height compensation are the ones the demo used, the five entries come back to the millimetre -- ball 5 missing included. A mismatch means every number below was measured on a different image than the demo, so it is reported FIRST. """ from poolcoach_cv import broadcast as bc run = run_config(cw, clip_path, model, {"name": "full640", "imgsz": 640, "band": False}, conf=OP_CONF) dets = sorted(run["frames"][0]["dets"], key=lambda d: -d["conf"]) got: list[dict] = [] for d in dets: if not d["in_tol"]: continue if any(math.hypot(d["x_m"] - k["x_m"], d["y_m"] - k["y_m"]) < bc.BALL_DEDUP_M for k in got): continue got.append({"x_m": round(d["x_m"], 4), "y_m": round(d["y_m"], 4), "type": "cue" if d["cls"] == "Cue" else "ball", "conf": round(d["conf"], 3)}) want = [{"x_m": b["x_m"], "y_m": b["y_m"], "type": b["type"], "conf": b["conf"]} for b in demo_shot.get("balls_init") or []] return {"ok": got == want, "n_demo": len(want), "n_probe": len(got), "demo": want, "probe": got, "ms_per_frame": run["ms_per_frame"]} # ------------------------------------------------------------------ report def md_report(rep: dict) -> str: """The tables a human reads. Everything here also lives in the JSON -- this is a view, never a second source of numbers.""" L: list[str] = [] add = L.append add("# Far-zone detection probe -- 28/08/2026") add("") add(f"Design doc `7_FarZone_Detection_BallID.md` section 3.1, step P of " f"BRIEF 28/08. VOD `{rep['vod']}`, corners `{rep['corners']}`, " f"device `{rep['device']}`, weights `{rep['weights']}`, " f"`conf={DET_CONF_FLOOR}` everywhere, `OP_CONF={rep['op_conf']}`.") add("") add("**This file MEASURES. It chooses nothing** -- the decision rule is " "section 3.1 of the design doc and it is Cowork's to apply.") add("") rc = rep["reproduction_check"] add("## 0. Reproduction check -- is this the same image the demo saw?") add("") add(f"full640 re-run at the operating confidence on shot_02 frame 0, " f"`balls_init` rebuilt the way `analyze_clip` builds it, compared " f"with bundle `{rep['bundle']}`: " f"**{'MATCH' if rc['ok'] else 'MISMATCH'}** " f"({rc['n_probe']} entries against {rc['n_demo']}).") add("") def _entry(b) -> str: if not b: return "-" return "{:.4f}, {:.4f} ({})".format(b["x_m"], b["y_m"], b["type"]) add("| # | demo x,y (m) | demo conf | probe x,y (m) | probe conf |") add("|---|---|---|---|---|") for i in range(max(len(rc["demo"]), len(rc["probe"]))): d = rc["demo"][i] if i < len(rc["demo"]) else None p = rc["probe"][i] if i < len(rc["probe"]) else None add("| {} | {} | {} | {} | {} |".format( i, _entry(d), d["conf"] if d else "-", _entry(p), p["conf"] if p else "-")) add("") add("## 1. T1 -- shot_02 frame 0, ball 5 in the mouth of the far right " "pocket") add("") t1 = rep["targets"]["T1"] gt = t1["by_config"][CONFIGS[0]["name"]] add(f"Ground truth: pixel `{gt['gt_px']}` (measured purple centroid, " f"visible bbox 28 x 22 px) -> table `{gt['gt_table']}`, match " f"radius {TARGET_MATCH_M} m.") add("") add("| config | dets on frame | hits | best conf (all) | hits in_tol | " "best conf (in_tol) | >= OP_CONF | bbox w x h px |") add("|---|---|---|---|---|---|---|---|") for cfg in CONFIGS: r = t1["by_config"][cfg["name"]] b = r["best_det"] bbox = "{} x {}".format(b["w_px"], b["h_px"]) if b else "-" reach = "YES" if (r["best_conf_in_tol"] or 0) >= rep["op_conf"] \ else "no" add("| {} | {} | {} | {} | {} | {} | {} | {} |".format( cfg["name"], r["n_dets_frame"], r["n_hits"], r["best_conf"] if r["best_conf"] is not None else "-", r["n_hits_in_tol"], r["best_conf_in_tol"] if r["best_conf_in_tol"] is not None else "-", reach, bbox)) add("") add("## 2. T2 -- shot_05, the cue against the far cushion, " f"t in [{CUE_WINDOW_S[0]}, {CUE_WINDOW_S[1]}]") add("") add("| config | frames | seen | frac | frac (real frames) | conf min | " "conf med | conf max | frac in_tol |") add("|---|---|---|---|---|---|---|---|---|") for cfg in CONFIGS: r = rep["targets"]["T2"]["by_config"][cfg["name"]] a, real = r["all_frames"], r["real_frames"] c = a["conf"] add(f"| {cfg['name']} | {a['n_frames']} | {a['n_seen']} | " f"{a['frac_seen']} | {real['frac_seen']} | " f"{c['min'] if c['min'] is not None else '-'} | " f"{c['med'] if c['med'] is not None else '-'} | " f"{c['max'] if c['max'] is not None else '-'} | " f"{a['frac_seen_in_tol']} |") add("") add("## 3. T3 -- shot_03, the last run into the far right pocket, " f"t in [{S03_WINDOW_S[0]}, {S03_WINDOW_S[1]}]") add("") add("(the demo's track dies at t = 1.300, 0.2375 m short of the pocket)") add("") add("| config | frames | seen (all) | seen (in_tol) | frac (in_tol) | " "last in-bounds x,y | gap to pocket (m) | last x,y incl. out of " "bounds |") add("|---|---|---|---|---|---|---|---|") for cfg in CONFIGS: r = rep["targets"]["T3"]["by_config"][cfg["name"]] add("| {} | {} | {} | {} | {} | {} | {} | {} |".format( cfg["name"], r["n_frames"], r["n_seen"], r["n_seen_in_tol"], r["frac_seen_in_tol"], r["last_xy"] or "-", r["last_gap_to_pocket_m"] if r["last_gap_to_pocket_m"] is not None else "-", r["last_xy_any"] or "-")) add("") add(f"## 4. T4 -- background, detections per frame with y_m > {FAR_Y_M}") add("") add("| config | clip | frames | dets/frame (conf >= 0.05) | " "dets/frame (conf >= OP_CONF) | max | frames with none |") add("|---|---|---|---|---|---|---|") for cfg in CONFIGS: for clip in ("shot_02", "shot_05"): r = rep["targets"]["T4"]["by_config"][cfg["name"]][clip] add(f"| {cfg['name']} | {clip} | {r['n_frames']} | " f"{r['per_frame_mean']} | {r['per_frame_mean_op']} | " f"{r['max']} | {r['n_frames_zero']} |") add("") add("## 5. T5 -- the rack (failure mode 2), shot_09 and shot_10 frame 0") add("") add(f"Balls counted inside {RACK_R_M} m of the cluster centre of the " f"demo's own `balls_init`. `conf={DET_CONF_FLOOR}`. The demo bundle " f"itself has " f"{rep['targets']['T5']['shot_09']['n_demo']}/9 and " f"{rep['targets']['T5']['shot_10']['n_demo']}/9.") add("") add("| clip | centre x,y | imgsz | iou | raw | after dedup | " ">= OP_CONF | ms | confs |") add("|---|---|---|---|---|---|---|---|---|") for clip in ("shot_09", "shot_10"): blk = rep["targets"]["T5"][clip] for row in blk["sweep"]: add(f"| {clip} | " f"{blk['centre'][0]:.3f}, {blk['centre'][1]:.3f} | " f"{row['imgsz']} | {row['iou']} | {row['n_raw']} | " f"{row['n_dedup']} | {row['n_dedup_op']} | {row['ms']} | " f"{row['confs']} |") add("") add("## 6. Cost -- CPU milliseconds per frame") add("") add(f"The `benchmark` column is the number decision rule (c) is read " f"from: {BENCH_FRAMES} frames of shot_02, every configuration warmed " f"at its own imgsz and then timed back to back, best of " f"{BENCH_REPEATS}. The two per-clip columns are the averages of the " f"measurement passes themselves -- they cover half an hour on a " f"machine doing other things and are kept only as a witness, not as " f"a ratio.") add("") add("| config | benchmark ms/frame | main | band | x baseline 640 | " "pass shot_02 | pass shot_05 |") add("|---|---|---|---|---|---|---|") base = rep["cost"]["full640"]["ms_per_frame"] for cfg in CONFIGS: c = rep["cost"][cfg["name"]] r = rep["runtime"][cfg["name"]] add("| {} | {} | {} | {} | {} | {} | {} |".format( cfg["name"], c["ms_per_frame"], c["ms_main"], c["ms_band"], round(c["ms_per_frame"] / base, 2), r["shot_02"], r["shot_05"])) add("") b = rep["band"] add(f"The band pass crops `{b['box']}` ({b['w']} x {b['h']} px) and runs " f"at `imgsz={b['imgsz']}` (scale {b['scale']} -- at or above native, " f"never a downscale).") add("") add("## 7. Archive integrity") add("") a = rep["archive"] add(f"`{a['root']}`: {a['n_files']} files fingerprinted (size + mtime) " f"before and after -- **{a['n_changed']} changed**.") add("") return "\n".join(L) + "\n" # -------------------------------------------------------------------- main OP_CONF = 0.2993 # replaced at runtime by cv_worker.OP_CONF -- declared # here only so the metric helpers can be read top down def main() -> int: global OP_CONF OUT.mkdir(parents=True, exist_ok=True) print(f"[probe] app root : {APP_ROOT}") print(f"[probe] vod : {VOD} ({VOD.stat().st_size / 1e6:.1f} MB)") print(f"[probe] bundle : {BUNDLE}") print(f"[probe] out : {OUT}", flush=True) print(f"[probe] fingerprinting the archive {ARCHIVE_ROOT} ...", flush=True) fp_before = fingerprint(ARCHIVE_ROOT) print(f"[probe] {len(fp_before)} files", flush=True) cw = load_worker() OP_CONF = float(cw.OP_CONF) from poolcoach_cv import broadcast as bc # noqa: F401 -- warm manifest = json.loads((BUNDLE / "shots.json").read_text(encoding="utf-8")) clips = cut_clips(cw, manifest["shots"]) weights = cw.resolve_weights() print(f"[probe] loading {weights} ...", flush=True) from ultralytics import YOLO model = YOLO(str(weights)) model.predict(np.zeros((640, 640, 3), dtype=np.uint8), conf=DET_CONF_FLOOR, imgsz=640, verbose=False, device=DEVICE) print("[probe] model ready", flush=True) demo02 = json.loads((BUNDLE / "shot_02.json").read_text(encoding="utf-8")) print("[probe] reproduction check on shot_02 frame 0 ...", flush=True) rc = reproduction_check(cw, clips[2], model, demo02) print(f"[probe] reproduction check: " f"{'MATCH' if rc['ok'] else 'MISMATCH'} " f"({rc['n_probe']} vs {rc['n_demo']} balls_init entries)", flush=True) runs: dict[str, dict] = {} for cfg in CONFIGS: for idx in (2, 3, 5): key = f"{cfg['name']}/shot_{idx:02d}" t = time.perf_counter() runs[key] = run_config(cw, clips[idx], model, cfg) print(f"[probe] {key:<26} {runs[key]['n_frames']:>4} frames " f"({runs[key]['n_dup']} dup) " f"{runs[key]['ms_per_frame']:>7.1f} ms/frame " f"in {time.perf_counter() - t:.1f}s", flush=True) th02, comp02 = homography_for(cw, clips[2]) targets = { "T1": {"by_config": {c["name"]: target_t1(runs[f"{c['name']}/shot_02"], th02) for c in CONFIGS}}, "T2": {"by_config": {c["name"]: target_t2(runs[f"{c['name']}/shot_05"]) for c in CONFIGS}}, "T3": {"by_config": {c["name"]: target_t3(runs[f"{c['name']}/shot_03"]) for c in CONFIGS}}, "T4": {"by_config": {c["name"]: { "shot_02": target_t4(runs[f"{c['name']}/shot_02"]), "shot_05": target_t4(runs[f"{c['name']}/shot_05"])} for c in CONFIGS}}, "T5": {}, } for idx in (9, 10): shot = json.loads((BUNDLE / f"shot_{idx:02d}.json") .read_text(encoding="utf-8")) centre = rack_centre(shot) sweep = rack_sweep(cw, clips[idx], model, centre) targets["T5"][f"shot_{idx:02d}"] = { "centre": [round(centre[0], 4), round(centre[1], 4)], "n_demo": sum(1 for b in shot.get("balls_init") or [] if b.get("type") != "cue"), "sweep": sweep} best = max(r["n_dedup"] for r in sweep) print(f"[probe] rack shot_{idx:02d}: demo " f"{targets['T5'][f'shot_{idx:02d}']['n_demo']}/9, probe best " f"{best}/9 over the sweep", flush=True) runtime = {} for cfg in CONFIGS: a = runs[f"{cfg['name']}/shot_02"]["ms_per_frame"] b = runs[f"{cfg['name']}/shot_05"]["ms_per_frame"] runtime[cfg["name"]] = {"shot_02": a, "shot_05": b, "mean": round((a + b) / 2, 1), "main": runs[f"{cfg['name']}/shot_02"] ["ms_per_frame_main"], "band": runs[f"{cfg['name']}/shot_02"] ["ms_per_frame_band"]} print("[probe] cost benchmark (isolated, back to back) ...", flush=True) cost = cost_benchmark(cw, clips[2], model) fp_after = fingerprint(ARCHIVE_ROOT) changed = sorted(k for k in set(fp_before) | set(fp_after) if fp_before.get(k) != fp_after.get(k)) report = { "date": "2026-08-28", "design_doc": "7_FarZone_Detection_BallID.md section 3.1", "vod": str(VOD), "corners": CORNERS, "bundle": BUNDLE.name, "weights": str(weights), "device": DEVICE, "op_conf": OP_CONF, "det_conf_floor": DET_CONF_FLOOR, "nms_iou": NMS_IOU, "height_comp": comp02, "band": runs["band/shot_02"]["band"], "reproduction_check": rc, "configs": CONFIGS, "targets": targets, "cost": cost, "runtime": runtime, "archive": {"root": str(ARCHIVE_ROOT), "n_files": len(fp_before), "n_changed": len(changed), "changed": changed[:20]}, "runs": {k: {"n_frames": v["n_frames"], "n_dup": v["n_dup"], "ms_per_frame": v["ms_per_frame"]} for k, v in runs.items()}, } (OUT / "probe_farzone.json").write_text( json.dumps(report, indent=1), encoding="utf-8") (OUT / "probe_farzone.md").write_text(md_report(report), encoding="utf-8") print(f"[probe] wrote {OUT / 'probe_farzone.json'}") print(f"[probe] wrote {OUT / 'probe_farzone.md'}") if changed: print(f"[probe] RED: {len(changed)} file(s) under {ARCHIVE_ROOT} " f"changed -- {changed[:5]}") return 7 print(f"[probe] archive untouched ({len(fp_before)} files identical)") if not rc["ok"]: print("[probe] RED: the reproduction check failed -- the probe did " "NOT measure the same image the demo did.") return 8 return 0 if __name__ == "__main__": raise SystemExit(main())