scripts/experiments/probe_farzone_det.py

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