scripts/cv_worker.py

"""CV worker — scan ảnh bàn ra thế bi, mode queue (bàn giao 18, 05/08/2026).

Process TÁCH KHỎI API, chạy trên venv `poolcoach-cv-env` (CUDA) — app/API
tuyệt đối không import torch/ultralytics (quyết định 2 venv, bàn giao 17).
Load YOLO MỘT lần lúc boot + warmup 1 predict, xong hết mới bật heartbeat
(cùng nếp engine_worker: không báo alive lúc còn đang khởi động). Vòng đời:

    BRPOP pc:jobs:scan → handle_scan → LPUSH pc:result:{job_id} (TTL dọn rác)

Job payload: ``{"image_b64": <JPEG/PNG base64>, "corners": [[x,y]×4]}`` —
corners là pixel trên ẢNH ĐÃ GỬI, thứ tự như ``table_corners()`` (đi vòng
quanh bàn, 2 góc đầu là một BĂNG NGẮN: (0,0) → (W,0) → (W,L) → (0,L)).
Đi vòng NGƯỢC chiều cũng được — ``orient_corners`` tự phát hiện và đảo
(06/08), toạ độ bi ra y hệt chiều xuôi.
Reply: ``{"balls": [{x, y, type, conf, ...}]}`` — toạ độ bàn hệ pooltool (m),
``type`` ∈ {"cue", "ball"}; class Dot (nút kim cương thành gỗ) bị LOẠI.
Bi ``type="ball"`` mang thêm ``number`` (int 1–9 | None — BallID hai tầng
theo MÀU, bàn giao 22), ``number_conf`` ∈ [0,1] và cờ ``wb`` (có chuẩn hoá
trắng theo bi cue không); bi ``cue`` KHÔNG mang các trường này. Mở rộng
thuần — client cũ bỏ qua trường lạ.

Conf vận hành 0.2993 — argmax smooth-F1 any-ball trên val (scripts/cv/
pick_conf.py, 05/08, chuẩn đo trong opconf.json: F1 0.9322 · P 0.9642 ·
R 0.9023, VAL-pipeline; số này không so ngang với predict-pipeline đang chạy
ở đây — hai cái thước, bàn giao 17). Override bằng env ``POOLCOACH_CV_CONF``.

Heartbeat RIÊNG (pc:worker:heartbeat:scan) — chết/sống độc lập engine worker.

Bàn giao 29 (12/08): khối ``shotnet`` mang thêm ``confidence_raw`` +
``det_density``, và ``confidence`` (hạng tin HIỂN THỊ) đi qua **van mật độ
detection** — xem bảng hằng ``SHOTNET_DET_DENSITY_*``. Van CHỈ đổi hạng chữ
và thêm một dòng ``warnings``: không đụng số model, không ẩn cột, không
chặn inference.

Bàn giao 31 (13/08): job analyze BÙ ĐỘ CAO TÂM BI cho mọi toạ độ map qua
homography (``comp_homography_cls`` — camera decompose từ homography của
chính clip, ``poolcoach_cv.camera``); kết quả mang cờ ``height_comp:
{on, h_m, f_px, cx_m, cy_m}`` để phân biệt hai thời kỳ quy ước (cx/cy
thêm ở BG32, cùng lúc siết biên sanity h ∈ [1.5, 6]m / f ∈ [800, 8000]px
— ca rác cú 11 h=0.686/f=12767 từng lọt biên rộng BG31). Decompose thất
bại / camera ngoài biên → không bù + cờ off (kèm warning). KHÔNG áp hồi
tố track cũ; scan giữ nguyên quy ước cũ (chưa bù — ảnh scan không chắc
cùng camera cao).

Bàn giao 24 (11/08): worker này phục vụ THÊM queue ``pc:jobs:analyze``
(Broadcast Analyzer — clip 1 cú → track + số giải tích, module
``poolcoach_cv.broadcast``). Mở rộng thuần: handler scan + contract
/api/scan không đổi một byte. Job analyze DÀI (cỡ phút) nên khác giao thức:
kết quả/tiến độ ghi vào status key (``jobqueue.set_analyze_status``), API
poll — không ai BLPOP reply; heartbeat được beat cả TRONG lúc chạy job
(progress callback) để scan/analyze không fast-fail oan trong lúc worker
bận. Trong lúc một analyze chạy, scan job phải chờ — chấp nhận, demo local
một người dùng (BRIEF #1: mọi trade-off nghiêng về đơn giản).

Lát A1 (13/08): worker phục vụ THÊM queue ``pc:jobs:segment`` (Analyzer đa
cú — video dài → ``poolcoach_cv.segmentation`` cắt danh sách cú → manifest
+ thumbnail + clip per cú ghi FILE ngay → enqueue N job analyze NGUYÊN
TRẠNG, chỉ thêm ``result_path`` để kết quả per cú cũng ra file — nếp
BG29b/BG31: Redis chỉ là kênh tiến độ, đĩa mới là nguồn sự thật). Phần SỐ
của scan/analyze (weights c4b, conf 0.2993, imgsz 640, van tin cậy) không
đổi một byte.

Chạy tay (Redis: docker compose -f docker-compose.dev.yml up -d redis):
    set REDIS_URL=redis://localhost:6379/0
    D:\\Khoa luan\\poolcoach-cv-env\\Scripts\\python.exe scripts\\cv_worker.py
Launcher: D:\\Khoa luan\\run_cv_worker.bat (ngoài repo, như mọi launcher).
"""

from __future__ import annotations

import base64
import binascii
import math
import os
import sys
import time
from pathlib import Path

# Repo KHÔNG cài package: src/ cho poolcoach_cv, ROOT cho app.* (quy ước
# scripts/ — xem CLAUDE.md). KHÔNG import poolcoach_rl: venv CV không có
# pooltool, và worker này không cần vật lý — chỉ cần homography thuần numpy.
ROOT = Path(__file__).resolve().parents[1]
for _p in (ROOT / "src", ROOT):
    if str(_p) not in sys.path:
        sys.path.insert(0, str(_p))

import numpy as np  # noqa: E402

from app import analyzer_store as astore  # noqa: E402 — thuần stdlib
from app import jobqueue  # noqa: E402 — chỉ kéo redis, không FastAPI/torch
from poolcoach_cv.ballid import identify_balls  # noqa: E402 — thuần numpy
from poolcoach_cv.camera import HeightCompensation  # noqa: E402 — thuần numpy
from poolcoach_cv.homography import (TABLE_L, TABLE_W,  # noqa: E402
                                     TableHomography, orient_corners)

DEFAULT_WEIGHTS = r"D:\Khoa luan\cv_full_20260805\best.pt"

# ---- ShotNet (BG28): model suy ngược (V0 gậy, phi, a, b) chạy KÈM analytic
# trong job analyze — hai phương pháp hiện song song (BRIEF 28 bối cảnh 1).
DEFAULT_SHOTNET = str(ROOT / "models" / "shotnet_20260812_c4b" / "best.pt")
                        # checkpoint c4b (augmentation BG29) — override bằng
                        # env POOLCOACH_SHOTNET_WEIGHTS; thiếu file → analyze
                        # vẫn chạy, chỉ không có khối shotnet (kèm warning rõ)
                        #
                        # Gate P2′ BG29 vẫn = ĐỎ theo luật "đạt bar VÀ hơn
                        # c3" (3/4 ô dưới bar tuyệt đối — kết quả đứng
                        # nguyên, không chấm lại). App đổi sang c4b theo
                        # QUYẾT ĐỊNH COWORK 13/08/2026, không phải vì gate
                        # xanh: c4b hơn c3 cả 4 ô trên thước chính thức
                        # (lát ≥0.3s, ghép cặp 669 cú: 2.24°/8.56%/74.3%/
                        # 72.9% vs 3.03/9.73/69.7/65.3, CI95 dphi/V0 không
                        # chứa 0), và không thụt lùi trên 5k đầy đủ.
SHOTNET_V0_SCALE_REAL = 2.54 / 1.9812
                        # quy đổi V0 model → m/s bàn THẬT (≈1.282). Thuần
                        # hình học, không fit: net train trên bàn pooltool
                        # (L=1.9812m), input thật chuẩn hoá đẳng hướng theo
                        # bàn giải (L=2.54m, cùng tỷ lệ 2:1) — thế giới net
                        # thấy là bản thu 0.78× của bàn thật với cùng
                        # timestamp, nên V0 nó đọc ra là m/s của thế giới
                        # thu; nhân L_thật/L_synth trả về m/s thật. φ/a/b
                        # bất biến qua phép đồng dạng — không đổi gì.
SHOTNET_CONF_HIGH = 0.8  # p_ident ≥ → hạng tin "high" cho spin model
SHOTNET_CONF_MED = 0.5   # p_ident ≥ → "medium"; dưới → "low" (spin chỉ lộ
                         # qua va chạm — p_ident thấp nghĩa là net tự khai
                         # không thấy va chạm để đọc spin)

# ---- Van hiển thị tin cậy theo MẬT ĐỘ DETECTION bi mục tiêu (BG29 bước 3)
# Bài học BG28: p_ident = 1.0 / conf "high" trên CẢ HAI cú thật lệch nặng
# (cú 12 φ lệch 84.9°) — head tin cậy KHÔNG tự khai được input lạ, vì nó
# học "cú này có va chạm không", không phải "input này có giống lúc train
# không". Van thứ hai đo đại lượng NGOÀI model: mật độ detection bi mục
# tiêu của chính clip (det không-cue/frame — đúng đại lượng đã đo được ở
# BG28: cú 11 = 1.01, cú 12 = 3.09).
#
# Ngưỡng lấy từ phân phối TRAIN đo được (shard 0, 2000 cú synth): trung
# bình 5.28 det không-cue/frame, p10 = 2.00, p90 = 8.21. Dưới p10 nghĩa là
# clip thưa hơn 90% dữ liệu net từng thấy lúc đo số val — hạ MỘT bậc; thưa
# hơn nữa (< 1.5) thì bản thân bài toán mất thông tin, hạ HAI bậc.
# Augmentation BG29 mở rộng vùng net CHỊU ĐƯỢC, nhưng không tạo lại thông
# tin đã mất — nên van giữ nguyên ngưỡng theo phân phối gốc.
SHOTNET_DET_DENSITY_MED = 2.0   # < → hạ 1 bậc hạng tin HIỂN THỊ
SHOTNET_DET_DENSITY_LOW = 1.5   # < → hạ 2 bậc
SHOTNET_CONF_RANKS = ("low", "medium", "high")


def density_confidence(rank: str, density: float) -> str:
    """Hạ cấp hạng tin HIỂN THỊ theo mật độ detection. CHỈ đổi hạng chữ —
    không đụng một số nào của model, không ẩn cột, không chặn inference
    (BRIEF 29 bước 3.1)."""
    if density < SHOTNET_DET_DENSITY_LOW:
        step = 2
    elif density < SHOTNET_DET_DENSITY_MED:
        step = 1
    else:
        return rank
    if rank not in SHOTNET_CONF_RANKS:
        return rank
    return SHOTNET_CONF_RANKS[max(SHOTNET_CONF_RANKS.index(rank) - step, 0)]


_CONF_VN = {"low": "thấp", "medium": "vừa", "high": "cao"}


def density_warning(blk: dict) -> str | None:
    """Dòng cảnh báo tiếng Việt khi van đã hạ cấp — None nếu không hạ.
    Server soạn sẵn (nếp ``spin_evidence[].text``): FE nối hiện thẳng."""
    raw, shown = blk.get("confidence_raw"), blk.get("confidence")
    if not raw or raw == shown:
        return None
    return (f"Mật độ detection bi mục tiêu chỉ {blk['det_density']:.2f}/frame "
            f"— thưa hơn vùng dữ liệu net được huấn luyện "
            f"(p10 ≈ {SHOTNET_DET_DENSITY_MED:g}/frame, trung bình 5.3). "
            f"Hạng tin cậy HIỂN THỊ của khối model đã hạ từ "
            f"\"{_CONF_VN.get(raw, raw)}\" xuống "
            f"\"{_CONF_VN.get(shown, shown)}\"; số model giữ nguyên, đọc "
            f"kèm dè dặt.")
OP_CONF = 0.2993       # pick_conf 05/08 (opconf.json) — xem docstring;
                       # env POOLCOACH_CV_CONF. 0.299 cũ là bake lệch (06/08)
IMGSZ = 640            # cùng imgsz train/val
MAX_DIM = 1280         # phòng hờ: FE đã resize trước khi gửi, nhưng client lạ
                       # gửi ảnh gốc thì worker tự thu về ~2x imgsz cho đủ nét
BALL_CLASSES = {"Black", "Cue", "Solid", "Striped"}   # Dot loại — không phải bi
OUT_TOL_M = 0.03       # tâm chiếu lệch ra ngoài bàn ≤ 3cm → kẹp vào mép (bi
                       # sát băng lệch ~9mm là hạn chế đã đo, bàn giao 16);
                       # lệch hơn → loại (detect ngoài mặt bàn: người, băng gỗ)
DEDUP_DIST_M = 0.03    # 2 tâm cách < ~1R (bi r=0.0286m) = cùng một bi bị double
                       # -detect → giữ conf cao hơn; 2 bi thật chạm nhau cách 2R


def scan_once(payload: dict, model, conf: float, predict_kw: dict) -> dict:
    """MỘT job scan: decode → (resize) → YOLO → tâm bbox → homography → balls.

    Ném ValueError cho MỌI lỗi thuộc về input (thiếu trường, base64 hỏng,
    ảnh không đọc được, corners suy biến — TableHomography tự ném) — phía
    API trả 422 message nguyên văn, cùng giao thức với engine worker.
    """
    image_b64 = payload.get("image_b64")
    corners = payload.get("corners")
    if not isinstance(image_b64, str) or not image_b64:
        raise ValueError("Thiếu image_b64 (ảnh JPEG/PNG mã hoá base64).")
    corners = np.asarray(corners, dtype=np.float64) if corners is not None else None
    if corners is None or corners.shape != (4, 2):
        raise ValueError("corners phải là 4 điểm [x, y] pixel — chấm đủ 4 góc bàn.")
    try:
        raw = base64.b64decode(image_b64, validate=True)
    except binascii.Error as e:
        raise ValueError(f"image_b64 không phải base64 hợp lệ ({e}).") from e

    import cv2

    img = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
    if img is None:
        raise ValueError("Không giải mã được ảnh — cần JPEG hoặc PNG.")
    h, w = img.shape[:2]
    scale = 1.0
    if max(h, w) > MAX_DIM:
        scale = MAX_DIM / max(h, w)
        img = cv2.resize(img, (round(w * scale), round(h * scale)),
                         interpolation=cv2.INTER_AREA)

    # Chấm 4 góc ngược chiều → tự đảo cho hết soi gương (06/08) — lỗi người
    # dùng duy nhất sửa được bằng toán; toạ độ bi ra y hệt lần chấm xuôi.
    corners = orient_corners(corners)
    th = TableHomography(corners)          # suy biến → ValueError tiếng Việt sẵn
    res = model.predict(img, conf=conf, imgsz=IMGSZ, verbose=False,
                        **predict_kw)[0]
    names = res.names

    cands = []
    n_dot = n_out = n_dup = 0
    for b in res.boxes:
        cls = names[int(b.cls)]
        if cls not in BALL_CLASSES:
            n_dot += 1
            continue
        x1, y1, x2, y2 = (float(v) for v in b.xyxy[0])
        # tâm bbox về pixel ảnh GỐC (corners đo trên ảnh client gửi, chưa resize)
        cx, cy = (x1 + x2) / 2.0 / scale, (y1 + y2) / 2.0 / scale
        tx, ty = th.px_to_table((cx, cy))
        if not (-OUT_TOL_M <= tx <= TABLE_W + OUT_TOL_M
                and -OUT_TOL_M <= ty <= TABLE_L + OUT_TOL_M):
            n_out += 1
            continue
        # "box" = bbox pixel trên `img` (ảnh ĐÃ resize — BallID crop từ chính
        # ảnh predict, không quy đổi scale); trường nội bộ, pop trước reply
        cands.append({"x": min(max(float(tx), 0.0), TABLE_W),
                      "y": min(max(float(ty), 0.0), TABLE_L),
                      "type": "cue" if cls == "Cue" else "ball",
                      "conf": float(b.conf),
                      "box": (x1, y1, x2, y2)})

    # dedupe double-detect (một bi ra 2 box khác class): duyệt conf giảm dần,
    # tâm nào dính tâm đã giữ trong < DEDUP_DIST_M thì bỏ
    kept: list[dict] = []
    for c in sorted(cands, key=lambda c: -c["conf"]):
        if any(math.hypot(c["x"] - k["x"], c["y"] - k["y"]) < DEDUP_DIST_M
               for k in kept):
            n_dup += 1
            continue
        kept.append(c)

    # cue ĐÚNG 1: giữ cue conf cao nhất (kept đang xếp theo conf giảm dần),
    # các "Cue" còn lại hạ xuống ball — bi trắng thứ hai gần như chắc là detect
    # nhầm loại, người chơi sửa tay ở van an toàn §5.2
    seen_cue = False
    for c in kept:
        if c["type"] == "cue":
            if seen_cue:
                c["type"] = "ball"
            seen_cue = True

    balls = ([c for c in kept if c["type"] == "cue"]
             + [c for c in kept if c["type"] != "cue"])

    # BallID hai tầng (bàn giao 22): gán SỐ 1–9 theo màu cho bi thường; WB
    # theo crop bi cue ĐÃ CHỐT (sau dedupe + hạ cue thừa). Bi cue không mang
    # number; number=None khi màu không khớp số nào dưới ngưỡng.
    others = [c for c in balls if c["type"] == "ball"]
    if others:
        cue = next((c for c in balls if c["type"] == "cue"), None)
        assigns, wb = identify_balls(img, [c["box"] for c in others],
                                     cue["box"] if cue else None)
        for c, (num, num_conf) in zip(others, assigns):
            c["number"] = num
            c["number_conf"] = num_conf
            c["wb"] = wb

    for c in balls:
        c.pop("box", None)
        c["x"], c["y"] = round(c["x"], 4), round(c["y"], 4)
        c["conf"] = round(c["conf"], 3)
    return {"balls": balls,
            "dropped": {"dot": n_dot, "ngoai_ban": n_out, "trung_tam": n_dup}}


def handle_scan(payload: dict, model, conf: float, predict_kw: dict) -> dict:
    """Reply protocol của transport — CÙNG shape với engine.handle_job:
    ok/result | error "validation" (422) | error "internal" (500). Worker
    KHÔNG được chết vì một job hỏng."""
    try:
        return {"ok": True,
                "result": scan_once(payload, model, conf, predict_kw)}
    except ValueError as e:
        return {"ok": False, "error": "validation", "message": str(e)}
    except Exception as e:  # noqa: BLE001 — job hỏng ≠ worker chết
        return {"ok": False, "error": "internal",
                "message": f"{type(e).__name__}: {e}"}


# ------------------------------------------- analyze clip (bàn giao 24)

def not_shot_flag(result: dict) -> dict:
    """Van "nghi không phải cú đánh" (lát A2 phần 1, quyết định Cowork
    13/08 — hướng (a) HANDOFF A1 câu 1): cờ HIỂN THỊ dựa hoàn toàn trên
    3 tín hiệu cú TỰ KHAI đã có trong result — KHÔNG tín hiệu mới, KHÔNG
    đụng segmentation, KHÔNG xoá/lọc cú:

    1. V0 giải tích ngoài khoảng tin ``bc.V0_SANE_MPS`` [1, 8] m/s;
    2. không tìm được motion_start;
    3. coverage < ``bc.COVERAGE_WARN`` (0.8 — mốc P0).

    Ba cú ma rack_a1 (idx 7/8/9) tự khai đúng các tín hiệu này (v0 0.27 /
    None / 0.99 m/s, coverage 45.5/38.9/60%), còn 11 cú thật sạch cả ba
    (coverage thấp nhất 92.6%) — gate A2.1: bắt 3/3 ma, 0/11 oan.

    V0 = None mà motion_start CÓ (hiếm: "không đủ frame sau lúc bắt đầu
    chạy") thì tín hiệu 1 KHÔNG nổ — BRIEF chốt đúng 3 tín hiệu, "ngoài
    khoảng" chỉ xét khi có số; ca đó thường dính tín hiệu 3 nếu thật sự
    là rác. Reasons tiếng Việt server soạn sẵn (nếp spin_evidence[].text).
    """
    from poolcoach_cv import broadcast as bc

    m = result.get("metrics") or {}
    reasons: list[str] = []
    v0 = m.get("v0_mps")
    if v0 is not None and not (bc.V0_SANE_MPS[0] <= v0 <= bc.V0_SANE_MPS[1]):
        reasons.append(f"V0 giải tích {v0:.2f} m/s ngoài khoảng tin "
                       f"[{bc.V0_SANE_MPS[0]:g}, {bc.V0_SANE_MPS[1]:g}] m/s")
    if m.get("motion_start_s") is None:
        reasons.append("không tìm được lúc cue ball bắt đầu chạy "
                       "(motion_start)")
    cov = m.get("coverage")
    if cov is not None and cov < bc.COVERAGE_WARN:
        reasons.append(f"track chỉ phủ {cov:.0%} thời lượng "
                       f"(< {bc.COVERAGE_WARN:.0%})")
    return {"flagged": bool(reasons), "reasons": reasons}


def load_shotnet(device: str):
    """Load ShotNet MỘT lần lúc boot (nếp giữ YOLO ấm — BRIEF 28 bối cảnh
    6). Trả dict {model, cfg, tag, device} hoặc None (thiếu file/lỗi load —
    analyze vẫn sống, khối shotnet vắng kèm warning per-job)."""
    import torch

    from poolcoach_cv.shotnet import ShotNet, ShotNetConfig

    path = Path(os.environ.get("POOLCOACH_SHOTNET_WEIGHTS", DEFAULT_SHOTNET))
    if not path.exists():
        print(f"[cv_worker] khong thay ShotNet {path} -- analyze chay "
              f"KHONG co khoi shotnet (set POOLCOACH_SHOTNET_WEIGHTS neu "
              f"model nam cho khac)", flush=True)
        return None
    dev = device if device else ("cuda" if torch.cuda.is_available()
                                 else "cpu")
    if dev.strip().isdigit():        # nếp YOLO: device="0" nghĩa là cuda:0
        dev = f"cuda:{dev.strip()}"
    ck = torch.load(path, map_location=dev)
    cfg = ShotNetConfig(**ck["model_config"])
    net = ShotNet(cfg).to(dev)
    net.load_state_dict(ck["state_dict"])
    net.eval()
    return {"model": net, "cfg": cfg, "tag": path.parent.name,
            "device": dev}


def load_resim():
    """Load + JIT pooltool cho resim MỘT lần lúc boot (lát A2 phần 2) —
    nếp load_shotnet: thiếu/hỏng → None, analyze vẫn sống, khối resim vắng
    kèm warning per-job. JIT nguội 40–90s nên phải nằm TRƯỚC heartbeat
    (nếp engine_worker: không báo alive lúc còn đang JIT)."""
    try:
        from poolcoach_cv import resim as rs
    except Exception as e:  # noqa: BLE001
        print(f"[cv_worker] khong import duoc poolcoach_cv.resim "
              f"({type(e).__name__}: {e}) -- resim TAT", flush=True)
        return None
    state = rs.warmup()
    if state is None:
        return None
    return {"module": rs, "state": state}


def shotnet_infer(sn: dict, rows: list[dict], others: list[dict]) -> dict:
    """Track/detections app → khối ``shotnet`` cho JSON kết quả. Feature
    dựng bằng ĐÚNG code loader (shot_from_track + featurize_shot — một
    nguồn sự thật, BRIEF 28 bước 2.1); bàn thật truyền vào featurize qua
    (w, l) của broadcast."""
    import torch

    from poolcoach_cv import broadcast as bc
    from poolcoach_cv.shotnet import featurize_shot, shot_from_track, \
        spin_classes

    t0 = time.perf_counter()
    shot = shot_from_track(rows, others)
    feats, t = featurize_shot(shot, w=bc.TABLE_W_M, l=bc.TABLE_L_M,
                              deltas=sn["cfg"].use_deltas)
    dev = sn["device"]
    x = torch.from_numpy(feats)[None].to(dev)
    tt = torch.from_numpy(t)[None].to(dev)
    mask = torch.ones(1, len(t), dtype=torch.bool, device=dev)
    p = sn["model"].predict(x, tt, mask)
    a, b = float(p["a"][0]), float(p["b"][0])
    vert, _side = spin_classes(np.array([a]), np.array([b]))
    p_ident = float(p["p_ident"][0])
    conf_raw = ("high" if p_ident >= SHOTNET_CONF_HIGH else
                "medium" if p_ident >= SHOTNET_CONF_MED else "low")
    # mật độ detection bi mục tiêu của CHÍNH clip này (det không-cue/frame)
    # — đại lượng ngoài model, đo trên đúng input đã đưa vào net
    density = len(others) / max(len(rows), 1)
    conf = density_confidence(conf_raw, density)
    return {
        "model": sn["tag"],
        # V0 GẬY (thước label/net — HANDOFF 26b), quy đổi hình học về m/s
        # bàn thật; KHÁC thước metrics.v0_mps (tốc độ BI đo từ track)
        "v0_cue_mps": round(float(p["v0"][0]) * SHOTNET_V0_SCALE_REAL, 3),
        "phi_deg": round(float(p["phi_deg"][0]), 1),
        "a": round(a, 3), "b": round(b, 3),
        # cùng mapping harness BG27: side = dấu a thuần (không dead-zone),
        # vert = 3 lớp quantize theo B_STUN_MAX (spin_classes)
        "spin_vert": str(vert[0]),
        "spin_side": "side-L" if a > 0 else "side-R",
        "identifiable_prob": round(p_ident, 3),
        # `confidence` = hạng HIỂN THỊ (đã qua van mật độ); `confidence_raw`
        # = hạng thô từ head identifiable — giữ cả hai để không mất số nào
        "confidence": conf,
        "confidence_raw": conf_raw,
        "det_density": round(density, 3),
        "inference_ms": round((time.perf_counter() - t0) * 1000, 1),
        "n_target_slots": int(shot["n_balls"]) - 1,
    }


def comp_homography_cls(image_wh: tuple[float, float], ball_r: float,
                        sink: dict):
    """Subclass TableHomography có bù độ cao tâm bi (BG31 bước 1.2).

    ``px_to_table`` áp bù k = R/(h−R) theo hướng camera→điểm, camera từ
    decompose homography CỦA CHÍNH CLIP (không hằng chung). Trong
    ``analyze_clip`` MỌI toạ độ map (track cue, detection không-cue,
    balls_init frame đầu) đều đi qua ``px_to_table`` nên bù phủ MỌI vị
    trí map, không riêng gần băng. Decompose thất bại/camera vô lý →
    HeightCompensation tự về identity (không bù — thà không bù còn hơn
    bù bằng camera rác). ``sink`` nhận meta ``height_comp`` khi
    homography được dựng — worker ghi vào JSON job (bước 1.4).

    Patch vào ``bc.TableHomography`` quanh MỘT lần gọi analyze_clip —
    nếp "script-level adapter, KHÔNG sửa broadcast.py" (BG26/BG28);
    worker phục vụ job tuần tự nên không có race.
    """

    class _CompHomography(TableHomography):
        def __init__(self, corners_px, **kw) -> None:
            super().__init__(corners_px, **kw)
            self._comp = HeightCompensation.from_homography(
                self.H, image_wh, ball_r)
            sink.clear()
            sink.update(self._comp.meta())

        def px_to_table(self, pts_px):
            return self._comp.apply(super().px_to_table(pts_px))

    return _CompHomography


def clip_frame_wh(clip_path) -> tuple[float, float]:
    """(rộng, cao) pixel của clip — principal point danh định cho decompose
    (giữa ảnh, quy ước BG30). Không đọc được → (0, 0): bộ bù tự off, còn
    clip hỏng thật thì analyze_clip ném ValueError sau đó với message sẵn."""
    import cv2

    cap = cv2.VideoCapture(str(clip_path))
    try:
        return (cap.get(cv2.CAP_PROP_FRAME_WIDTH),
                cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    finally:
        cap.release()


def match_ballid_to_init(balls_init: list[dict], numbered: list[tuple],
                         tol_m: float) -> int:
    """Gắn (number, number_conf) vào entry ``balls_init`` GẦN NHẤT trong
    ``tol_m`` — thuần số học, tách riêng cho unit test (venv app, không cv2).

    ``numbered``: [(tx, ty, number, number_conf), ...] toạ độ bàn của bbox
    frame đầu, ĐÃ qua cùng homography với balls_init nên khớp gần-đúng-điểm.
    Ưu tiên conf cao gắn trước; entry đã có số không bị đè (hai bbox sát
    nhau tranh một entry thì bản conf cao thắng). Trả số entry được gắn."""
    import math

    n = 0
    for tx, ty, num, nc in sorted(numbered, key=lambda q: -q[3]):
        if num is None:
            continue
        best, bd = None, float("inf")
        for entry in balls_init:
            if entry.get("type") != "ball" or "number" in entry:
                continue
            d = math.hypot(entry["x_m"] - tx, entry["y_m"] - ty)
            if d < bd:
                best, bd = entry, d
        if best is not None and bd <= tol_m:
            best["number"] = int(num)
            best["number_conf"] = round(float(nc), 3)
            n += 1
    return n


def ballid_balls_init(clip_path, corners_px, result: dict,
                      first_dets: list) -> str | None:
    """Gắn số bi 1–9 vào ``balls_init`` (A2b phần 4) — CÙNG nguồn màu với
    overlay.mp4: ``identify_balls`` (BallID hai tầng theo màu, bàn giao 22)
    trên FRAME ĐẦU + đúng bbox mà tầng số đã dùng (RecordingModel), tâm
    bbox map qua CÙNG phép homography bù độ cao của job (dựng lại
    ``comp_homography_cls`` — decompose tất định nên cùng input là cùng
    phép map). Màu không nhận ra / kết quả cũ → entry không có trường
    ``number``, FE vẽ bi xám như trước (không bịa số).

    Trả None khi trơn, message warning khi có chuyện — kết quả SỐ không
    bị đụng trong mọi trường hợp (nếp overlay: hỏng ≠ mất số đo)."""
    import cv2

    from poolcoach_cv import broadcast as bc

    balls_init = result.get("balls_init") or []
    balls = [(cls, conf, box) for cls, conf, box in first_dets
             if cls in bc.BALL_CLASSES]
    others = [b for b in balls if b[0] != "Cue"]
    if not balls_init or not others:
        return None

    cap = cv2.VideoCapture(str(clip_path))
    try:
        ok, frame = cap.read()
    finally:
        cap.release()
    if not ok:
        return ("Không đọc lại được frame đầu để gắn số bi — bi trên bàn "
                "metric hiện xám.")

    cue = next((b for b in balls if b[0] == "Cue"), None)
    assigns, _wb = identify_balls(frame, [b[2] for b in others],
                                  cue[2] if cue else None)
    th_cls = comp_homography_cls(clip_frame_wh(clip_path), bc.BALL_R_M, {})
    th = th_cls(orient_corners(np.asarray(corners_px, dtype=np.float64)),
                table_w=bc.TABLE_W_M, table_l=bc.TABLE_L_M)
    numbered = []
    for (_cls, _conf, (x1, y1, x2, y2)), (num, nc) in zip(others, assigns):
        tx, ty = th.px_to_table(((x1 + x2) / 2.0, (y1 + y2) / 2.0))
        numbered.append((float(tx), float(ty), num, float(nc or 0.0)))
    match_ballid_to_init(balls_init, numbered, bc.BALL_DEDUP_M)
    return None


def analyze_once(payload: dict, model, conf: float, predict_kw: dict,
                 status, shotnet: dict | None = None,
                 resim: dict | None = None) -> dict:
    """MỘT job analyze: đọc clip từ đường dẫn tạm → poolcoach_cv.broadcast,
    rồi ShotNet (BG28) trên cùng track/detections — kết quả có THÊM khối
    ``shotnet`` (mở rộng thuần; ``shotnet=None`` → chỉ analytic + warning).

    Ném ValueError cho lỗi thuộc về input (clip mất/hỏng/quá dài, corners
    suy biến) — cùng phân loại với scan_once. `status(dict)` cập nhật tiến
    độ; progress callback đồng thời BEAT heartbeat vì job dài hơn TTL 15s
    nhiều lần — không beat là /api/scan lẫn /api/analyze fast-fail oan.
    """
    clip_path = payload.get("clip_path")
    if not clip_path or not Path(clip_path).exists():
        raise ValueError("Clip không còn trong thư mục tạm — upload lại "
                         "(file tạm sống theo phiên, không lưu lâu).")

    from poolcoach_cv import broadcast as bc

    def progress(stage: str, frac: float) -> None:
        jobqueue.beat(key=jobqueue.SCAN_HEARTBEAT_KEY)
        status({"status": "running", "progress": round(min(frac, 0.99), 3),
                "stage": stage})

    # ShotNet (BG28) cần đúng input của analyze_track (rows per-frame +
    # detections KHÔNG-cue) mà analyze_clip không trả ra ngoài — bắt bằng
    # wrapper mức module quanh MỘT lần gọi (adapter phía worker, nếp
    # "script-level adapter, KHÔNG sửa broadcast.py" của eval harness BG26;
    # worker phục vụ job tuần tự nên không có race). BG31 patch thêm
    # TableHomography cùng nếp: px_to_table áp bù độ cao tâm bi cho MỌI vị
    # trí map (comp_homography_cls) — ShotNet vì thế cũng ăn track ĐÃ bù.
    captured: dict = {}
    comp_meta: dict = {}
    orig_analyze_track = bc.analyze_track
    orig_homography = bc.TableHomography

    def _capture(rows, others=None):
        captured["rows"], captured["others"] = rows, others
        return orig_analyze_track(rows, others=others)

    # overlay per cú (lát A2 phần 3): bọc model để GHI LẠI đúng bbox mà
    # tầng số dùng — render không detect lại (cùng nếp adapter quanh một
    # lần gọi). Chỉ luồng video đa cú gửi overlay_path; /api/analyze cũ
    # không gửi → không đổi một byte hành vi.
    overlay_path = payload.get("overlay_path")
    rec = None
    if overlay_path:
        from poolcoach_cv.overlay import RecordingModel

        rec = RecordingModel(model)

    t0 = time.perf_counter()
    bc.analyze_track = _capture
    bc.TableHomography = comp_homography_cls(clip_frame_wh(clip_path),
                                             bc.BALL_R_M, comp_meta)
    try:
        result = bc.analyze_clip(clip_path, payload.get("corners"),
                                 rec if rec is not None else model,
                                 op_conf=conf, progress=progress,
                                 device=predict_kw.get("device"))
    finally:
        bc.analyze_track = orig_analyze_track
        bc.TableHomography = orig_homography
    result["metrics"]["elapsed_s"] = round(time.perf_counter() - t0, 2)

    # cờ quy ước toạ độ (BG31 bước 1.4): track sinh ra ĐÃ/CHƯA bù độ cao —
    # đọc lại JSON là biết thời kỳ; số cũ (BG23–30) không có cờ = chưa bù
    result["height_comp"] = dict(comp_meta) if comp_meta else {"on": False}
    if not result["height_comp"]["on"]:
        result["warnings"].append(
            "Không bù độ cao tâm bi ("
            + result["height_comp"].get("reason", "không rõ vì sao")
            + ") — toạ độ theo quy ước cũ, bi gần băng có thể lệch ra xa "
              "băng cỡ 1–3cm.")

    # khối shotnet — HAI phương pháp hiện song song (analytic giữ nguyên);
    # net lỗi/vắng model KHÔNG giết kết quả analytic, khai qua warnings
    if shotnet is None:
        result["warnings"].append(
            "ShotNet chưa load được (thiếu checkpoint?) — chỉ có số đo "
            "giải tích.")
    elif captured.get("rows"):
        progress("shotnet", 0.99)
        try:
            result["shotnet"] = shotnet_infer(shotnet, captured["rows"],
                                              captured.get("others") or [])
            # van mật độ (BG29 bước 3): CHỈ thêm một dòng cảnh báo, không
            # đụng số model và không chặn gì
            note = density_warning(result["shotnet"])
            if note:
                result["warnings"].append(note)
        except Exception as e:  # noqa: BLE001 — net hỏng ≠ mất analytic
            result["warnings"].append(
                f"ShotNet lỗi khi suy đoán ({type(e).__name__}: {e}) — "
                f"chỉ có số đo giải tích.")

    # RMSE resim (lát A2 phần 2): nạp params suy được vào pooltool sim lại
    # cú → khối ``resim`` (hai bộ shotnet/analytic, series + rmse per bộ).
    # Bước THÊM SAU analyze — không đụng một số nào có sẵn; resim hỏng/vắng
    # pooltool → khai qua warnings, không giết kết quả (nếp shotnet).
    if resim is None:
        result["warnings"].append(
            "pooltool chưa sẵn trong venv CV — không có RMSE resim.")
    else:
        progress("resim", 0.99)
        try:
            result["resim"] = resim["module"].resim_result(result,
                                                           resim["state"])
        except Exception as e:  # noqa: BLE001 — resim hỏng ≠ mất analyze
            result["warnings"].append(
                f"Resim lỗi ({type(e).__name__}: {e}) — không có RMSE.")

    # van "nghi không phải cú đánh" (lát A2 phần 1) — cờ HIỂN THỊ từ tín
    # hiệu tự khai, luôn có mặt trong kết quả mới (flagged False = đã kiểm
    # và sạch); kết quả cũ không có key = thời kỳ trước van (exclude_none)
    result["suspect_not_shot"] = not_shot_flag(result)

    # số bi frame đầu (A2b phần 4): gắn `number` vào balls_init từ CÙNG
    # nguồn màu với overlay.mp4 — chỉ luồng có RecordingModel (job per cú
    # của video đa cú); hỏng chỉ thành warning, không đụng số đo nào
    if rec is not None and rec.frames:
        try:
            note = ballid_balls_init(clip_path, payload.get("corners"),
                                     result, rec.frames[0])
            if note:
                result["warnings"].append(note)
        except Exception as e:  # noqa: BLE001 — màu hỏng ≠ mất kết quả số
            result["warnings"].append(
                f"Không gắn được số bi khung đầu ({type(e).__name__}: {e}) "
                f"— bi trên bàn metric hiện xám.")

    # overlay.mp4 per cú (lát A2 phần 3): render NGAY TRONG job, khi clip
    # CÒN trên đĩa (handle_analyze xoá clip sau khi analyze_once trả về —
    # overlay không render nổi hồi tố, bài học lần đo A1). Render cả cú bị
    # van gắn cờ (banner đỏ trên frame — giúp người kiểm bằng mắt); render
    # hỏng chỉ thành warning, không giết kết quả số.
    if rec is not None:
        from poolcoach_cv.overlay import render_shot_overlay

        progress("overlay", 0.99)
        try:
            note = render_shot_overlay(
                clip_path, payload.get("corners"), result, rec.frames,
                overlay_path, label=Path(str(overlay_path)).stem,
                beat=lambda: jobqueue.beat(key=jobqueue.SCAN_HEARTBEAT_KEY))
            result["overlay"] = {"file": Path(str(overlay_path)).name}
            if note:
                result["warnings"].append(note)
        except Exception as e:  # noqa: BLE001 — overlay hỏng ≠ mất số đo
            result["warnings"].append(
                f"Không render được overlay ({type(e).__name__}: {e}) — "
                f"số đo không ảnh hưởng.")
    return result


def handle_analyze(payload: dict, model, conf: float,
                   predict_kw: dict, shotnet: dict | None = None,
                   resim: dict | None = None) -> dict:
    """Reply protocol như handle_scan, NHƯNG người đọc thật là status key
    (API poll GET /api/analyze/{id}) — reply chỉ về result-list mồ côi rồi
    hết TTL. Status ghi best-effort: Redis rớt một nhịp giữa job không được
    giết worker (job vẫn chạy nốt, GET thấy status cũ tới khi nhịp sau ghi
    đè). Clip tạm XOÁ khi xong/lỗi-input (bản quyền, BRIEF #7); GIỮ khi lỗi
    nội bộ để còn truy được vì sao.

    Lát A1 (13/08) CHỈ THÊM ``result_path`` (tuỳ chọn trong payload — job
    per cú của luồng video đa cú): kết quả/lỗi ghi thêm ra FILE ngay tại
    đây, vì kết quả job từng MẤT THẬT khi chỉ nằm Redis TTL 1h (BG29b).
    Job /api/analyze cũ không gửi trường này — không đổi một byte hành vi.
    """
    analyze_id = str(payload.get("analyze_id", ""))
    result_path = payload.get("result_path")

    def status(data: dict) -> None:
        try:
            jobqueue.set_analyze_status(analyze_id, data)
        except jobqueue.QueueDown:
            pass

    def persist(data: dict) -> None:
        if not result_path:
            return
        try:
            astore.write_json(Path(result_path), data)
        except OSError as e:
            print(f"[cv_worker] WARNING: khong ghi duoc {result_path}: "
                  f"{type(e).__name__}: {e}", flush=True)

    def drop_clip() -> None:
        p = payload.get("clip_path")
        if p:
            Path(p).unlink(missing_ok=True)

    status({"status": "running", "progress": 0.0, "stage": "decode"})
    try:
        result = analyze_once(payload, model, conf, predict_kw, status,
                              shotnet=shotnet, resim=resim)
    except ValueError as e:
        drop_clip()
        persist({"error": "validation", "message": str(e)})
        status({"status": "error", "message": str(e)})
        return {"ok": False, "error": "validation", "message": str(e)}
    except Exception as e:  # noqa: BLE001 — job hỏng ≠ worker chết
        persist({"error": "internal",
                 "message": f"{type(e).__name__}: {e}"})
        status({"status": "error",
                "message": f"{type(e).__name__}: {e}"})
        return {"ok": False, "error": "internal",
                "message": f"{type(e).__name__}: {e}"}
    persist(result)          # file TRƯỚC Redis — nguồn sự thật nằm trên đĩa
    drop_clip()
    status({"status": "done", "progress": 1.0, "result": result})
    return {"ok": True, "result": {"analyze_id": analyze_id}}


# ------------------------------------- segment video đa cú (lát A1, 13/08)

def cut_clip(video_path: Path, t0: float, dur: float, out: Path) -> str | None:
    """Cắt MỘT clip per cú bằng ffmpeg hệ thống (re-encode h264 — seek
    chính xác tới frame, stream-copy sẽ lệch về keyframe trước đó cả giây).
    Trả None khi OK, message lỗi tiếng Việt khi hỏng. ``-an`` bỏ audio.
    ffmpeg 9.0 đã có trên máy demo (BRIEF 24 #8 — cùng máy chạy ffprobe)."""
    import shutil
    import subprocess

    exe = shutil.which("ffmpeg")
    if exe is None:
        return "máy không có ffmpeg — không cắt được clip per cú"
    out.parent.mkdir(parents=True, exist_ok=True)
    try:
        r = subprocess.run(
            [exe, "-y", "-v", "error", "-ss", f"{max(t0, 0.0):.3f}",
             "-i", str(video_path), "-t", f"{dur:.3f}", "-an",
             "-c:v", "libx264", "-preset", "veryfast", "-crf", "18",
             str(out)],
            capture_output=True, text=True, timeout=300)
    except subprocess.TimeoutExpired:
        return "ffmpeg cắt clip quá 300s — treo?"
    if r.returncode != 0 or not out.exists():
        err = (r.stderr or "ffmpeg lỗi không rõ").strip().splitlines()[-1]
        return f"ffmpeg không cắt được clip ({err[:200]})"
    return None


def segment_once(payload: dict, model, conf: float, predict_kw: dict,
                 status) -> dict:
    """MỘT job segment: video dài → danh sách cú → manifest ghi FILE ngay
    → thumbnail + clip per cú → enqueue N job analyze NGUYÊN TRẠNG.

    Ném ValueError cho lỗi thuộc về input (video mất/hỏng/quá dài, corners
    suy biến) — cùng phân loại scan/analyze. Job DÀI nhất hệ (video 5–20
    phút): progress callback beat heartbeat như analyze.
    """
    video_path = payload.get("video_path")
    if not video_path or not Path(video_path).exists():
        raise ValueError("Video không còn trên đĩa — upload lại.")
    video_path = Path(video_path)
    out_dir = Path(payload["out_dir"])
    video_id = str(payload.get("video_id", ""))

    from poolcoach_cv.segmentation import segment_video

    def progress(stage: str, frac: float) -> None:
        jobqueue.beat(key=jobqueue.SCAN_HEARTBEAT_KEY)
        status({"status": "running", "progress": round(min(frac, 0.99), 3),
                "stage": stage})

    t0 = time.perf_counter()
    seg = segment_video(video_path, payload.get("corners"), model,
                        op_conf=conf, device=predict_kw.get("device"),
                        max_dur_s=astore.VIDEO_MAX_DUR_S, progress=progress)

    # manifest per cú — trạng thái ban đầu queued (cú lành) / error (cú
    # segmentation đã kết án: mất cảnh, không đóng được ranh giới...)
    shots = []
    for s in seg["shots"]:
        idx = int(s["idx"])
        dur = float(s["t_end_s"]) - float(s["t_start_s"])
        entry = dict(s)
        entry["analyze_id"] = astore.shot_analyze_id(video_id, idx)
        entry["thumb"] = astore.thumb_path(out_dir, idx).name
        if s["status"] == "ok":
            if dur > bc_max_dur():
                entry["status"] = "error"
                entry["reason"] = (f"cú dài {dur:.1f}s vượt trần clip phân "
                                   f"tích {bc_max_dur():g}s")
            else:
                entry["status"] = "queued"
                entry["clip"] = f"clips/{astore.clip_path(out_dir, idx).name}"
        shots.append(entry)

    # thumbnail frame đầu cú (seek theo PTS tuyệt đối — cùng thước t_start)
    import cv2

    cap = cv2.VideoCapture(str(video_path))
    try:
        for entry in shots:
            jobqueue.beat(key=jobqueue.SCAN_HEARTBEAT_KEY)
            cap.set(cv2.CAP_PROP_POS_MSEC, float(entry["t_start_s"]) * 1000.0)
            ok, frame = cap.read()
            if not ok:
                continue
            h, w = frame.shape[:2]
            scale = 320.0 / max(w, 1)
            thumb = cv2.resize(frame, (320, max(1, round(h * scale))))
            cv2.imwrite(str(astore.thumb_path(out_dir, int(entry["idx"]))),
                        thumb, [cv2.IMWRITE_JPEG_QUALITY, 85])
    finally:
        cap.release()

    # cắt clip per cú lành — hỏng thì cú đó thành error kèm lý do, các cú
    # khác vẫn đi tiếp (một cú hỏng không giết cả danh sách)
    n_cut = 0
    for entry in shots:
        if entry["status"] != "queued":
            continue
        jobqueue.beat(key=jobqueue.SCAN_HEARTBEAT_KEY)
        status({"status": "running", "stage": "cut",
                "progress": round(0.92 + 0.07 * n_cut / max(len(shots), 1),
                                  3)})
        err = cut_clip(video_path,
                       float(entry["t_start_s"]),
                       float(entry["t_end_s"]) - float(entry["t_start_s"]),
                       astore.clip_path(out_dir, int(entry["idx"])))
        if err is not None:
            entry["status"] = "error"
            entry["reason"] = err
            entry.pop("clip", None)
        n_cut += 1

    manifest = {
        "status": "done",
        "video_id": video_id,
        "filename": payload.get("filename"),
        "fps_nominal": seg.get("fps_nominal"),
        "t_first_s": seg.get("t_first_s"),
        "t_last_s": seg.get("t_last_s"),
        "duration_s": seg.get("duration_s"),
        "n_frames": seg.get("n_frames"),
        "n_dup_frames": seg.get("n_dup_frames"),
        "elapsed_scan_s": round(time.perf_counter() - t0, 2),
        "warnings": seg.get("warnings", []),
        "shots": shots,
    }
    # manifest nằm trên ĐĨA trước khi bất kỳ cú nào được phân tích — chết
    # giữa chừng thì danh sách cú + lý do vẫn còn (nếp BG29b/BG31)
    astore.write_json(astore.manifest_path(out_dir), manifest)

    # enqueue job analyze per cú NGUYÊN TRẠNG — chỉ thêm result_path;
    # status "queued" ghi TRƯỚC enqueue (race đã bắt được bằng test BG24)
    for entry in shots:
        if entry["status"] != "queued":
            continue
        idx = int(entry["idx"])
        aid = entry["analyze_id"]
        try:
            jobqueue.set_analyze_status(aid, {"status": "queued",
                                              "progress": 0.0})
            jobqueue.enqueue(
                {"analyze_id": aid,
                 "clip_path": str(astore.clip_path(out_dir, idx)),
                 "corners": payload.get("corners"),
                 "result_path": str(astore.shot_json_path(out_dir, idx)),
                 # lát A2 phần 3: overlay render trong job analyze, trước
                 # khi clip per cú bị xoá
                 "overlay_path": str(astore.overlay_path(out_dir, idx))},
                jobs_key=jobqueue.ANALYZE_JOBS_KEY, job_id=aid)
        except jobqueue.QueueDown:
            # Redis rớt nhịp này: cú nằm "queued" trong manifest mà không
            # có job — GET sẽ báo unknown; đường phục hồi là upload lại
            print(f"[cv_worker] WARNING: khong enqueue duoc {aid} "
                  f"(Redis down)", flush=True)

    # video gốc XONG VIỆC (đã có manifest + thumb + clip) — xoá, không đọng
    # video bản quyền (nếp BRIEF 24 #7); kết quả per cú sống bằng file JSON
    video_path.unlink(missing_ok=True)
    n_queued = sum(1 for s in shots if s["status"] == "queued")
    return {"n_shots": len(shots), "n_queued": n_queued,
            "elapsed_scan_s": manifest["elapsed_scan_s"]}


def bc_max_dur() -> float:
    """Trần thời lượng clip analyze (30s) — đọc từ broadcast để không mang
    bản sao có thể trôi; hàm vì broadcast import lười được ở mọi venv."""
    from poolcoach_cv import broadcast as bc

    return bc.ANALYZE_MAX_DUR_S


def handle_segment(payload: dict, model, conf: float,
                   predict_kw: dict) -> dict:
    """Reply protocol như handle_analyze — người đọc thật là status key
    per video (GET /api/analyzer/shots) + manifest file. Lỗi cũng ghi
    manifest error ra ĐĨA (file-first): status key hết TTL 1h thì người
    dùng vẫn thấy vì sao video này không ra danh sách."""
    video_id = str(payload.get("video_id", ""))
    out_dir = Path(payload.get("out_dir", ""))

    def status(data: dict) -> None:
        try:
            jobqueue.set_video_status(video_id, data)
        except jobqueue.QueueDown:
            pass

    def persist_error(message: str) -> None:
        if not str(out_dir):
            return
        try:
            astore.write_json(astore.manifest_path(out_dir),
                              {"status": "error", "video_id": video_id,
                               "message": message})
        except OSError as e:
            print(f"[cv_worker] WARNING: khong ghi duoc manifest loi: "
                  f"{type(e).__name__}: {e}", flush=True)

    status({"status": "running", "progress": 0.0, "stage": "decode"})
    try:
        result = segment_once(payload, model, conf, predict_kw, status)
    except ValueError as e:
        p = payload.get("video_path")
        if p:
            Path(p).unlink(missing_ok=True)     # video hỏng — không đọng
        persist_error(str(e))
        status({"status": "error", "message": str(e)})
        return {"ok": False, "error": "validation", "message": str(e)}
    except Exception as e:  # noqa: BLE001 — job hỏng ≠ worker chết
        # GIỮ video khi lỗi nội bộ — còn truy được vì sao (nếp analyze)
        persist_error(f"{type(e).__name__}: {e}")
        status({"status": "error", "message": f"{type(e).__name__}: {e}"})
        return {"ok": False, "error": "internal",
                "message": f"{type(e).__name__}: {e}"}
    status({"status": "done", "progress": 1.0, **result})
    return {"ok": True, "result": {"video_id": video_id, **result}}


def resolve_weights() -> Path:
    """Đường dẫn weights detector — env > máy Danh > BẢN TRONG REPO.

    Thứ tự cố ý: máy phát triển giữ nguyên đường cũ ``D:\\Khoa luan\\
    cv_full_20260805\\best.pt`` nên số local KHÔNG đổi một byte; container
    (HF Spaces) không có ổ D nên rơi xuống bản force-add trong repo
    ``models/cv_full_20260805/best.pt`` — cùng file, chép byte-for-byte.
    """
    env = os.environ.get("POOLCOACH_CV_WEIGHTS")
    if env:
        return Path(env)
    for p in (Path(DEFAULT_WEIGHTS),
              ROOT / "models" / "cv_full_20260805" / "best.pt"):
        if p.exists():
            return p
    return Path(DEFAULT_WEIGHTS)      # báo lỗi với đường quen thuộc


def load_all(conf: float, device: str, predict_kw: dict):
    """Nạp YOLO + ShotNet + pooltool(resim) MỘT lần — phần "khởi động" của
    worker, tách ra để chế độ nhúng (app/localcv.py) dùng CHUNG code, không
    có bản sao thứ hai của thứ tự nạp/warmup.

    Ném FileNotFoundError khi thiếu weights detector; ShotNet/resim thiếu
    thì trả None (analyze vẫn sống kèm warning per-job — nếp cũ)."""
    weights = resolve_weights()
    if not weights.exists():
        raise FileNotFoundError(
            f"khong thay weights {weights} -- set POOLCOACH_CV_WEIGHTS neu "
            f"model nam cho khac")
    t0 = time.time()
    from ultralytics import YOLO

    model = YOLO(str(weights))
    # warmup 1 predict anh den: dung CUDA context + fuse model ngay bay gio,
    # de job dau tien khong phai tra gia (cung ly do JIT warmup cua engine)
    model.predict(np.zeros((IMGSZ, IMGSZ, 3), dtype=np.uint8), conf=conf,
                  imgsz=IMGSZ, verbose=False, **predict_kw)
    # ShotNet (BG28): load MOT lan luc boot nhu YOLO — 10.7MB, ~1s; thieu
    # file thi None + analyze van song (warning per-job)
    shotnet = load_shotnet(device)
    # Resim pooltool (lat A2): JIT nguoi 40-90s — TRUOC heartbeat, nhu JIT
    # cua engine_worker; thieu pooltool thi None + analyze van song
    resim = load_resim()
    print(f"[cv_worker] model {weights.name} san sang sau "
          f"{time.time() - t0:.1f}s (conf={conf:g}"
          f"{', device=' + device if device else ''}"
          f", shotnet={'OK ' + shotnet['tag'] if shotnet else 'KHONG'}"
          f", resim={'OK' if resim else 'KHONG'})"
          f" -- bat heartbeat, cho job tren "
          f"'{jobqueue.SCAN_JOBS_KEY}'", flush=True)
    return model, shotnet, resim


def serve_forever(model, conf: float, predict_kw: dict, shotnet, resim,
                  should_stop=None) -> int:
    """Vòng phục vụ 3 queue — thân của worker, dùng chung với chế độ nhúng.

    ``should_stop()`` (tuỳ chọn) cho chế độ nhúng thoát sạch khi app tắt;
    worker process không truyền, thoát bằng Ctrl+C như cũ.
    """
    import redis

    served = 0
    try:
        while True:
            if should_stop is not None and should_stop():
                print(f"[cv_worker] dung theo yeu cau sau {served} job",
                      flush=True)
                break
            try:
                jobqueue.beat(key=jobqueue.SCAN_HEARTBEAT_KEY)
                t_job = time.perf_counter()
                # 3 queue thay phiên, lát chờ ngắn (3×2s ≈ nhịp 5s cũ) để
                # heartbeat vẫn beat đều; scan giữ LƯỢT ĐẦU — job tương
                # tác nhanh không phải xếp sau một analyze đang nằm queue.
                # segment xếp CUỐI (lát A1): một job segment đẻ N job
                # analyze, và analyze phải chạy hết trước khi nhận video
                # mới — danh sách cú của video đang xem hiện dần đúng nhịp.
                if jobqueue.serve_one(
                        lambda p: handle_scan(p, model, conf, predict_kw),
                        timeout_s=2, jobs_key=jobqueue.SCAN_JOBS_KEY):
                    served += 1
                    print(f"[cv_worker] scan job #{served} xong sau "
                          f"{time.perf_counter() - t_job:.2f}s", flush=True)
                elif jobqueue.serve_one(
                        lambda p: handle_analyze(p, model, conf, predict_kw,
                                                 shotnet=shotnet,
                                                 resim=resim),
                        timeout_s=2, jobs_key=jobqueue.ANALYZE_JOBS_KEY):
                    served += 1
                    print(f"[cv_worker] analyze job #{served} xong sau "
                          f"{time.perf_counter() - t_job:.2f}s", flush=True)
                elif jobqueue.serve_one(
                        lambda p: handle_segment(p, model, conf, predict_kw),
                        timeout_s=2, jobs_key=jobqueue.SEGMENT_JOBS_KEY):
                    served += 1
                    print(f"[cv_worker] segment job #{served} xong sau "
                          f"{time.perf_counter() - t_job:.2f}s", flush=True)
            except redis.RedisError as e:
                # Redis chet/khoi dong lai KHONG duoc giet worker da load model
                print(f"[cv_worker] mat ket noi Redis ({type(e).__name__}: "
                      f"{e}) -- thu lai sau 2s", flush=True)
                time.sleep(2)
    except KeyboardInterrupt:
        print(f"\n[cv_worker] Ctrl+C -- thoat sach sau {served} job",
              flush=True)
    finally:
        jobqueue.clear_heartbeat(key=jobqueue.SCAN_HEARTBEAT_KEY)
        if should_stop is None:
            # chỉ worker PROCESS mới được đóng transport; chế độ nhúng dùng
            # CHUNG client với app — teardown ở đây là giết luôn app
            jobqueue.teardown()
    return 0


def main() -> int:
    if not os.environ.get("REDIS_URL"):
        print("cv_worker: thieu env REDIS_URL -- worker chi co nghia o mode "
              "queue (vi du: redis://localhost:6379/0). Chay CHUNG tien "
              "trinh voi app thi dat POOLCOACH_LOCAL_CV=1 khi chay app.",
              flush=True)
        return 2
    conf = float(os.environ.get("POOLCOACH_CV_CONF", OP_CONF))
    device = os.environ.get("POOLCOACH_CV_DEVICE", "")
    predict_kw = {"device": device} if device else {}
    jobqueue.setup()
    try:
        model, shotnet, resim = load_all(conf, device, predict_kw)
    except FileNotFoundError as e:
        print(f"cv_worker: {e}", flush=True)
        return 2
    return serve_forever(model, conf, predict_kw, shotnet, resim)


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