"""Đo phân rã "hết đường" của ZonePlanner V2 theo TỪNG tiêu chí (04/08/2026). Việc treo từ 31/07: số "48% bàn random không có cú đạt 5 tiêu chí" chưa trích được làm kết luận vì chưa đo tách nguyên nhân. Script này là ĐO LƯỜNG THUẦN — không chỉnh engine, không kết luận: chạy V2 NGUYÊN TRẠNG (đúng chuỗi ``validate_full → target_ball → feasible_pockets_full → plan_shot_v2`` của ``recommend_v2``) với ``debug_candidates=True``, rồi đếm offline trên mask. Hai bộ thế, seed cố định, TÁCH BẠCH (không dùng seed demo — đó là curation): --source random : ``--boards`` bàn 10 bi (cue + 1..9) sinh bằng ĐÚNG ``_gen_table`` của ``eval_fullrack.py`` (import, không copy — một code path). Bàn k dùng ``default_rng([seed_base, 0, k])`` → từng bàn tự tái lập. --source break : ``--boards`` thế SAU BREAK sinh bằng sim: rack 9 bi chuẩn ``pt.get_rack`` (bỏ cue mặc định của rack), cue jitter trong kitchen (x ∈ [0.25, 0.75]·w, y ∈ [0.10, 0.24]·l — head string ở l/4), phi nhắm apex bi 1 ± 1.5°, v0 break CỐ ĐỊNH 7.0 m/s (break không phải cú của engine — lực ngoài grid, khai tường minh), a = b = 0. Bàn k dùng ``default_rng([seed_base, 1, k])``. LOẠI (báo tỉ lệ): scratch (ball-in-hand ngoài scope) và golden break (bi 9 rớt). GIỮ thế có bi khác rớt. Thế break không qua nổi ``validate_full`` là THẾ KỲ DỊ: đếm riêng, > ``ANOMALY_ABORT`` là dừng cứng — đừng lọc âm thầm (BRIEF 04/08 "Nếu bí"). Mỗi thế: cú ĐẦU TIÊN, target = bi nhỏ nhất (V2 tự suy). Đếm trên mask ĐỦ 5 tiêu chí (không short-circuit — ``CandidateDebugV2``), nên khác ``fail_counts`` của engine (bản đó short-circuit (ii)/(iv)). Mask ``None`` ("không đo được": cue vào lỗ / sim chết) đếm riêng ``und_*``, KHÔNG tính vào trượt. Output (rơi ``ROOT.parent`` — nếp ``DEFAULT_OUT_DIR`` cũ, ngoài git): failcounts_.csv : per-thế (seed, mask counts, saved/only per tiêu chí, thời gian) failcounts__out.txt : bảng tổng (a) % hết đường (b) đếm trượt theo tiêu chí (c) one-left-out per criterion = bỏ RIÊNG tiêu chí X thì bao nhiêu thế hết-đường có cú (tính lại từ mask, KHÔNG sửa engine) + kiểm G3. G3 (tự nhất quán, in ĐẠT/FAIL + exit code): số thế đúng đủ; có_cú + hết_đường = tổng; thế có cú ⇔ ∃ ứng viên pass đủ 5 (mask trùng engine từng thế); one-left-out(X) ≥ số thế hết-đường có ứng viên chỉ trượt đúng mình X. Worker qua env ``POOLCOACH_WORKERS`` (quy ước 27/07; 0/1 = serial, mỗi worker tự trả warmup JIT ~2 s khi cache Numba ấm, ~40 s nguội). Chạy: python scripts/eval_failcounts.py --source random python scripts/eval_failcounts.py --source break """ from __future__ import annotations import argparse import csv import math import multiprocessing import os import statistics import sys import time from pathlib import Path import numpy as np sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") ROOT = Path(__file__).resolve().parents[1] for _p in (ROOT / "src", ROOT / "scripts"): if str(_p) not in sys.path: sys.path.insert(0, str(_p)) DEFAULT_OUT_DIR = ROOT.parent # nếp cũ: file đo rơi ở thư mục cha, ngoài git SEED_BASE_DEFAULT = 20260804 # seed base RIÊNG của lần đo này (BRIEF 04/08) # ------------------------- tham số break (tự quyết trong BRIEF, khai ở đây) V0_BREAK = 7.0 # m/s — cố định, NGOÀI grid lực engine BREAK_CUE_X = (0.25, 0.75) # × w — jitter vị trí cue trong kitchen BREAK_CUE_Y = (0.10, 0.24) # × l — kitchen là y ≤ l/4 (rack ở y = 3l/4) BREAK_PHI_JITTER_DEG = 1.5 # ± độ quanh đường ngắm apex (bi 1) ANOMALY_ABORT = 10 # >10/200 thế break kỳ dị (5%) → dừng cứng CRITERIA = ("i", "ii", "iii", "iv", "v") CSV_FIELDS = (["source", "seed_base", "idx", "excluded", "phi", "cue_x", "cue_y", "potted_break", "target", "n_balls", "n_candidates", "n_pockets_tried", "n_sim_none", "n_pass", "het_duong"] + [f"fail_{k}" for k in CRITERIA] + ["und_ii", "und_iv"] + [f"saved_{k}" for k in CRITERIA] + [f"only_{k}" for k in CRITERIA] + ["elapsed_s"]) # ------------------------------------------------------------ sinh thế bàn def gen_random_board(seed_base, idx, w, l): """Bàn 10 bi bằng ĐÚNG generator của eval_fullrack (import, không copy).""" from eval_fullrack import _gen_table from poolcoach_rl.envs.position_env import BALL_R rng = np.random.default_rng([seed_base, 0, idx]) return _gen_table(rng, w, l, BALL_R, full=True) def gen_break_board(env_h, rack_xy, seed_base, idx): """Sim MỘT cú break → (balls sau break | None nếu loại, meta có lý do).""" from poolcoach_rl.recommend.simulate import simulate_shot_multi rng = np.random.default_rng([seed_base, 1, idx]) cue = np.array([env_h.w * rng.uniform(*BREAK_CUE_X), env_h.l * rng.uniform(*BREAK_CUE_Y)]) apex = rack_xy["1"] d = apex - cue phi = (math.degrees(math.atan2(d[1], d[0])) + rng.uniform(-BREAK_PHI_JITTER_DEG, BREAK_PHI_JITTER_DEG)) % 360.0 meta = {"phi": round(phi, 3), "cue_x": round(float(cue[0]), 4), "cue_y": round(float(cue[1]), 4)} balls = {bid: xy.copy() for bid, xy in rack_xy.items() if bid != "cue"} balls["cue"] = cue m = simulate_shot_multi(env_h, balls, phi, V0_BREAK, 0.0, 0.0) if m is None: return None, {**meta, "excluded": "sim_none"} # kỳ dị — đếm riêng meta["potted_break"] = "+".join(m["potted"]) if m["scratch"]: return None, {**meta, "excluded": "scratch"} if "9" in m["potted"]: return None, {**meta, "excluded": "golden"} post = {bid: np.asarray(xy, dtype=np.float64) for bid, xy in m["balls_final"].items() if xy is not None} return post, {**meta, "excluded": ""} # ------------------------------------------------------------------ đo 1 thế def measure_board(env_h, balls): """Chạy V2 nguyên trạng (đúng chuỗi recommend_v2) + đếm offline từ mask.""" from poolcoach_rl.recommend.core import validate_full from poolcoach_rl.recommend.geometry import feasible_pockets_full from poolcoach_rl.recommend.rules import target_ball from poolcoach_rl.recommend.zone_v2 import plan_shot_v2 validate_full(env_h, balls) # thế kỳ dị nổ ValueError → caller bắt target = target_ball(balls) pockets = feasible_pockets_full(env_h, balls, target) t0 = time.perf_counter() plan = plan_shot_v2(env_h, balls, target, pockets=pockets, debug_candidates=True) dt = time.perf_counter() - t0 ents = plan.debug_candidates dry = plan.shot is None rec = { "target": target, "n_balls": len(balls) - 1, "n_candidates": len(ents), "n_pockets_tried": len(plan.pockets), "n_sim_none": sum(1 for e in ents if e.sim_none), "n_pass": sum(1 for e in ents if e.pass_all), "het_duong": int(dry), "elapsed_s": round(dt, 3), } for k in CRITERIA: rec[f"fail_{k}"] = sum(1 for e in ents if e.mask[k] is False) for k in ("ii", "iv"): # "không đo được" ≠ trượt — đếm riêng rec[f"und_{k}"] = sum(1 for e in ents if not e.sim_none and e.mask[k] is None) for k in CRITERIA: others = [o for o in CRITERIA if o != k] saved = any(all(e.mask[o] is True for o in others) for e in ents) only = any(e.mask[k] is False and all(e.mask[o] is True for o in others) for e in ents) rec[f"saved_{k}"] = int(dry and saved) rec[f"only_{k}"] = int(dry and only) # G3 per-thế: mask và engine phải là MỘT code path — lệch là dừng cứng, # đừng để số hai nguồn lẫn nhau (bẫy 27/07) engine_pass = sum(r.n_pass for r in plan.pockets) if rec["n_pass"] != engine_pass: raise AssertionError( f"mask n_pass={rec['n_pass']} != engine n_pass={engine_pass}") if dry != (rec["n_pass"] == 0): raise AssertionError( f"het_duong={dry} nhưng n_pass={rec['n_pass']} — mask lệch engine") return rec # ------------------------------------------------------- worker (pool/serial) _ENV = None _RACK_XY = None def _init_worker(): """Mỗi process: dựng env + trả warmup JIT một lần, tính rack một lần.""" global _ENV, _RACK_XY import pooltool as pt from poolcoach_rl.envs import PositionPlayEnv from poolcoach_rl.recommend.core import warmup _ENV = PositionPlayEnv() t = warmup(_ENV) rack = pt.get_rack(pt.GameType.NINEBALL, _ENV.table) _RACK_XY = {bid: np.asarray(b.state.rvw[0][:2], dtype=np.float64) for bid, b in rack.items()} print(f" [worker {os.getpid()}] warmup {t:.1f}s", flush=True) def _measure_task(task): source, seed_base, idx = task row = {"source": source, "seed_base": seed_base, "idx": idx} if source == "random": balls = gen_random_board(seed_base, idx, _ENV.w, _ENV.l) meta = {"excluded": ""} else: balls, meta = gen_break_board(_ENV, _RACK_XY, seed_base, idx) row.update(meta) if balls is None: return row try: row.update(measure_board(_ENV, balls)) except ValueError as e: # validate_full từ chối = thế kỳ dị (bi chồng/ngoài bàn sau sim) row["excluded"] = f"validate: {e}" return row # ------------------------------------------------------------------ tổng hợp def _pct(a, b): return f"{100.0 * a / b:.1f}%" if b else "n/a" def summarize(source, rows, seed_base, workers, total_s): kept = [r for r in rows if not r["excluded"]] dry = [r for r in kept if r["het_duong"]] n_judged = sum(r["n_candidates"] - r["n_sim_none"] for r in kept) L = [] L.append(f"=== eval_failcounts — source={source} ===") L.append(f" seed_base {seed_base} | boards {len(rows)} | workers " f"{workers} | tổng {total_s:.0f}s") if source == "break": L.append(f" break: v0={V0_BREAK} m/s (ngoài grid, cố định) | cue " f"x∈{BREAK_CUE_X}·w y∈{BREAK_CUE_Y}·l | phi apex " f"±{BREAK_PHI_JITTER_DEG}°") reasons = {} for r in rows: if r["excluded"]: key = r["excluded"].split(":")[0] reasons[key] = reasons.get(key, 0) + 1 L.append(f" loại {len(rows) - len(kept)}/{len(rows)} " f"({_pct(len(rows) - len(kept), len(rows))}): " + (", ".join(f"{k}={v}" for k, v in sorted(reasons.items())) or "không")) n_pot_break = sum(1 for r in kept if r.get("potted_break")) L.append(f" thế GIỮ có bi rớt lúc break: {n_pot_break}/{len(kept)}") L.append("") L.append(f"(a) hết đường: {len(dry)}/{len(kept)} thế " f"= {_pct(len(dry), len(kept))} (có cú: {len(kept) - len(dry)})") L.append("") L.append(f"(b) đếm trượt theo tiêu chí trên TOÀN ứng viên đã chấm " f"(mask đủ, không short-circuit; n = {n_judged}):") for k in CRITERIA: f = sum(r[f"fail_{k}"] for r in kept) L.append(f" ({k:>3s}) trượt {f:6d} ({_pct(f, n_judged)})") und_ii = sum(r["und_ii"] for r in kept) und_iv = sum(r["und_iv"] for r in kept) n_sim_none = sum(r["n_sim_none"] for r in kept) L.append(f" không đo được: (ii) {und_ii} · (iv) {und_iv} " f"(cue vào lỗ) · sim_none {n_sim_none}") n_pass = sum(r["n_pass"] for r in kept) L.append(f" pass đủ 5: {n_pass} ({_pct(n_pass, n_judged)})") L.append("") L.append(f"(c) one-left-out per criterion — bỏ RIÊNG tiêu chí X thì bao " f"nhiêu thế hết-đường được cứu (trên {len(dry)} thế hết đường):") for k in CRITERIA: s = sum(r[f"saved_{k}"] for r in dry) L.append(f" bỏ ({k:>3s}): cứu {s:3d} ({_pct(s, len(dry))})") times = [r["elapsed_s"] for r in kept] if times: L.append("") L.append(f" search/thế: median {statistics.median(times):.2f}s | " f"max {max(times):.2f}s") # ------------------------------------------------------------ kiểm G3 L.append("") L.append("=== G3 — sanity ===") checks = [] checks.append((f"số thế sinh = {len(rows)}", len(rows) >= 1)) if source == "random": checks.append(("random: không thế nào bị loại", len(kept) == len(rows))) else: n_anom = sum(1 for r in rows if r["excluded"].startswith(("validate", "sim_none"))) checks.append((f"break: kỳ dị {n_anom} ≤ {ANOMALY_ABORT} " f"(quá là nghi break sim sai)", n_anom <= ANOMALY_ABORT)) checks.append((f"có_cú + hết_đường = tổng ({len(kept) - len(dry)} + " f"{len(dry)} = {len(kept)})", True)) # theo cách dựng checks.append(("mọi thế có cú: ∃ ứng viên pass đủ 5", all(r["n_pass"] >= 1 for r in kept if not r["het_duong"]))) checks.append(("mọi thế hết đường: 0 ứng viên pass đủ 5", all(r["n_pass"] == 0 for r in dry))) olo_ok = all(sum(r[f"saved_{k}"] for r in dry) >= sum(r[f"only_{k}"] for r in dry) for k in CRITERIA) checks.append(("one-left-out(X) ≥ số thế chỉ-trượt-đúng-X", olo_ok)) ok_all = True for name, ok in checks: ok_all &= bool(ok) L.append(f" [{'ĐẠT' if ok else 'FAIL'}] {name}") return "\n".join(L), ok_all # ---------------------------------------------------------------------- main def main(): ap = argparse.ArgumentParser() ap.add_argument("--source", choices=["random", "break"], required=True) ap.add_argument("--boards", type=int, default=200) ap.add_argument("--seed-base", type=int, default=SEED_BASE_DEFAULT) ap.add_argument("--out-dir", type=Path, default=DEFAULT_OUT_DIR) args = ap.parse_args() workers = int(os.environ.get("POOLCOACH_WORKERS", "0") or "0") tasks = [(args.source, args.seed_base, k) for k in range(args.boards)] print(f"eval_failcounts: source={args.source} boards={args.boards} " f"seed_base={args.seed_base} workers={workers or 'serial'}", flush=True) t0 = time.time() rows = [] if workers >= 2: with multiprocessing.Pool(workers, initializer=_init_worker) as pool: for i, row in enumerate(pool.imap_unordered(_measure_task, tasks), start=1): rows.append(row) if i % 10 == 0 or i == len(tasks): print(f" [{i}/{len(tasks)}] {time.time() - t0:.0f}s", flush=True) else: _init_worker() for i, t in enumerate(tasks, start=1): rows.append(_measure_task(t)) if i % 10 == 0 or i == len(tasks): print(f" [{i}/{len(tasks)}] {time.time() - t0:.0f}s", flush=True) total_s = time.time() - t0 rows.sort(key=lambda r: r["idx"]) args.out_dir.mkdir(parents=True, exist_ok=True) csv_path = args.out_dir / f"failcounts_{args.source}.csv" with csv_path.open("w", newline="", encoding="utf-8") as f: wr = csv.DictWriter(f, fieldnames=CSV_FIELDS, restval="") wr.writeheader() wr.writerows(rows) summary, ok = summarize(args.source, rows, args.seed_base, workers, total_s) print() print(summary) txt_path = args.out_dir / f"failcounts_{args.source}_out.txt" txt_path.write_text(summary + "\n", encoding="utf-8") print(f"\nCSV: {csv_path}\nTXT: {txt_path}") sys.exit(0 if ok else 1) if __name__ == "__main__": main()