"""Prove that a full-environment race render can resolve every input asset. This check is intentionally read-only. It must run before TTS synthesis, browser capture, or any other expensive render step. Cloud checks require every input to be deployable from a clean checkout; local checks accept private, Git-ignored inputs present on the recording machine. A non-zero status means that the report is the render stop signal for the selected target. """ from __future__ import annotations import argparse import hashlib import json import os import re import subprocess import time import urllib.error import urllib.request from dataclasses import dataclass from pathlib import Path HERE = Path(__file__).resolve().parent ROOT = HERE.parent.parent REPO = ROOT.parents[1] DEFAULT_RACES = ("gsys04", "gsys07", "gsys06") J80_FILES = ( "j80.obj", "j80.mtl", "J-80_base_D.tga.png", "J-80_base_N.tga.png", "J-80_base_R.tga.png", "J-80_base_A.tga.png", "J-80_sail_D.tga.png", "J-80_sail_N.tga.png", "J-80_sail_R.tga.png", "J-80_sail_MT.tga.png", ) @dataclass class Finding: category: str asset: str status: str detail: str remedy: str = "" #: The ion assets the 3D world needs: Google Photorealistic 3D Tiles, Cesium World Terrain, Bing #: aerial imagery, and OSM Buildings. Only the first is opt-in per account, so it is the one that can #: fail while everything else passes. CESIUM_ASSETS = {"2275207": "photoreal 3D tiles", "1": "world terrain", "2": "aerial imagery", "96188": "OSM buildings"} #: Which of them a render actually loads, by backdrop. `tiles.js` asks for the photoreal tileset ALONE #: when it has one (Google's own plugin takes over the tile traffic from there), and for terrain + #: draped imagery + OSM buildings otherwise. CESIUM_ASSETS_BY_WORLD = {"photoreal": ("2275207",), "terrain": ("1", "2", "96188")} #: How long a successful probe stands. A CESIUM ION SESSION IS ONE SUCCESSFUL GET OF AN ASSET'S #: `/endpoint` — the very request this preflight makes — so an unconditional probe charged the account #: once per render, before the render had drawn anything. A batch of twenty renders is one probe. #: Only AVAILABLE is cached: MISSING and UNPROVEN are the answers somebody acts on, and a stale "the #: token is refused" is worth re-asking every time. _ION_PROBE_TTL_S = 12 * 3600 def cesium_token() -> tuple[str, str]: """`(token, source)` — the env var the deployment supplies, else a developer's local file. Parses BOTH quote styles on purpose. `core/web/cesium-config.js` is hand-written with single quotes; the copy `live_feed_server` generates from the environment uses `json.dumps`, so double. A single-quote-only reader silently returns "" against the generated file, which reads exactly like a missing token — that mistake cost two false diagnoses of production being broken. """ env = os.environ.get("VR_CESIUM_ION_TOKEN", "").strip() if env: return env, "VR_CESIUM_ION_TOKEN" cfg = ROOT / "core" / "web" / "cesium-config.js" if cfg.is_file(): m = re.search(r"""CESIUM_ION_TOKEN\s*=\s*['"]([^'"]+)""", cfg.read_text(encoding="utf-8")) if m: return m.group(1), "core/web/cesium-config.js" return "", "" def _ion_probe_cache(report: Path) -> Path: """Beside the report, because that is the directory a render run already owns.""" return report.parent / ".ion-probe.json" def _load_ion_cache(path: Path, now_s: float) -> dict: """Fresh AVAILABLE verdicts only; anything else is treated as never asked.""" try: doc = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} if not isinstance(doc, dict): return {} return {k: v for k, v in (doc.get("ok") or {}).items() if isinstance(v, int | float) and now_s - v < _ION_PROBE_TTL_S} def _probe_cesium(findings: list, *, world: str = "terrain", report: Path | None = None, now_s: float | None = None) -> None: """ASK ion whether the token works, rather than recording that nobody has asked. This finding used to be a permanent "UNPROVEN" whose own remedy said to run a token-scoped asset probe — so the check documented the thing it declined to do, and a dead or unscoped token would sail through every preflight and surface as a bare sea in the render. The world is the one asset here that lives behind a credential; it is also the one nobody notices missing, because the scene still draws. A network failure is NOT a hard stop: rendering offline from a warm tile cache is legitimate, and a preflight that blocks on no internet would just get skipped. A token that is present and REFUSED is a different answer, and that one is actionable. """ token, source = cesium_token() if not token: findings.append(Finding( "venue / shoreline", "Cesium ion token", "MISSING", "no VR_CESIUM_ION_TOKEN in the environment and no core/web/cesium-config.js on disk — " "the 3D world cannot load, and the scene will render open water with no shoreline", "Set VR_CESIUM_ION_TOKEN on the render/producer service (Secret Manager), or place a " "local core/web/cesium-config.js for a developer run. Do not commit that file.")) return wanted = CESIUM_ASSETS_BY_WORLD.get(world) or CESIUM_ASSETS_BY_WORLD["terrain"] now_s = time.time() if now_s is None else now_s cache_path = _ion_probe_cache(report) if report is not None else None # The cache is keyed on the TOKEN too: a rotated token is a different question, and a cached # "yes" from the old one would be an answer about a credential that no longer exists. fingerprint = hashlib.sha256(token.encode("utf-8")).hexdigest()[:16] cached = _load_ion_cache(cache_path, now_s) if cache_path else {} fresh: dict[str, float] = {} for asset in wanted: what = CESIUM_ASSETS[asset] key = f"{fingerprint}:{asset}" if key in cached: findings.append(Finding("venue / shoreline", f"ion asset {asset} ({source})", "AVAILABLE", f"{what}: ion returned the tileset endpoint " f"{int((now_s - cached[key]) / 60)} min ago (cached; a probe is a " f"charged session)")) fresh[key] = cached[key] continue url = f"https://api.cesium.com/v1/assets/{asset}/endpoint?access_token={token}" try: with urllib.request.urlopen(url, timeout=8) as r: # noqa: S310 — fixed https host ok = r.status == 200 status, detail = ("AVAILABLE", f"{what}: ion returned the tileset endpoint") if ok else ( "UNPROVEN", f"{what}: unexpected status from ion") remedy = "" except urllib.error.HTTPError as e: status = "MISSING" detail = f"{what}: ion refused the token ({e.code}) for asset {asset}" remedy = ("Check the token is not revoked and that the ion account has this asset " "enabled — Google Photorealistic 3D Tiles (2275207) is opt-in per account.") except Exception as e: # offline / DNS / timeout status = "UNPROVEN" detail = f"{what}: could not reach ion ({type(e).__name__}) — cannot prove or disprove" remedy = "Allow HTTPS egress to api.cesium.com, or accept a cached-tile render." findings.append(Finding("venue / shoreline", f"ion asset {asset} ({source})", status, detail, remedy)) if status == "AVAILABLE": fresh[key] = now_s if cache_path is not None: try: cache_path.parent.mkdir(parents=True, exist_ok=True) cache_path.write_text(json.dumps({"ok": fresh}), encoding="utf-8") except OSError: pass # a cache that cannot be written costs a probe def tracked(path: Path) -> bool: rel = path.relative_to(REPO) return subprocess.run( ["git", "ls-files", "--error-unmatch", str(rel)], cwd=REPO, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ).returncode == 0 def add_file(findings: list[Finding], category: str, path: Path, *, remedy: str = "") -> None: rel = path.relative_to(REPO) if not path.is_file(): findings.append(Finding(category, str(rel), "MISSING", "not present in this environment", remedy)) elif not tracked(path): findings.append(Finding(category, str(rel), "LOCAL_ONLY", f"present ({path.stat().st_size} bytes) but not in Git", remedy)) else: findings.append(Finding(category, str(rel), "AVAILABLE", f"Git-backed ({path.stat().st_size} bytes)")) def inspect(race_ids: tuple[str, ...], lang: str, *, world: str = "terrain", report: Path | None = None) -> list[Finding]: findings: list[Finding] = [] manifest_path = ROOT / "demo/data/races/manifest.json" add_file(findings, "race track", manifest_path) if not manifest_path.is_file(): return findings manifest = json.loads(manifest_path.read_text(encoding="utf-8")) races = {race["id"]: race for race in manifest.get("races", [])} selected = [] referenced_videos = 0 for race_id in race_ids: race = races.get(race_id) if race is None: findings.append(Finding("race track", race_id, "MISSING", "race is absent from the programme manifest")) continue selected.append(race) feed = race.get("data") or {} meta = feed.get("meta") or {} gun = (meta.get("gunUtcMs", 0) - meta.get("trackStartUtcMs", 0)) / 1000 frame_times = sorted(float(t) for t in (feed.get("frames") or {})) start = gun - 120 coverage = bool(frame_times and frame_times[0] <= start <= frame_times[-1]) status = "AVAILABLE" if coverage else "MISSING" detail = f"inlined track; race time -2:00 maps to track second {start:g}; action coverage={frame_times[0]:g}..{frame_times[-1]:g}s" if frame_times else "no action frames" remedy = "" if coverage else "Re-export this race with telemetry frames covering at least gun minus 120 seconds, then replace only its inlined manifest feed." findings.append(Finding("race track", race_id, status, detail, remedy)) media_root = ROOT / "demo/data/media" remote_clips = 0 for entries in (feed.get("media") or {}).values(): for item in entries: referenced_videos += int(bool(item.get("video"))) kind = "video" if item.get("video") else "boat-camera image" ref = str(item.get("file") or "") # A boat-cam clip may be a SIGNED URL rather than a file in this checkout. The export # carries URLs on purpose — one championship's footage is ~600 GB, so the feed is render # input with a shelf life and the renderer fetches each clip over the network. Treating # a URL as a relative path builds `demo/data/media/https:/storage.googleapis.com/...`, # which is missing by construction: the first feed ever exported WITH media reported # 2278 blockers and refused to render, one per clip that was in fact perfectly fetchable. if ref.startswith(("http://", "https://")): remote_clips += 1 continue add_file(findings, kind, media_root / ref, remedy="Copy this referenced file to demo/data/media/ and upload that directory as a private render-input artifact (or include it in the cloud build context).") if remote_clips: # Reported, never silently passed: these are the one asset class this check cannot prove, # and a signed URL expires. The renderer holds the last picture when a clip fails, so a # dead URL costs tiles rather than the film -- which is why it is not a blocker. findings.append(Finding( "boat camera", f"{race_id}: {remote_clips} clip(s) by signed URL", "UNPROVEN", "fetched from cloud storage at capture time, not from this checkout; signatures expire", "If the tiles come out blank, re-export the race to mint fresh signed URLs.")) for item in feed.get("video") or []: referenced_videos += 1 name = item.get("file") if isinstance(item, dict) else item add_file(findings, "video", media_root / name, remedy="Upload the source clip to the private render-input artifact at the same relative path.") if not referenced_videos: findings.append(Finding("video", "selected programme feeds", "AVAILABLE", "no standalone or boat-camera video clips are referenced; still images are inventoried separately")) assets = ROOT / "core/web/assets" for name in J80_FILES: add_file(findings, "3D boat model / texture", assets / "j80" / name) add_file(findings, "crew texture", assets / "bibs.png") findings.append(Finding("3D crew model", "core/web/j80.js procedural crew", "AVAILABLE", "generated from committed geometry code; no binary model required")) _probe_cesium(findings, world=world, report=report) findings.append(Finding("water material", "core/web/scene.js procedural ripple and wave meshes", "AVAILABLE", "generated from committed code; no texture file required")) findings.append(Finding("lighting / shadow", "core/web/scene.js hemisphere, ambient, sun and shadow catcher", "AVAILABLE", "generated from committed code; no asset file required")) findings.append(Finding("audio", "countdown beeps and gun tone", "AVAILABLE", "generated by demo/record/build_audio.py; no source audio file required")) findings.append(Finding("runtime module", "https://esm.sh/3d-tiles-renderer@0.4.27", "UNPROVEN", "cloud egress and module availability have not been authenticated by this local check", "Allow HTTPS egress to esm.sh or vendor the pinned module into the render image.")) index_path = ROOT / "demo/data/audio/index.json" add_file(findings, "audio index", index_path, remedy="Synthesize the selected language once, then upload demo/data/audio/ as a private render-input artifact included by v3/broadcast/.gcloudignore.") index = {} if index_path.is_file(): index = json.loads(index_path.read_text(encoding="utf-8")).get(lang, {}) texts = { (event.get("i18n") or {}).get(lang) for race in selected for event in race["data"].get("events", []) if (event.get("i18n") or {}).get(lang) } for text in sorted(texts): clip = index.get(text) if not clip: findings.append(Finding("audio clip", f"{lang}:{text[:72]}", "MISSING", "commentary line has no indexed neural-audio clip", "Run live/tts/synth.py for the programme, then upload demo/data/audio/ as a private render-input artifact.")) continue name = clip.get("file") if isinstance(clip, dict) else clip add_file(findings, "audio clip", ROOT / "demo/data/audio" / name, remedy="Upload demo/data/audio/ as a private render-input artifact at the same relative path.") return findings def blockers_for(findings: list[Finding], target: str) -> list[Finding]: """Return findings that prevent a render on the selected execution target.""" blocked_statuses = {"MISSING"} if target == "local" else {"MISSING", "LOCAL_ONLY", "UNPROVEN"} return [finding for finding in findings if finding.status in blocked_statuses] def write_report(findings: list[Finding], output: Path, target: str = "cloud") -> None: blockers = blockers_for(findings, target) counts = {s: sum(f.status == s for f in findings) for s in ("AVAILABLE", "MISSING", "LOCAL_ONLY", "UNPROVEN")} decision = ( f"**Decision: STOP — do not start the {target} render.**" if blockers else f"**Decision: GO — all required assets are usable for the {target} render.**" ) lines = [ "# Full-environment render asset preflight", "", decision, "", "The programme starts each race at exactly race time −2:00; the inlined telemetry action remains visible beneath introduction overlays.", "", f"Summary: {counts['AVAILABLE']} available, {counts['MISSING']} missing, {counts['LOCAL_ONLY']} local-only, {counts['UNPROVEN']} unproven.", "", "| Category | Asset / expected path | Status | Evidence | Smallest safe remediation |", "|---|---|---|---|---|", ] def esc(value: object) -> str: return str(value).replace("|", "\\|").replace("\n", " ") for f in findings: lines.append(f"| {esc(f.category)} | `{esc(f.asset)}` | **{f.status}** | {esc(f.detail)} | {esc(f.remedy) or 'None'} |") output.parent.mkdir(parents=True, exist_ok=True) output.write_text("\n".join(lines) + "\n", encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--race", action="append", dest="races", help="race id; repeat as needed") parser.add_argument("--lang", default="en") parser.add_argument("--target", choices=("cloud", "local"), default="cloud") parser.add_argument("--report", type=Path, default=REPO / "docs/full-environment-asset-preflight.md") parser.add_argument("--world", choices=("terrain", "photoreal"), default="terrain", help="which 3D backdrop the render will use, so only the ion assets it " "actually loads are probed (default terrain). Each probe IS a charged " "ion session, so a terrain render must not pay for the photoreal bucket.") args = parser.parse_args() findings = inspect(tuple(args.races or DEFAULT_RACES), args.lang, world=args.world, report=args.report) write_report(findings, args.report, args.target) blockers = blockers_for(findings, args.target) print( f"asset preflight ({args.target}): {'FAIL' if blockers else 'PASS'} " f"({len(blockers)} blocker(s)); report: {args.report}" ) return 1 if blockers else 0 if __name__ == "__main__": raise SystemExit(main())