# -*- coding: utf-8 -*- """BG30 Bước 2 — đo offset field sự kiện chạm băng trên 10 cú pilot + fit mô hình độ cao tâm bi. Chẩn đoán, KHÔNG sửa pipeline: BG29b thấy apex track dừng ở ~2R trước băng (hụt ~27.5–32.6mm so với tâm-bi-chạm-băng = 1R) trên cú 11/12, đối chứng băng dài giữa bàn ~1.1R. Câu hỏi: lệch có khớp mô hình "tâm bi cao hơn mặt vải 1R chiếu phối cảnh qua homography mặt vải" không? Hình học mô hình (fit trực tiếp — cách đã chọn, BRIEF cho tự quyết): camera ở C = (Cx, Cy, h) trên mặt vải z=0; tâm bi ở P = (x, y, R). Tia C→P cắt mặt vải tại X' = P_xy + (P_xy − C_xy)·R/(h − R) — pipeline map tâm bbox qua homography MẶT VẢI nên vị trí track chính là X', lệch khỏi P_xy một vec-tơ hướng RA XA chân camera, độ lớn tăng theo khoảng cách. Tại sự kiện chạm băng, tâm bi thật cách mép đúng R ⇒ apex đo được ≈ R + (R/(h−R))·((P−C)·n̂_in) + lượng tử hoá frame (≥0). Fit (Cx, Cy, h) bằng least squares trên chính offset field; ĐỐI CHIẾU độc lập bằng decompose từng homography (tự ước f từ ràng buộc trực giao r1⊥r2, principal point danh định giữa ảnh 1920×1080) → so hai đường có ra cùng camera không. Sự kiện băng = đảo chiều thành phần vuông góc với một trong 4 mép bàn: local minimum của khoảng-cách-tới-mép trên track đã dedup (luật dedup của chính broadcast.py), có pha TỚI (≥ APPROACH_MIN, v⊥ ≥ V_PERP_MIN) và pha LÙI (≥ RECEDE_MIN) quanh apex, apex ≤ NEAR_M. Cờ loại trừ (không vào fit, vẫn báo): gần bi tĩnh ≤ CONTACT_BALL_M (nghi chạm bi), gần miệng lỗ ≤ POCKET_NEAR_M (hình học jaws không phải băng), gap frame quanh apex. Ba nguồn sự kiện, ba PHIÊN CLICK khác nhau của cùng MỘT camera vật lý (đã xác minh cu11/cu12.mp4 cùng trận/cùng góc máy với pilot — frame đầu): - "pilot": shot_01..10 (track.csv + homography.json trên đĩa); - "bg28": track cú 11/12 trong JSON kết quả BG28 (poolcoach-docs, chỉ đọc) — chạy CÙNG hàm đo sự kiện trên track đó; - "bg29b": 3 apex băng gần 56.1/57.6/61.2mm — track hết TTL Redis, số chép nguyên văn HANDOFF 13/08 10:25 (vị trí dọc mép mượn từ track BG28 cùng cú — chỉ ảnh hưởng bậc hai lên hình chiếu vuông góc). Mô hình mở rộng kiểm luôn giả thuyết MẶT PHẲNG CLICK: nếu 4 góc được chấm ở độ cao z_c trên vải (miệng lỗ trên mặt gỗ ~30–40mm thay vì đúng mặt vải), homography chuẩn theo mặt z=z_c và offset = ((R−z_c)/(h−z_c))·proj — z_c=0 là mô hình "1R qua homography mặt vải" nguyên bản; fit z_c RIÊNG TỪNG PHIÊN click với camera chung từ decompose. Chạy (venv app, thuần CPU/numpy + scipy): python scripts/broadcast/diag_cushion_offset.py Artifact (mới, không đè): D:\\Khoa luan\\bb9_diag30_cushion_events.csv, bb9_diag30_offset_fit.csv, bb9_diag30_offset_field.png. Unit test hàm đo sự kiện: tests/test_diag_cushion_offset.py. """ from __future__ import annotations import csv import json import math import subprocess import sys from pathlib import Path import numpy as np ROOT = Path(__file__).resolve().parents[2] for _p in (ROOT / "src",): if str(_p) not in sys.path: sys.path.insert(0, str(_p)) from poolcoach_cv import broadcast as bc # noqa: E402 — thuần numpy WORKSPACE = ROOT.parent PILOT_DIR = ROOT / "datasets" / "bb9_pilot" OUT_EVENTS = WORKSPACE / "bb9_diag30_cushion_events.csv" OUT_FIT = WORKSPACE / "bb9_diag30_offset_fit.csv" OUT_PNG = WORKSPACE / "bb9_diag30_offset_field.png" R = bc.BALL_R_M # 0.028575 m — cùng bi pipeline W, L = bc.TABLE_W_M, bc.TABLE_L_M BG28_DIR = WORKSPACE / "poolcoach-docs" / "results" / "2026-08-12_shotnet_p2" # 3 apex băng-gần BG29b — nguyên văn HANDOFF 13/08 10:25 (mm); rail y0 hệ # cú (= yL hệ chuẩn — camera phía y<0 hệ cú, xác minh bằng frame clip); # x dọc mép mượn track BG28 cùng cú (ảnh hưởng bậc hai). BG29B_EVENTS = ( {"shot": "cu11_lan1", "apex_mm": 56.1, "x_m": 0.335, "v_perp": 0.5}, {"shot": "cu11_lan2", "apex_mm": 57.6, "x_m": 0.335, "v_perp": 0.5}, {"shot": "cu12_t2.6", "apex_mm": 61.2, "x_m": 0.423, "v_perp": 0.66}, ) NEAR_M = 0.15 # apex ≤ ~5.2R mới coi là ứng viên sự kiện băng APPROACH_MIN_M = 0.03 # pha tới phải áp sát thêm ≥ 30mm trong cửa sổ RECEDE_MIN_M = 0.02 # pha lùi phải rời ra ≥ 20mm (lăn chậm tắt dần) V_PERP_MIN_MPS = 0.10 # v⊥ vào băng tối thiểu (trên V_STILL 0.055) EVENT_WIN_S = 0.40 # cửa sổ đo pha tới/lùi quanh apex EVENT_DEDUP_S = 0.50 # 2 apex cùng mép cách < mức này = một sự kiện POCKET_NEAR_M = 0.12 # cách miệng lỗ (góc + lỗ giữa băng dài) — jaws GAP_MAX_FACTOR = 2.5 # Δt quanh apex > 2.5× dt median → cờ "gap" # 4 mép: (tên hệ-cú, hàm khoảng cách, pháp tuyến hướng VÀO bàn) RAILS = ( ("x0", lambda x, y: x, (1.0, 0.0)), ("xW", lambda x, y: W - x, (-1.0, 0.0)), ("y0", lambda x, y: y, (0.0, 1.0)), ("yL", lambda x, y: L - y, (0.0, -1.0)), ) # miệng lỗ trên hệ bàn: 4 góc + 2 lỗ giữa hai băng DÀI (bàn 9ft) POCKETS = ((0.0, 0.0), (W, 0.0), (0.0, L), (W, L), (0.0, L / 2), (W, L / 2)) # ------------------------------------------------------- đo sự kiện băng def dedup_track(rows: list[dict]) -> tuple[np.ndarray, np.ndarray, float]: """CHÍNH luật của analyze_track (broadcast.py:600-611): bỏ frame trùng theo img_diff + chỉ giữ frame covered. Trả (t, xy, dt_med_all).""" names = [str(r["frame_file"]) for r in rows] t_all = np.array([float(r["t_s"]) for r in rows]) covered = np.array([str(r.get("covered", "0")) in ("1", "True", "true") for r in rows]) diffs = [float(r.get("img_diff", -1.0)) for r in rows] dups = bc.find_dup_frames(names, diffs) keep = [i for i in range(len(rows)) if covered[i] and names[i] not in dups] t = t_all[keep] xy = (np.array([[float(rows[i]["table_x_m"]), float(rows[i]["table_y_m"])] for i in keep]) if keep else np.empty((0, 2))) dt_med = float(np.median(np.diff(t_all))) if len(t_all) > 1 else 1 / 30 return t, xy, dt_med def find_rail_events(t: np.ndarray, xy: np.ndarray, t_from: float, dt_med: float) -> list[dict]: """Sự kiện đảo chiều gần băng trên track đã dedup, từ ``t_from`` (motion_start). Thuần numpy, unit-test được. Mỗi sự kiện: {rail, t_s, x_m, y_m, apex_mm, offset_mm, v_in_mps, v_perp_in_mps, v_perp_out_mps, quant_mm, gap_flag}.""" events: list[dict] = [] if len(t) < 3: return events for rail, dist_fn, _n_in in RAILS: d = dist_fn(xy[:, 0], xy[:, 1]) cand: list[int] = [] for i in range(1, len(t) - 1): if t[i] < t_from or d[i] > NEAR_M: continue if d[i] <= d[i - 1] and d[i] < d[i + 1]: cand.append(i) picked: list[int] = [] for i in cand: pre = np.flatnonzero((t >= t[i] - EVENT_WIN_S) & (t < t[i])) post = np.flatnonzero((t > t[i]) & (t <= t[i] + EVENT_WIN_S)) if not len(pre) or not len(post): continue appr = float(d[pre].max() - d[i]) rec = float(d[post].max() - d[i]) j0 = int(pre[np.argmax(d[pre])]) v_perp_in = appr / max(float(t[i] - t[j0]), 1e-6) if appr < APPROACH_MIN_M or rec < RECEDE_MIN_M \ or v_perp_in < V_PERP_MIN_MPS: continue picked.append(i) # dedup: cùng mép, cách < EVENT_DEDUP_S → giữ apex sâu nhất picked.sort() groups: list[list[int]] = [] for i in picked: if groups and t[i] - t[groups[-1][-1]] < EVENT_DEDUP_S: groups[-1].append(i) else: groups.append([i]) for g in groups: i = int(g[int(np.argmin(d[np.asarray(g)]))]) pre = np.flatnonzero((t >= t[i] - EVENT_WIN_S) & (t < t[i])) post = np.flatnonzero((t > t[i]) & (t <= t[i] + EVENT_WIN_S)) j0 = int(pre[np.argmax(d[pre])]) j1 = int(post[np.argmax(d[post])]) v_perp_in = float((d[j0] - d[i]) / max(t[i] - t[j0], 1e-6)) v_perp_out = float((d[j1] - d[i]) / max(t[j1] - t[i], 1e-6)) chord = xy[i] - xy[j0] v_in = float(np.hypot(*chord) / max(t[i] - t[j0], 1e-6)) gap = (float(t[i] - t[i - 1]) > GAP_MAX_FACTOR * dt_med or float(t[i + 1] - t[i]) > GAP_MAX_FACTOR * dt_med) \ if 0 < i < len(t) - 1 else True events.append({ "rail": rail, "t_s": round(float(t[i]), 3), "x_m": round(float(xy[i, 0]), 4), "y_m": round(float(xy[i, 1]), 4), "apex_mm": round(float(d[i]) * 1000, 1), "offset_mm": round((float(d[i]) - R) * 1000, 1), "v_in_mps": round(v_in, 3), "v_perp_in_mps": round(v_perp_in, 3), "v_perp_out_mps": round(v_perp_out, 3), # lượng tử hoá frame MỘT PHÍA: min trên mẫu rời chỉ có thể # ĐO THỪA, tối đa ~min(v_in,v_out)·dt/2 "quant_mm": round(min(v_perp_in, max(v_perp_out, 1e-9)) * dt_med / 2 * 1000, 1), "gap_flag": int(gap), }) events.sort(key=lambda e: e["t_s"]) return events def flag_events(events: list[dict], statics: list[tuple[float, float]] ) -> None: """Cờ loại-khỏi-fit: gần bi tĩnh (nghi chạm bi thay vì băng), gần miệng lỗ (hình học jaws). Chỉ CỜ, không xoá — bảng báo đủ.""" for e in events: p = (e["x_m"], e["y_m"]) d_ball = (min(math.hypot(p[0] - sx, p[1] - sy) for sx, sy in statics) if statics else math.inf) e["near_static_mm"] = (round(d_ball * 1000, 1) if math.isfinite(d_ball) else "") e["ball_flag"] = int(d_ball <= bc.CONTACT_BALL_M) e["pocket_flag"] = int(min(math.hypot(p[0] - px, p[1] - py) for px, py in POCKETS) <= POCKET_NEAR_M) # ------------------------------------------- hướng camera + hệ chuẩn hoá def camera_side(H: np.ndarray) -> dict: """Phía camera trong hệ bàn CỦA CÚ, từ homography ảnh→bàn: đáy ảnh (y_px lớn) là phía gần camera. Trả {t_bot, flip} — ``flip`` True nghĩa là phải xoay 180° để về hệ CHUẨN 'camera phía y > L'.""" bot = np.array([960.0, 1070.0, 1.0]) @ np.asarray(H).T bot = bot[:2] / bot[2] return {"t_bot": (round(float(bot[0]), 3), round(float(bot[1]), 3)), "flip": bool(bot[1] < L / 2)} def to_canonical(e: dict, flip: bool) -> tuple[float, float, str]: """(x, y, rail) của sự kiện về hệ chuẩn (camera phía y > L).""" x, y, rail = e["x_m"], e["y_m"], e["rail"] if flip: x, y = W - x, L - y rail = {"x0": "xW", "xW": "x0", "y0": "yL", "yL": "y0"}[rail] return x, y, rail # --------------------------------------------------- mô hình chiếu + fit _N_IN = {"x0": np.array([1.0, 0.0]), "xW": np.array([-1.0, 0.0]), "y0": np.array([0.0, 1.0]), "yL": np.array([0.0, -1.0])} _RAIL_PT = {"x0": lambda p: np.array([0.0, p[1]]), "xW": lambda p: np.array([W, p[1]]), "y0": lambda p: np.array([p[0], 0.0]), "yL": lambda p: np.array([p[0], L])} def event_proj_m(cam: np.ndarray, ev_xy: np.ndarray, rails: list[str], k: float) -> np.ndarray: """Hình chiếu (P−C)·n̂_in (m) từng sự kiện, P = tâm bi thật ước từ apex đo (2 vòng lặp trừ vec-tơ lệch k·(P−C) để sửa thành phần dọc mép).""" cx, cy, h = cam out = np.empty(len(rails)) for i, rail in enumerate(rails): n_in = _N_IN[rail] p = ev_xy[i].copy() delta = np.zeros(2) for _ in range(2): base = _RAIL_PT[rail](p) center = base + n_in * R # tâm bi thật cách mép đúng R delta = (center - (cx, cy)) * k p = ev_xy[i] - delta out[i] = float((center - (cx, cy)) @ n_in) return out def k_coeff(h: float, z_c: float) -> float: """offset ⊥ = k·proj với k = (R − z_c)/(h − R): giao tia camera→tâm-bi với mặt phẳng chuẩn z=z_c của homography (z_c=0 = mô hình '1R qua homography mặt vải' nguyên bản; z_c>R đổi DẤU — lệch về phía camera).""" return (R - z_c) / max(h - R, 1e-6) def fit_session_zc(h: float, proj: np.ndarray, offset_m: np.ndarray, w: np.ndarray) -> tuple[float, float]: """Nghiệm đóng weighted-LSQ k_s = Σw·proj·off / Σw·proj² cho MỘT phiên click, rồi z_c = R − k_s·(h − R). Trả (k_s, z_c_mm).""" k_s = float((w * proj * offset_m).sum() / max((w * proj ** 2).sum(), 1e-12)) return k_s, (R - k_s * (h - R)) * 1000 # ------------------------------------- decompose homography (đối chiếu) def decompose_h(H: np.ndarray, pp=(960.0, 540.0)) -> dict | None: """Ước camera từ MỘT homography, không cần offset field: G = H⁻¹ (bàn→ảnh) ≅ K[r1 r2 t]; tự ước f từ ràng buộc r1⊥r2 và ‖r1‖=‖r2‖ (principal point danh định giữa ảnh). Trả {f_px, cx, cy, h} hệ bàn CỦA CÚ, hoặc None nếu nghiệm f không dương.""" G = np.linalg.inv(np.asarray(H, dtype=np.float64)) G = G / G[2, 2] a, b = G[:, 0].copy(), G[:, 1].copy() a[0] -= pp[0] * a[2] a[1] -= pp[1] * a[2] b[0] -= pp[0] * b[2] b[1] -= pp[1] * b[2] # sau khi khử principal point: m_i = (u_i/f, v_i/f, w_i) — hai ràng buộc # trên f²: u1u2 + v1v2 + f²w1w2 = 0 ; (u1²+v1²) − (u2²+v2²) = f²(w2²−w1²) f2 = [] den = a[2] * b[2] if abs(den) > 1e-15: f2.append(-(a[0] * b[0] + a[1] * b[1]) / den) den = b[2] ** 2 - a[2] ** 2 if abs(den) > 1e-15: f2.append(((a[0] ** 2 + a[1] ** 2) - (b[0] ** 2 + b[1] ** 2)) / den) f2 = [v for v in f2 if v > 1e3] if not f2: return None f = math.sqrt(float(np.mean(f2))) Kinv = np.array([[1 / f, 0, -pp[0] / f], [0, 1 / f, -pp[1] / f], [0, 0, 1.0]]) M = Kinv @ G lam = 2.0 / (np.linalg.norm(M[:, 0]) + np.linalg.norm(M[:, 1])) r1, r2, tvec = lam * M[:, 0], lam * M[:, 1], lam * M[:, 2] r3 = np.cross(r1, r2) Rm = np.stack([r1, r2, r3], axis=1) cam = -Rm.T @ tvec # hmm: X_cam = Rm·X_table + t ⇒ C_table = −Rmᵀ t if cam[2] < 0: # nghiệm λ âm — lật dấu (camera phải ở TRÊN mặt bàn) cam = -cam return {"f_px": round(f, 1), "cx": float(cam[0]), "cy": float(cam[1]), "h": float(cam[2])} # ------------------------------------------------------------------ main def git_head() -> str: try: return subprocess.run(["git", "-C", str(ROOT), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() except Exception: return "unknown" def load_shot(shot_dir: Path): rows = list(csv.DictReader((shot_dir / "track.csv").open( encoding="utf-8"))) others = [] with (shot_dir / "detections.csv").open(encoding="utf-8") as f: for d in csv.DictReader(f): if d["cls"] != "Cue" and d.get("in_table", "1") == "1": others.append({"t_s": float(d["t_s"]), "x_m": float(d["table_x_m"]), "y_m": float(d["table_y_m"])}) hom = json.loads((shot_dir / "homography.json").read_text( encoding="utf-8")) return rows, others, np.asarray(hom["H"], dtype=np.float64) def load_bg28_events() -> list[dict]: """Chạy CHÍNH ``find_rail_events`` trên track cú 11/12 lưu trong JSON kết quả BG28 (chỉ đọc). Track đó là output analyze_track: đã dedup, chỉ điểm covered, t_s tương đối. Bi tĩnh xấp xỉ bằng ``balls_init`` không-cue (frame đầu). Hệ cú → hệ chuẩn: flip 180° (camera phía y<0 hệ cú — xác minh bằng frame clip: cú xuất phát đầu XA có y≈2.2).""" out = [] for cu, fname in ((11, "cu11_app_bg28.json"), (12, "cu12_app_bg28.json")): r = json.loads((BG28_DIR / fname).read_text(encoding="utf-8"))[ "result"] t = np.array([p["t_s"] for p in r["track"]]) xy = np.array([[p["x_m"], p["y_m"]] for p in r["track"]]) dt_med = float(np.median(np.diff(t))) events = find_rail_events(t, xy, float(r["metrics"] ["motion_start_s"]), dt_med) statics = [(b["x_m"], b["y_m"]) for b in r.get("balls_init", []) if b.get("type") != "cue"] flag_events(events, statics) for e in events: ecx, ecy, rail_c = to_canonical(e, flip=True) near = [c for c in r["collisions"] if abs(c["t_s"] - e["t_s"]) <= 0.25] e.update({"session": "bg28", "shot": f"cu{cu}", "t_rel_s": e["t_s"], "x_canon_m": round(ecx, 4), "y_canon_m": round(ecy, 4), "rail_canon": rail_c, "analyze_contact": (near[0].get("contact", "") if near else "KHONG-BAT-DUOC")}) out.append(e) return out def bg29b_events() -> list[dict]: """3 sự kiện băng-gần từ HANDOFF (hằng ở đầu file) → cùng shape.""" out = [] for s in BG29B_EVENTS: apex_m_ = s["apex_mm"] / 1000 e = {"session": "bg29b", "shot": s["shot"], "rail": "y0", "t_rel_s": "", "t_s": "", "x_m": s["x_m"], "y_m": round(apex_m_, 4), "apex_mm": s["apex_mm"], "offset_mm": round(s["apex_mm"] - R * 1000, 1), "v_in_mps": "", "v_perp_in_mps": s["v_perp"], "v_perp_out_mps": "", "quant_mm": round(s["v_perp"] * (1 / 30) / 2 * 1000, 1), "near_static_mm": "", "ball_flag": 0, "pocket_flag": 0, "gap_flag": 0, "analyze_contact": "unknown (HANDOFF)"} ecx, ecy, rail_c = to_canonical(e, flip=True) e.update({"x_canon_m": round(ecx, 4), "y_canon_m": round(ecy, 4), "rail_canon": rail_c}) out.append(e) return out def main() -> int: commit = git_head() all_events: list[dict] = [] cam_rows: list[dict] = [] for shot_dir in sorted(PILOT_DIR.glob("shot_*")): rows, others, H = load_shot(shot_dir) out = bc.analyze_track(rows, others=others) ms_rel = out["metrics"]["motion_start_s"] t, xy, dt_med = dedup_track(rows) if ms_rel is None or len(t) < 3: print(f"{shot_dir.name}: khong co motion_start -- bo qua") continue t0_abs = float(rows[0]["t_s"]) ms_abs = t0_abs + float(ms_rel) events = find_rail_events(t, xy, ms_abs, dt_med) statics = bc._static_balls(others, ms_abs) flag_events(events, statics) side = camera_side(H) dec = decompose_h(H) cam_rows.append({"shot": shot_dir.name, "flip_180": int(side["flip"]), "t_bot_x": side["t_bot"][0], "t_bot_y": side["t_bot"][1], **({f"dec_{k}": (round(v, 3) if isinstance(v, float) else v) for k, v in dec.items()} if dec else {})}) for e in events: ecx, ecy, rail_c = to_canonical(e, side["flip"]) # cross-ref va chạm analyze gần nhất — collisions của # analyze_track giữ NGUYÊN timebase của rows (tuyệt đối với # track.csv pilot), so bằng t_s tuyệt đối near = [c for c in out["collisions"] if abs(c["t_s"] - e["t_s"]) <= 0.25] e.update({"session": "pilot", "shot": shot_dir.name, "t_rel_s": round(e["t_s"] - t0_abs, 3), "x_canon_m": round(ecx, 4), "y_canon_m": round(ecy, 4), "rail_canon": rail_c, "analyze_contact": (near[0].get("contact", "") if near else "KHONG-BAT-DUOC")}) all_events.append(e) print(f"{shot_dir.name}: {len(events)} su kien bang " f"(flip={int(side['flip'])}, " f"dec_h={dec['h']:.3f})" if dec else f"{shot_dir.name}: {len(events)} su kien bang (dec FAIL)") all_events += load_bg28_events() all_events += bg29b_events() n_pilot = sum(e["session"] == "pilot" for e in all_events) if not all_events: print("KHONG tim thay su kien bang nao -- dung o day.") return 1 for e in all_events: e["commit_luc_chay"] = commit cols = ["session", "shot", "t_rel_s", "rail", "x_m", "y_m", "rail_canon", "x_canon_m", "y_canon_m", "apex_mm", "offset_mm", "v_in_mps", "v_perp_in_mps", "v_perp_out_mps", "quant_mm", "near_static_mm", "ball_flag", "pocket_flag", "gap_flag", "analyze_contact", "t_s", "commit_luc_chay"] if OUT_EVENTS.exists(): print(f"BO QUA ghi {OUT_EVENTS}: da ton tai (khong de artifact cu)") else: with OUT_EVENTS.open("w", newline="", encoding="utf-8") as f: wr = csv.DictWriter(f, fieldnames=cols) wr.writeheader() wr.writerows([{k: e.get(k, "") for k in cols} for e in all_events]) print(f"da ghi {OUT_EVENTS} ({len(all_events)} su kien, " f"{n_pilot} tu pilot)") # ---- camera từ decompose homography (median 10 cú, hệ chuẩn) dec_can = [] for cr in cam_rows: if "dec_h" not in cr: continue dx, dy = cr["dec_cx"], cr["dec_cy"] if cr["flip_180"]: dx, dy = W - dx, L - dy dec_can.append((dx, dy, cr["dec_h"], cr.get("dec_f_px", 0.0))) dec_arr = np.array([d[:3] for d in dec_can], dtype=float) cam = np.median(dec_arr, axis=0) dec_spread = dec_arr.std(axis=0) f_med = float(np.median([d[3] for d in dec_can])) print(f"\nCAMERA tu decompose {len(dec_can)} homography pilot " f"(he chuan, camera phia y>L):") print(f" C = ({cam[0]:+.3f} +- {dec_spread[0]:.3f}, " f"{cam[1]:+.3f} +- {dec_spread[1]:.3f}) m, " f"h = {cam[2]:.3f} +- {dec_spread[2]:.3f} m, f ~ {f_med:.0f} px") # ---- dự đoán + phần dư, hai mô hình clean = [e for e in all_events if not (e["ball_flag"] or e["pocket_flag"] or e["gap_flag"])] print(f"su kien sach (vao mo hinh): {len(clean)}/{len(all_events)} " f"(co ball {sum(e['ball_flag'] for e in all_events)}, " f"pocket {sum(e['pocket_flag'] for e in all_events)}, " f"gap {sum(e['gap_flag'] for e in all_events)})") if sum(e["session"] == "pilot" for e in clean) < 5: print("DUOI ~5 su kien pilot sach — theo BRIEF: hoi Danh truoc " "khi ket luan.") ev_xy = np.array([[e["x_canon_m"], e["y_canon_m"]] for e in clean]) rails = [e["rail_canon"] for e in clean] offs_m = np.array([e["offset_mm"] for e in clean]) / 1000.0 w = 1.0 / (np.array([max(float(e["quant_mm"]), 0.5) for e in clean]) / 2 + 3.0) ** 2 k0 = k_coeff(cam[2], 0.0) proj = event_proj_m(cam, ev_xy, rails, k0) pred0 = k0 * proj resid0 = (offs_m - pred0) * 1000 # mô hình 2: z_c RIÊNG từng phiên click (camera chung) sessions = sorted({e["session"] for e in clean}) zc_by_s: dict[str, tuple[float, float]] = {} resid1 = np.empty(len(clean)) for s in sessions: m = np.array([e["session"] == s for e in clean]) k_s, zc = fit_session_zc(cam[2], proj[m], offs_m[m], w[m]) zc_by_s[s] = (k_s, zc) resid1[m] = (offs_m[m] - k_s * proj[m]) * 1000 def med(a): return float(np.median(np.abs(np.asarray(a)))) if len(a) else np.nan print(f"\n{'session':<8}{'shot':<12}{'rail':>5}{'proj_m':>8}" f"{'offset':>8}{'pred_z0':>9}{'du_z0':>7}{'pred_zs':>9}" f"{'du_zs':>7}{'quant':>7}") for i, e in enumerate(clean): _k = zc_by_s[e["session"]][0] print(f"{e['session']:<8}{e['shot']:<12}{e['rail_canon']:>5}" f"{proj[i]:>+8.2f}{e['offset_mm']:>+8.1f}" f"{pred0[i] * 1000:>+9.1f}{resid0[i]:>+7.1f}" f"{_k * proj[i] * 1000:>+9.1f}{resid1[i]:>+7.1f}" f"{e['quant_mm']:>7.1f}") big = np.abs(offs_m) * 1000 >= 15.0 # nhóm offset LỚN — ca đang chẩn đoán print(f"\nMO HINH 1 — 'tam bi cao 1R qua homography MAT VAI' (z_c=0), " f"k = {k0:.5f}:") print(f" nhom offset >= 15mm (n={int(big.sum())}): median |du| = " f"{med(resid0[big]):.1f} mm / median |offset| = " f"{med(offs_m[big] * 1000):.1f} mm " f"(nguong 1/3: {med(offs_m[big] * 1000) / 3:.1f})") print(f" toan bo (n={len(clean)}): median |du| = {med(resid0):.1f} mm") print(f"MO HINH 2 — mat phang click z_c rieng tung phien (camera chung):") for s in sessions: m = np.array([e["session"] == s for e in clean]) print(f" {s:<7} k={zc_by_s[s][0]:+.5f} z_c={zc_by_s[s][1]:+6.1f}mm" f" median |du|={med(resid1[m]):.1f}mm (n={int(m.sum())})") print(f" nhom offset >= 15mm: median |du| = {med(resid1[big]):.1f} mm") if OUT_FIT.exists(): print(f"BO QUA ghi {OUT_FIT}: da ton tai") else: with OUT_FIT.open("w", newline="", encoding="utf-8") as f: wr = csv.writer(f) wr.writerow(["muc", "gia_tri", "ghi_chu"]) wr.writerow(["camera_cx_m", round(float(cam[0]), 4), f"decompose median {len(dec_can)} homography pilot," f" he chuan y>L, +-{dec_spread[0]:.3f}"]) wr.writerow(["camera_cy_m", round(float(cam[1]), 4), f"+-{dec_spread[1]:.3f}"]) wr.writerow(["camera_h_m", round(float(cam[2]), 4), f"do cao tren vai, +-{dec_spread[2]:.3f}"]) wr.writerow(["camera_f_px", round(f_med, 1), "tu rang buoc truc giao r1/r2, pp giua anh"]) wr.writerow(["n_events", len(all_events), f"pilot {n_pilot} + bg28 + bg29b"]) wr.writerow(["n_events_clean", len(clean), ""]) wr.writerow(["k_z0", round(k0, 5), "R/(h-R), mo hinh mat vai"]) wr.writerow(["med_resid_z0_big_mm", round(med(resid0[big]), 1), f"nhom offset>=15mm n={int(big.sum())}"]) wr.writerow(["med_offset_big_mm", round(med(offs_m[big] * 1000), 1), ""]) wr.writerow(["med_resid_z0_all_mm", round(med(resid0), 1), ""]) for s in sessions: m = np.array([e["session"] == s for e in clean]) wr.writerow([f"zc_{s}_mm", round(zc_by_s[s][1], 1), f"k={zc_by_s[s][0]:+.5f}, " f"med|du|={med(resid1[m]):.1f}mm, " f"n={int(m.sum())}"]) wr.writerow(["med_resid_zs_big_mm", round(med(resid1[big]), 1), ""]) wr.writerow(["commit_luc_chay", commit, ""]) wr.writerow([]) wr.writerow(["session", "shot", "rail_canon", "proj_m", "apex_mm", "offset_mm", "pred_z0_mm", "resid_z0_mm", "pred_zs_mm", "resid_zs_mm"]) for i, e in enumerate(clean): _k = zc_by_s[e["session"]][0] wr.writerow([e["session"], e["shot"], e["rail_canon"], round(float(proj[i]), 3), e["apex_mm"], e["offset_mm"], round(float(pred0[i]) * 1000, 1), round(float(resid0[i]), 1), round(_k * float(proj[i]) * 1000, 1), round(float(resid1[i]), 1)]) print(f"da ghi {OUT_FIT}") # ---- plot offset field (đọc bằng mắt cho Cowork) try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt fig, ax = plt.subplots(figsize=(5.5, 8)) ax.add_patch(plt.Rectangle((0, 0), W, L, fill=False, lw=1.2)) mk = {"pilot": "o", "bg28": "s", "bg29b": "D"} offsets = offs_m * 1000 for s in sessions: m = np.array([e["session"] == s for e in clean]) ax.scatter(ev_xy[m, 0], ev_xy[m, 1], c=offsets[m], cmap="coolwarm", vmin=-10, vmax=40, s=70, marker=mk.get(s, "o"), edgecolors="k", lw=0.4, zorder=3, label=s) for (x, y), off in zip(ev_xy, offsets): ax.annotate(f"{off:+.0f}", (x, y), textcoords="offset points", xytext=(6, 4), fontsize=7) flagged = [e for e in all_events if e not in clean] if flagged: ax.scatter([e["x_canon_m"] for e in flagged], [e["y_canon_m"] for e in flagged], marker="x", c="gray", s=40, zorder=2, label="loai (co)") ax.plot(cam[0], L + 0.30, "k^", ms=11, zorder=4) ax.annotate(f"camera: C=({cam[0]:+.2f},{cam[1]:+.2f}), " f"h={cam[2]:.2f}m", (cam[0], L + 0.30), textcoords="offset points", xytext=(-60, 12), fontsize=7) ax.legend(fontsize=7, loc="lower left") ax.set_title("BG30 offset field mm (apex - 1R), he chuan — camera " "phia TREN hinh\nmau = offset; +30mm o bang ngan gan " "camera (yL), ~0..+7 bang dai", fontsize=8) ax.set_aspect("equal") ax.set_xlim(-0.25, W + 0.25) ax.set_ylim(-0.25, L + 0.55) fig.tight_layout() if not OUT_PNG.exists(): fig.savefig(OUT_PNG, dpi=130) print(f"da ghi {OUT_PNG}") plt.close(fig) except Exception as e: # noqa: BLE001 — plot là phụ, không giết số đo print(f"plot bo qua ({type(e).__name__}: {e})") return 0 if __name__ == "__main__": sys.exit(main())