#!/usr/bin/env python3 """Capture web-runner reference frames for battle effect pixel-oracle checks. This does not prove original-game fidelity by itself. It captures the current web runner's preview canvas at deterministic effect-active ticks so that an original runtime capture can later be cropped/aligned and compared by ``tools/compare_battle_effect_pixel_oracle.py``. """ from __future__ import annotations import argparse import base64 import json import shutil import subprocess import sys from pathlib import Path from typing import Any from browser_smoke_helpers import ( WebDriverError, execute_async_js, free_port, request_json, wait_for_driver, ) from verify_battle_skill_timeline_browser import ( wait_for_http, wait_for_page_ready, select_action, ) ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "out" DATA = ROOT / "data" CAPTURES = ROOT / "captures" / "battle_effect_pixel_oracle" PLAN = OUT / "battle_effect_pixel_oracle_plan.json" MANIFEST = DATA / "battle_effect_capture_manifest.json" def read_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def row_id(row: dict[str, Any]) -> str: return str(row.get("id") or f"{row.get('ownerKey','unknown')}:{row.get('skillIdHex','unknown')}") def safe_row_dir(value: str) -> str: return "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in value).strip("._") or "row" def load_plan_rows(scope: str) -> list[dict[str, Any]]: data = read_json(PLAN) if scope == "first-pass": rows = data.get("recommendedFirstPassRows") or [] elif scope == "full": rows = data.get("fullOracleRows") or [] else: raise ValueError(f"unknown scope: {scope}") if not rows: raise RuntimeError(f"{PLAN} has no rows for scope {scope!r}") return rows def tick_samples(marker: dict[str, Any], mode: str, max_frames: int) -> list[int]: diagnostics = marker.get("effectDiagnostics") or {} if mode == "samples": raw = diagnostics.get("activeTickSamples") or [] ticks = sorted({int(value) for value in raw if isinstance(value, (int, float, str)) and str(value).lstrip("-").isdigit()}) elif mode == "range": tick_min = diagnostics.get("tickMin") tick_max = diagnostics.get("tickMax") if tick_min is None or tick_max is None: ticks = [] else: ticks = list(range(int(tick_min), int(tick_max) + 1)) else: raise ValueError(f"unknown tick mode: {mode}") if max_frames > 0: ticks = ticks[:max_frames] return ticks def select_payload(row: dict[str, Any]) -> dict[str, Any]: return { "ownerKey": row.get("ownerKey") or "", "skillIdHex": str(row.get("skillIdHex") or "").lower(), "levelOrFixed": row.get("levelOrFixed"), "name": row.get("skillName") or row.get("name") or "", } def save_preview_png(driver_port: int, session_id: str, tick: int, output: Path) -> dict[str, Any]: state = execute_async_js( driver_port, session_id, f""" const done = arguments[arguments.length - 1]; (async () => {{ try {{ if (typeof window.HWANSE_BATTLE_TIMELINE_RENDER_TICK !== 'function') {{ throw new Error('HWANSE_BATTLE_TIMELINE_RENDER_TICK is not available'); }} const render = await window.HWANSE_BATTLE_TIMELINE_RENDER_TICK({int(tick)}); const canvas = document.querySelector('#previewCanvas'); if (!canvas) throw new Error('previewCanvas not found'); done({{ ok: true, tick: {int(tick)}, width: canvas.width, height: canvas.height, render, dataUrl: canvas.toDataURL('image/png'), }}); }} catch (error) {{ done({{ ok: false, error: String(error && error.stack ? error.stack : error) }}); }} }})(); """, timeout=12, ) if not state.get("ok"): raise WebDriverError(f"could not capture tick {tick}: {state!r}") data_url = str(state.get("dataUrl") or "") prefix = "data:image/png;base64," if not data_url.startswith(prefix): raise WebDriverError(f"preview canvas returned non-PNG data URL for tick {tick}") output.parent.mkdir(parents=True, exist_ok=True) output.write_bytes(base64.b64decode(data_url[len(prefix):], validate=True)) return { "tick": tick, "path": str(output.relative_to(ROOT)), "width": state.get("width"), "height": state.get("height"), "activeEffectCount": (state.get("render") or {}).get("activeEffectCount"), } def update_capture_manifest(entries: list[dict[str, Any]]) -> None: DATA.mkdir(parents=True, exist_ok=True) existing: dict[str, Any] if MANIFEST.exists(): try: existing = read_json(MANIFEST) except json.JSONDecodeError: existing = {} else: existing = {} by_id = {str(entry.get("id") or ""): dict(entry) for entry in existing.get("captures") or []} for entry in entries: previous = by_id.get(entry["id"], {}) merged = { **previous, "id": entry["id"], "ownerKey": entry["ownerKey"], "skillIdHex": entry["skillIdHex"], "levelOrFixed": entry.get("levelOrFixed"), "skillName": entry.get("skillName") or "", "webReferenceCapture": entry["webReferenceCapture"], "originalCapture": previous.get("originalCapture", ""), "captureTicks": entry["captureTicks"], } by_id[entry["id"]] = merged payload = { "version": 1, "kind": "hwanse-battle-effect-capture-manifest", "note": "Local-only manifest. Add originalCapture paths, then run tools/compare_battle_effect_pixel_oracle.py.", "captures": sorted(by_id.values(), key=lambda item: item["id"]), } MANIFEST.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def capture_with_browser(base: str, rows: list[dict[str, Any]], tick_mode: str, max_frames: int, keep_log: bool) -> dict[str, Any]: driver_path = shutil.which("WebKitWebDriver") if not driver_path: raise WebDriverError("WebKitWebDriver is not installed; install webkit2gtk-driver and run under Xvfb") OUT.mkdir(parents=True, exist_ok=True) driver_port = free_port() log_path = OUT / "battle_effect_web_reference_webkitdriver.log" with log_path.open("wb") as log: proc = subprocess.Popen( [ driver_path, "--host=127.0.0.1", f"--port={driver_port}", "--replace-on-new-session", ], stdout=log, stderr=subprocess.STDOUT, ) session_id = "" try: wait_for_driver(driver_port, proc) session = request_json( driver_port, "POST", "/session", {"capabilities": {"alwaysMatch": {"browserName": "MiniBrowser"}}}, timeout=30, ) session_id = str((session or {}).get("value", {}).get("sessionId") or "") if not session_id: raise WebDriverError(f"could not create WebKit session: {session!r}") request_json( driver_port, "POST", f"/session/{session_id}/window/rect", {"x": 0, "y": 0, "width": 1420, "height": 960}, timeout=8, ) request_json( driver_port, "POST", f"/session/{session_id}/url", {"url": base.rstrip("/") + "/web/battle_skill_timeline_review.html"}, timeout=30, ) wait_for_page_ready(driver_port, session_id) captured: list[dict[str, Any]] = [] for row in rows: check = select_payload(row) marker = select_action(driver_port, session_id, check) ticks = tick_samples(marker, tick_mode, max_frames) rid = row_id(row) out_dir = CAPTURES / safe_row_dir(rid) / "web_reference" frames = [] for tick in ticks: frames.append(save_preview_png(driver_port, session_id, tick, out_dir / f"tick_{tick:05d}.png")) captured.append({ "id": rid, "ownerKey": row.get("ownerKey") or "", "skillIdHex": str(row.get("skillIdHex") or "").lower(), "levelOrFixed": row.get("levelOrFixed"), "skillName": row.get("skillName") or "", "webReferenceCapture": str(out_dir.relative_to(ROOT)), "captureTicks": ticks, "frameCount": len(frames), "frames": frames, "effectDiagnostics": marker.get("effectDiagnostics") or {}, }) return {"status": "ok", "captured": captured} finally: if session_id: try: request_json(driver_port, "DELETE", f"/session/{session_id}", timeout=5) except Exception: pass proc.terminate() try: proc.wait(timeout=3) except subprocess.TimeoutExpired: proc.kill() if not keep_log: try: log_path.unlink() except FileNotFoundError: pass def capture_with_managed_server(rows: list[dict[str, Any]], tick_mode: str, max_frames: int, keep_log: bool) -> dict[str, Any]: server_port = free_port() server_log_path = OUT / "battle_effect_web_reference_server.log" OUT.mkdir(parents=True, exist_ok=True) with server_log_path.open("wb") as log: proc = subprocess.Popen( [ sys.executable, str(ROOT / "tools" / "serve_utf8.py"), str(server_port), "--bind", "127.0.0.1", "--directory", str(ROOT), ], cwd=ROOT, stdout=log, stderr=subprocess.STDOUT, ) try: base = f"http://127.0.0.1:{server_port}" wait_for_http(base + "/web/battle_skill_timeline_review.html", proc) return capture_with_browser(base, rows, tick_mode=tick_mode, max_frames=max_frames, keep_log=keep_log) finally: proc.terminate() try: proc.wait(timeout=3) except subprocess.TimeoutExpired: proc.kill() if not keep_log: try: server_log_path.unlink() except FileNotFoundError: pass def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--scope", choices=["first-pass", "full"], default="first-pass") parser.add_argument("--tick-mode", choices=["samples", "range"], default="samples") parser.add_argument("--max-frames", type=int, default=12, help="Maximum ticks per row; 0 means no cap.") parser.add_argument("--base", default="", help="Existing static server base URL. If omitted, a temporary serve_utf8 server is started.") parser.add_argument("--keep-log", action="store_true") parser.add_argument("--no-manifest", action="store_true", help="Do not update the local ignored capture manifest.") args = parser.parse_args() rows = load_plan_rows(args.scope) if args.base: report = capture_with_browser(args.base, rows, tick_mode=args.tick_mode, max_frames=args.max_frames, keep_log=args.keep_log) else: report = capture_with_managed_server(rows, tick_mode=args.tick_mode, max_frames=args.max_frames, keep_log=args.keep_log) if not args.no_manifest: update_capture_manifest(report["captured"]) summary = { "status": report["status"], "scope": args.scope, "tickMode": args.tick_mode, "rows": len(report["captured"]), "frames": sum(int(row.get("frameCount") or 0) for row in report["captured"]), "captureRoot": str(CAPTURES.relative_to(ROOT)), "manifest": str(MANIFEST.relative_to(ROOT)), } print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())