Spaces:
Sleeping
Sleeping
| # -*- coding: utf-8 -*- | |
| """Unit nhiễu P0 của gen_synth_shots (BG26 bước 2, gate G-26.1b) — thuần | |
| numpy, KHÔNG pooltool (phần thuần của module import lọt dưới stub conftest; | |
| tầng sim gen_one/pooltool nghiệm thu bằng sanity + verify-repro lúc sinh). | |
| Ba đặc trưng nhiễu là HỢP ĐỒNG (BRIEF bối cảnh 3): test khoá từng đặc trưng | |
| xuất hiện ĐÚNG trong output làm bẩn, không chỉ "có gọi hàm": | |
| 1. ellipse dị hướng XOAY theo hướng vận tốc; | |
| 2. chu kỳ frame trùng 1/6 kiểu 25→30 upconvert (chép nguyên vị trí, | |
| img_diff nhỏ, PTS vẫn đều); | |
| 3. dropout đầu cú đúng cửa sổ [strike, strike + frac·T) + gap ≤ 0.3s. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import pytest | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "scripts" / "broadcast")) | |
| import gen_synth_shots as gs # noqa: E402 | |
| def _straight_sim(heading_deg: float, speed: float = 2.0, dur: float = 2.0): | |
| """Quỹ đạo lý tưởng 1 bi chạy thẳng đều (T, 1, 2) @100Hz từ strike.""" | |
| t = np.arange(0.0, dur, 0.01) | |
| d = np.stack([speed * t * np.cos(np.radians(heading_deg)), | |
| speed * t * np.sin(np.radians(heading_deg))], axis=1) | |
| xy = (np.array([0.45, 0.3]) + d)[:, None, :] | |
| return t, xy | |
| def _corrupt(rng, heading, *, fps=50, upconvert=False, still=0.5, | |
| dropout=0.0, speed=2.0, dur=2.0): | |
| t_sim, xy_sim = _straight_sim(heading, speed, dur) | |
| return t_sim, xy_sim, gs.corrupt_shot( | |
| rng, t_sim, xy_sim, np.array([np.inf]), fps, upconvert, still, | |
| dropout) | |
| def _residual(t_sim, xy_sim, out, still): | |
| """Nhiễu đo được = vị trí bẩn − lý tưởng tại đúng mốc frame.""" | |
| ts = np.clip(out["t"] - still, 0.0, float(t_sim[-1])) | |
| ideal = np.stack([np.interp(ts, t_sim, xy_sim[:, 0, ax]) | |
| for ax in (0, 1)], axis=1) | |
| ideal[out["t"] < still] = xy_sim[0, 0] | |
| return out["xy"][:, 0, :] - ideal | |
| def test_ellipse_di_huong_xoay_theo_huong_van_toc(monkeypatch): | |
| """Đặc trưng 1: chạy dọc +x → nhiễu x ≫ nhiễu y; chạy dọc +y thì NGƯỢC | |
| LẠI — ellipse phải xoay theo vận tốc, không phải trục bàn. Cỡ nhiễu đúng | |
| số P0: dọc ~13mm @2 m/s, ngang ~0.8mm, đứng yên ~0.3mm.""" | |
| monkeypatch.setattr(gs, "GAP_RATE_PER_BALL", 0.0) | |
| for heading, i_along, i_perp in [(0.0, 0, 1), (90.0, 1, 0)]: | |
| t_sim, xy_sim, out = _corrupt(np.random.default_rng(7), heading) | |
| res = _residual(t_sim, xy_sim, out, 0.5) | |
| moving = out["t"] >= 0.5 + 2 * 0.02 | |
| s_along = float(res[moving, i_along].std()) | |
| s_perp = float(res[moving, i_perp].std()) | |
| assert 0.008 < s_along < 0.020, (heading, s_along) | |
| assert s_perp < 0.003, (heading, s_perp) | |
| assert s_along > 4 * s_perp | |
| still_part = res[out["t"] < 0.45] | |
| assert float(np.abs(still_part).std()) < 0.0012 | |
| def test_chu_ky_frame_trung_1_6_kieu_upconvert(monkeypatch): | |
| """Đặc trưng 2: fps=30 upconvert → đúng chu kỳ 1 frame trùng mỗi 6; | |
| frame trùng CHÉP NGUYÊN vị trí (cùng một lần đo) + img_diff nhỏ hơn hẳn | |
| lân cận; PTS vẫn đều 1/30 (bẫy thật: chỉ img_diff lộ).""" | |
| monkeypatch.setattr(gs, "GAP_RATE_PER_BALL", 0.0) | |
| t_sim, xy_sim, out = _corrupt(np.random.default_rng(3), 80.0, fps=30, | |
| upconvert=True, dur=3.0) | |
| dup = out["dup_mask"] | |
| n = len(dup) | |
| assert dup.sum() == pytest.approx(n / 6, abs=2) # chu kỳ 1/6 | |
| idx = np.flatnonzero(dup) | |
| assert np.all(np.diff(idx) == 6) # đều đặn mỗi 6 | |
| for i in idx: | |
| assert np.array_equal(out["xy"][i], out["xy"][i - 1]) | |
| # PTS đều — không được lộ dup qua timestamp | |
| assert np.allclose(np.diff(out["t"]), 1 / 30, atol=1e-6) | |
| # img_diff: frame trùng < 0.3 × median lân cận đang chạy (luật dedup | |
| # broadcast.find_dup_frames phải bắt được) | |
| moving = out["t"] > 0.6 | |
| med = float(np.median(out["img_diff"][moving & ~dup])) | |
| assert med > 0.08 | |
| assert out["img_diff"][idx].max() < 0.3 * med | |
| # và find_dup_frames THẬT nuốt trọn các frame này | |
| from poolcoach_cv.broadcast import find_dup_frames | |
| names = [f"{i:05d}" for i in range(n)] | |
| flagged = find_dup_frames(names, [float(v) for v in out["img_diff"]]) | |
| assert {names[i] for i in idx if out["t"][i] > 0.6} <= flagged | |
| def test_khong_upconvert_khong_frame_trung(monkeypatch): | |
| monkeypatch.setattr(gs, "GAP_RATE_PER_BALL", 0.0) | |
| for fps in (25, 50, 60): | |
| _t, _xy, out = _corrupt(np.random.default_rng(5), 45.0, fps=fps) | |
| assert out["dup_mask"].sum() == 0 | |
| def test_dropout_dau_cu_dung_cua_so(monkeypatch): | |
| """Đặc trưng 3a: dropout 20% × T=2s → cue mất track ĐÚNG [strike, | |
| strike+0.4s); ngoài cửa sổ (không gap) vẫn thấy.""" | |
| monkeypatch.setattr(gs, "GAP_RATE_PER_BALL", 0.0) | |
| still = 1.0 | |
| t_sim, xy_sim, out = _corrupt(np.random.default_rng(11), 60.0, | |
| still=still, dropout=0.2) | |
| t = out["t"] | |
| hide = (t >= still) & (t < still + 0.2 * float(t_sim[-1])) | |
| assert hide.sum() >= 5 | |
| assert not out["covered"][hide, 0].any() | |
| assert out["covered"][~hide, 0].all() | |
| def test_gap_ngan_toi_da_03s(): | |
| """Đặc trưng 3b: gap rải ngẫu nhiên — mỗi run mất track dài ≤ 0.3s | |
| (+1 frame biên).""" | |
| rng = np.random.default_rng(2) # seed này chắc chắn có gap (λ=3) | |
| import unittest.mock as mock | |
| with mock.patch.object(gs, "GAP_RATE_PER_BALL", 3.0): | |
| t_sim, xy_sim, out = _corrupt(rng, 30.0, dropout=0.0) | |
| cov = out["covered"][:, 0] | |
| assert (~cov).sum() > 0 | |
| runs, cur = [], 0 | |
| for c in cov: | |
| cur = cur + 1 if not c else 0 | |
| if cur: | |
| runs.append(cur) | |
| fps = 50 | |
| assert max(runs) <= int(0.30 * fps) + 1 | |
| def test_shard_roundtrip_bit_giong(tmp_path, monkeypatch): | |
| """Format npz: write_shard → iter_shots trả lại đúng bit từng cú (loader | |
| này là đường đọc CHUNG của eval BG26 + train BG27).""" | |
| monkeypatch.setattr(gs, "GAP_RATE_PER_BALL", 1.0) | |
| metas, arrays = [], [] | |
| for i, (hd, fps, up) in enumerate([(0.0, 30, True), (90.0, 60, False), | |
| (200.0, 25, False)]): | |
| rng = np.random.default_rng(100 + i) | |
| t_sim, xy_sim, out = _corrupt(rng, hd, fps=fps, upconvert=up, | |
| dropout=0.1) | |
| meta = {"label_v0": 2.0 + i, "label_phi": hd, "label_a": 0.1 * i, | |
| "label_b": -0.1 * i, "v0_ball": 2.5, "phi_ball": hd, | |
| "identifiable": i % 2, "scratch": 0, "potted_any": 1, | |
| "n_bb": i, "n_cush": 2, "t_first_bb": 0.5, | |
| "t_first_cush": float("nan"), "fps": fps, | |
| "upconvert": int(up), "still_s": 0.5, "dropout_frac": 0.1, | |
| "n_frames": len(out["t"]), "n_balls": 1, "shot_idx": i} | |
| metas.append(meta) | |
| arrays.append({"xy": out["xy"], "covered": out["covered"], | |
| "img_diff": out["img_diff"], | |
| "ball_ids": np.array([0], dtype=np.uint8)}) | |
| path = tmp_path / "train_0000000.npz" | |
| gs.write_shard(path, metas, arrays) | |
| shots = list(gs.iter_shots(path)) | |
| assert len(shots) == 3 | |
| for m, a, s in zip(metas, arrays, shots): | |
| assert np.array_equal(s["xy"], a["xy"]) | |
| assert np.array_equal(s["covered"], a["covered"]) | |
| assert np.array_equal(s["img_diff"], a["img_diff"]) | |
| assert s["fps"] == m["fps"] and s["n_frames"] == m["n_frames"] | |
| assert s["label_v0"] == np.float32(m["label_v0"]) | |
| assert len(s["t"]) == m["n_frames"] | |
| assert s["t"][1] - s["t"][0] == pytest.approx(1 / m["fps"]) | |