"""Unit test shot segmentation (lát A1, BRIEF 13/08/2026) — track TỔNG HỢP. Kiểm LÕI cắt ranh giới (``segment_shots`` — thuần numpy, chạy venv app, không cv2/torch): cú đơn, đa cú, "mọi bi đứng yên" chứ không riêng cue, coast/mất det giữa cú, mất cảnh, video tĩnh không đẻ cú ma, đệm biên. KHÔNG kiểm YOLO/decode (``segment_video`` nghiệm thu trên VOD thật ở gate A1.2) — cùng phân vai với test_broadcast_analytics vs gate G-24.3. Ngưỡng vật lý trong kịch bản lấy từ bộ chốt 11/08 (import broadcast): tốc độ cú > V_INIT 0.25 m/s, jitter bi tĩnh ~2mm/frame (p99 apparent speed 55 mm/s ở 30fps — đúng cỡ nhiễu đo được P0). """ from __future__ import annotations import numpy as np import pytest from poolcoach_cv.broadcast import V_INIT_MPS, V_STILL_MPS from poolcoach_cv.segmentation import (BLIND_MAX_S, PAD_POST_S, PAD_PRE_S, STILL_WIN_S, segment_shots) FPS = 30.0 def _pos_at(t: float, p0, segs): """Vị trí theo lịch chuyển động: ``segs`` = [(t_a, t_b, p_a, p_b)] đoạn thẳng đều; ngoài đoạn thì đứng yên ở mút gần nhất.""" x, y = p0 for t_a, t_b, p_a, p_b in segs: if t < t_a: break if t >= t_b: x, y = p_b else: f = (t - t_a) / (t_b - t_a) x = p_a[0] + (p_b[0] - p_a[0]) * f y = p_a[1] + (p_b[1] - p_a[1]) * f return x, y def build(dur_s, cue_p0, cue_segs=(), balls=(), cue_missing=(), blind=(), jitter=0.0, seed=7): """Track tổng hợp 30fps → (rows, others) đúng shape segment_shots. ``balls``: tuple (x, y) bi tĩnh, hoặc dict {"p0": (x,y), "segs": [...]} bi có lịch chạy riêng. ``cue_missing``/``blind``: list (t_a, t_b) — khoảng cue mất det / khoảng KHÔNG thấy bi nào (mất cảnh). """ rng = np.random.default_rng(seed) rows, others = [], [] n = int(round(dur_s * FPS)) + 1 for i in range(n): t = i / FPS name = f"{i:05d}" is_blind = any(a <= t < b for a, b in blind) cue_gone = is_blind or any(a <= t < b for a, b in cue_missing) cx, cy = _pos_at(t, cue_p0, cue_segs) if jitter: cx += rng.normal(0, jitter) cy += rng.normal(0, jitter) rows.append({ "frame_file": name, "t_s": round(t, 6), "covered": 0 if cue_gone else 1, "table_x_m": "" if cue_gone else cx, "table_y_m": "" if cue_gone else cy, # img_diff hằng 1.0 (bàn "đang sống") — không kích luật frame # trùng của find_dup_frames, thứ đã có test riêng ở broadcast "img_diff": 1.0 if i else -1.0, }) if is_blind: continue for spec in balls: if isinstance(spec, dict): bx, by = _pos_at(t, spec["p0"], spec["segs"]) else: bx, by = spec if jitter: bx += rng.normal(0, jitter) by += rng.normal(0, jitter) others.append({"frame_file": name, "t_s": round(t, 6), "x_m": bx, "y_m": by}) return rows, others BALLS_TINH = [(0.4, 2.0), (0.9, 1.8), (0.6, 0.4)] # ------------------------------------------------------------------ cú đơn def test_cu_don_mot_cu_ok_moc_dung(): rows, others = build( 6.0, cue_p0=(0.3, 0.5), cue_segs=[(1.0, 2.0, (0.3, 0.5), (0.9, 1.7))], # ~1.34 m/s balls=BALLS_TINH) out = segment_shots(rows, others) assert [s["status"] for s in out["shots"]] == ["ok"] s = out["shots"][0] assert s["idx"] == 1 # onset đúng luật V_INIT_RUN bước > V_INIT — bắt tại lúc bắt đầu chạy assert s["t_onset_s"] == pytest.approx(1.0, abs=0.15) # settle trễ tối đa ~1 cửa sổ chord sau lúc dừng thật (2.0s) assert 2.0 <= s["t_settle_s"] <= 2.0 + 2 * STILL_WIN_S # đệm hai đầu, trong biên video assert s["t_start_s"] == pytest.approx(s["t_onset_s"] - PAD_PRE_S, abs=0.05) assert s["t_settle_s"] < s["t_end_s"] <= s["t_settle_s"] + PAD_POST_S assert 0.0 <= s["t_start_s"] < s["t_onset_s"] def test_video_tinh_khong_de_cu_ma(): """Bàn tĩnh 10s + jitter detect 2mm/frame (~p99 apparent speed 55mm/s của P0) — không được đẻ cú nào.""" rows, others = build(10.0, cue_p0=(0.5, 1.0), balls=BALLS_TINH, jitter=0.002) out = segment_shots(rows, others) assert out["shots"] == [] def test_track_rong_nem_valueerror(): with pytest.raises(ValueError): segment_shots([]) # ------------------------------------------------------------------- đa cú def test_da_cu_ba_cu_theo_thu_tu(): segs = [(1.0, 2.0, (0.3, 0.5), (0.9, 1.7)), (5.0, 6.2, (0.9, 1.7), (0.4, 0.8)), (10.0, 10.8, (0.4, 0.8), (1.0, 2.2))] rows, others = build(14.0, cue_p0=(0.3, 0.5), cue_segs=segs, balls=BALLS_TINH) out = segment_shots(rows, others) shots = out["shots"] assert [s["status"] for s in shots] == ["ok", "ok", "ok"] assert [s["idx"] for s in shots] == [1, 2, 3] for s, (t_a, _b, _pa, _pb) in zip(shots, segs): assert s["t_onset_s"] == pytest.approx(t_a, abs=0.15) # mốc thô tăng dần, không chồng nhau for a, b in zip(shots, shots[1:]): assert a["t_settle_s"] <= b["t_onset_s"] # đệm không nuốt sang chuyển động cú lân cận assert a["t_end_s"] <= b["t_onset_s"] + 1e-6 assert b["t_start_s"] >= a["t_settle_s"] - 1e-6 def test_hai_cu_sat_nhau_van_tach(): """Cú 2 nổ ngay sau khi bàn vừa yên lại — đóng cú xong phải armed luôn (không đợi thêm một cửa sổ yên nữa) mới bắt được onset kế.""" segs = [(1.0, 2.0, (0.3, 0.5), (0.9, 1.7)), (3.2, 4.0, (0.9, 1.7), (0.5, 0.9))] rows, others = build(7.0, cue_p0=(0.3, 0.5), cue_segs=segs, balls=BALLS_TINH) out = segment_shots(rows, others) assert [s["status"] for s in out["shots"]] == ["ok", "ok"] assert out["shots"][1]["t_onset_s"] == pytest.approx(3.2, abs=0.15) # ------------------------------------------- "MỌI bi đứng yên", không chỉ cue def test_bi_muc_tieu_con_lan_thi_chua_dong_cu(): """Cue dừng ở 2.0 nhưng bi mục tiêu lăn tiếp tới 4.0 — settle phải chờ MỌI bi (định nghĩa ranh giới của design §4), không chốt sớm theo cue.""" ball_lan = {"p0": (0.9, 1.8), "segs": [(2.0, 4.0, (0.9, 1.8), (0.5, 1.2))]} # ~0.36 m/s rows, others = build( 7.0, cue_p0=(0.3, 0.5), cue_segs=[(1.0, 2.0, (0.3, 0.5), (0.9, 1.7))], balls=[ball_lan, (0.4, 2.0)]) out = segment_shots(rows, others) assert [s["status"] for s in out["shots"]] == ["ok"] assert out["shots"][0]["t_settle_s"] >= 4.0 - 0.1 # --------------------------------------------------- coast / mất det giữa cú def test_cue_mat_det_ngan_giua_cu_khong_tach_cu(): rows, others = build( 6.0, cue_p0=(0.3, 0.5), cue_segs=[(1.0, 3.0, (0.3, 0.5), (1.1, 2.3))], balls=BALLS_TINH, cue_missing=[(1.8, 2.1)]) out = segment_shots(rows, others) assert [s["status"] for s in out["shots"]] == ["ok"] assert out["shots"][0]["t_settle_s"] >= 3.0 - 0.1 def test_cue_mat_det_dai_hon_cua_so_van_khong_dong_som(): """Cue biến mất TRỌN một cửa sổ chord khi đang bay (người che — bài học shot_07): mọi bi khác đứng yên nhưng chưa được kết luận 'đã yên' chỉ vì vắng mặt nhân chứng chính.""" rows, others = build( 6.0, cue_p0=(0.3, 0.5), cue_segs=[(1.0, 3.0, (0.3, 0.5), (1.1, 2.3))], balls=BALLS_TINH, cue_missing=[(1.5, 2.5)]) out = segment_shots(rows, others) assert [s["status"] for s in out["shots"]] == ["ok"] s = out["shots"][0] # không được đóng trong khoảng cue tàng hình (1.5–2.5) assert s["t_settle_s"] >= 3.0 - 0.1 # ------------------------------------------------------------ mất cảnh (§5) def test_mat_canh_giua_cu_bao_loi_khong_nuot(): rows, others = build( 9.0, cue_p0=(0.3, 0.5), cue_segs=[(1.0, 2.0, (0.3, 0.5), (0.9, 1.7)), (6.0, 6.8, (0.9, 1.7), (0.5, 1.0))], balls=BALLS_TINH, blind=[(1.5, 4.0)]) out = segment_shots(rows, others) shots = out["shots"] assert len(shots) == 2 assert shots[0]["status"] == "error" assert "mất cảnh" in shots[0]["reason"] # cú sau khi cảnh quay lại + bàn yên lại vẫn bắt bình thường assert shots[1]["status"] == "ok" assert shots[1]["t_onset_s"] == pytest.approx(6.0, abs=0.15) def test_blind_ngan_hon_nguong_khong_pha_cu(): """Đứt detect ngắn (< BLIND_MAX_S) giữa cú — video giật/che thoáng qua — không được chẻ cú thành hai hay báo mất cảnh.""" assert BLIND_MAX_S >= 0.5 rows, others = build( 6.0, cue_p0=(0.3, 0.5), cue_segs=[(1.0, 3.0, (0.3, 0.5), (1.1, 2.3))], balls=BALLS_TINH, blind=[(1.8, 2.1)]) out = segment_shots(rows, others) assert [s["status"] for s in out["shots"]] == ["ok"] def test_video_ket_thuc_giua_cu_bao_loi(): rows, others = build( 4.5, cue_p0=(0.3, 0.5), cue_segs=[(4.0, 6.0, (0.3, 0.5), (0.9, 1.7))], balls=BALLS_TINH) out = segment_shots(rows, others) assert len(out["shots"]) == 1 assert out["shots"][0]["status"] == "error" assert "kết thúc" in out["shots"][0]["reason"] # ------------------------------------------------------------------ biên/đệm def test_dem_bien_khong_vuot_video(): rows, others = build( 3.0, cue_p0=(0.3, 0.5), cue_segs=[(0.8, 1.6, (0.3, 0.5), (0.9, 1.7))], balls=BALLS_TINH) out = segment_shots(rows, others) assert len(out["shots"]) == 1 s = out["shots"][0] assert s["t_start_s"] >= 0.0 assert s["t_end_s"] <= 3.0 + 1e-6 assert s["t_start_s"] < s["t_onset_s"] < s["t_settle_s"] <= s["t_end_s"] def test_nguong_dung_bo_chot_khong_bi_dinh_nghia_lai(): """Khoá hợp đồng "dùng nguyên ngưỡng chốt 11/08": segmentation phải đọc V_STILL/V_INIT từ broadcast, không mang bản sao riêng có thể trôi.""" import poolcoach_cv.broadcast as bc import poolcoach_cv.segmentation as seg assert seg.V_STILL_MPS is bc.V_STILL_MPS assert seg.V_INIT_MPS is bc.V_INIT_MPS assert seg.VMAX_MPS is bc.VMAX_MPS assert V_STILL_MPS == 0.055 and V_INIT_MPS == 0.25