#!/usr/bin/env python3 """Capture matched eye-height views from both ends of every owner tunnel.""" from __future__ import annotations import json import math import os from pathlib import Path import re import subprocess import sys ROOT = Path(__file__).resolve().parents[4] DATA = ROOT / "Standalone/data/owner_world.json" OUT = ROOT / "Saved/Linux/l152/captures" LAYOUT = OUT / "after-resolved-layout.json" ARTIFACTS = ROOT / "Standalone/Research/WorldAssemblerM3/Linux152/captures" LIMIT = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--"] VERSIONS = { "before-l151": { "headless": ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-l151-clang", "renderer": ROOT / "Standalone/bin/linux/VoxelForgeOffscreenCapture-l151-clang", "stamp": "1ca3b84f6d35f7f355ec0f72dabd3b87a455ea02e5789fea611916962e5db8e5", }, "after-l152": { "headless": ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-l152-clang", "renderer": ROOT / "Standalone/bin/linux/VoxelForgeOffscreenCapture-l152-clang", "stamp": "b299076d814ff7cc098aa373cdc3a26533ed8bfe1ef6df1ea9de4d400d231b25", }, } STAMP_RE = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=([0-9a-f]{64})$", re.M) def fmt(values) -> str: return ",".join(f"{value:.6f}" for value in values) def angles(camera, target) -> tuple[float, float]: dx, dy, dz = (target[index] - camera[index] for index in range(3)) return math.degrees(math.atan2(dx, dy)), math.degrees( math.atan2(dz, math.hypot(dx, dy))) def point_q(layout: dict, values) -> tuple[float, float, float]: scale = layout["core_voxel_size_um"] / (256.0 * 1_000_000.0) return tuple(value * scale for value in values) def chamber_views(layout: dict) -> list[dict]: result = [] for passage in layout["passages"]: controls = passage["control_points"] for side, endpoint, neighbor in ( ("upper", controls[0], controls[1]), ("lower", controls[-1], controls[-2])): chamber = passage[f"{side}_landing_chamber"] center = point_q(layout, chamber["floor_center_q"]) a = point_q(layout, neighbor["position_q"]) b = point_q(layout, endpoint["position_q"]) dx, dy = a[0] - b[0], a[1] - b[1] length = math.hypot(dx, dy) if length <= 1.0e-9: raise ValueError(f"degenerate {side} tunnel tangent: {passage['stable_id']}") dx, dy = dx / length, dy / length eye_z = center[2] + 1.45 distance = 9.0 target = (center[0] + dx * distance, center[1] + dy * distance, eye_z + (chamber["floor_gradient_q16"][0] / 65536.0 * dx + chamber["floor_gradient_q16"][1] / 65536.0 * dy) * distance) camera = (center[0], center[1], eye_z) slug = passage["stable_id"].replace("passage/", "").replace("/", "-") result.append({"name": f"{slug}-{side}-landing", "passage": passage["stable_id"], "side": side, "camera": camera, "target": target, "focus": (center[0], center[1], eye_z, 18.0)}) return result def stamped_run(name: str, command: list[object], log: Path, expected_stamp: str, env: dict[str, str]) -> str: binary = Path(str(command[0])) probe = subprocess.run([*LIMIT, "timeout", "15s", str(binary), "--print-source-stamp"], cwd=ROOT, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False, timeout=20) match = STAMP_RE.search(probe.stdout or "") if probe.returncode or not match or match.group(1) != expected_stamp: raise RuntimeError(f"{name}: source stamp mismatch for {binary}") log.parent.mkdir(parents=True, exist_ok=True) with log.open("w", encoding="utf-8") as stream: result = subprocess.run([*LIMIT, "timeout", "1740s", *map(str, command)], cwd=ROOT, env=env, stdout=stream, stderr=subprocess.STDOUT, timeout=1745, check=False) output = log.read_text(encoding="utf-8", errors="replace") emitted = STAMP_RE.search(output) if result.returncode or not emitted or emitted.group(1) != expected_stamp: raise RuntimeError(f"{name}: export or render failed; see {log}") return output def resolve_layout(env: dict[str, str]) -> dict: config = VERSIONS["after-l152"] command = [config["headless"], DATA, "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", "--workers=12", "--clip-radius=6", "--max-clip-level=0", "--max-level=0", "--force-generate", "--seed=0", "--recipe=surface-world", "--resolved-layout-only", f"--resolved-layout-output={LAYOUT}"] stamped_run("after-l152-resolve-layout", list(map(str, command)), OUT / "after-l152-resolved-layout.log", config["stamp"], env) layout = json.loads(LAYOUT.read_text(encoding="utf-8")) if len(layout.get("passages", [])) != 3: raise RuntimeError("owner layout did not resolve the expected three tunnels") return layout def main() -> int: if not DATA.is_file(): raise SystemExit("missing owner fixture") OUT.mkdir(parents=True, exist_ok=True) ARTIFACTS.mkdir(parents=True, exist_ok=True) env = os.environ.copy() env["LP_NUM_THREADS"] = "12" env["XDG_CACHE_HOME"] = str(OUT / "xdg-cache") layout = resolve_layout(env) views = chamber_views(layout) quality_tool = ROOT / "Standalone/Tools/verify/capture_quality.py" manifest = [] failures = [] for version, config in VERSIONS.items(): for view in views: camera, target = view["camera"], view["target"] yaw, pitch = angles(camera, target) mesh_dir = OUT / "meshes" / version mesh_dir.mkdir(parents=True, exist_ok=True) mesh = mesh_dir / f"{view['name']}.vfmesh" capture_layout = mesh_dir / f"{view['name']}-layout.json" export = [config["headless"], DATA, "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", "--workers=12", "--clip-radius=6", "--max-clip-level=0", "--max-level=0", "--force-generate", "--seed=0", "--recipe=surface-world", "--render-distance-chunks=0", f"--camera={fmt((*camera, yaw, pitch))}", f"--resolved-layout-output={capture_layout}", f"--mesh-export={mesh}"] stamped_run(f"{version}-{view['name']}-export", list(map(str, export)), OUT / f"{version}-{view['name']}-export.log", config["stamp"], env) if not mesh.is_file() or mesh.stat().st_size == 0: raise RuntimeError(f"{version}-{view['name']}: empty mesh export") png = OUT / f"{version}-{view['name']}.png" jpg = ARTIFACTS / f"{version}-{view['name']}.jpg" render = [config["renderer"], mesh, png, "--size=1280x720", "--fov-deg=78.0", f"--camera-m={fmt(camera)}", f"--yaw-deg={yaw:.6f}", f"--pitch-deg={pitch:.6f}", "--tile-skirts=off", "--shade=readable", "--headlamp=on", "--height-tint=on", "--height-center-m=-950", "--height-range-m=60", "--tile-edges=off", f"--focus-region={fmt(view['focus'])}"] stamped_run(f"{version}-{view['name']}-render", list(map(str, render)), OUT / f"{version}-{view['name']}-render.log", config["stamp"], env) subprocess.run([*LIMIT, "timeout", "120s", "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y", "-i", png, "-frames:v", "1", "-q:v", "4", jpg], cwd=ROOT, env=env, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT, timeout=125) check = subprocess.run([*LIMIT, "timeout", "120s", sys.executable, quality_tool, jpg], cwd=ROOT, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False, timeout=125) quality_ok = check.returncode == 0 and "status=PASS" in check.stdout if not quality_ok: failures.append(f"{version}-{view['name']}") quality_line = (check.stdout.strip().splitlines()[-1] if check.stdout.strip() else f"quality exit={check.returncode}") manifest.append({"version": version, "view": view["name"], "passage": view["passage"], "side": view["side"], "file": str(jpg.relative_to(ROOT)), "camera_m": camera, "target_m": target, "yaw_deg": yaw, "pitch_deg": pitch, "source_stamp": config["stamp"], "pose_layout_identity": layout.get("stable_identity"), "quality_status": "PASS" if quality_ok else "FAIL", "quality": quality_line}) print(f"[L152Capture] version={version} view={view['name']} " f"quality={'PASS' if quality_ok else 'FAIL'} " f"file={jpg.relative_to(ROOT)}", flush=True) (ARTIFACTS / "manifest.json").write_text( json.dumps(manifest, indent=2) + "\n", encoding="utf-8") with (ARTIFACTS / "INDEX.md").open("w", encoding="utf-8") as index: index.write("# L152 matched owner tunnel landing captures\n\n" "Seed 0 camera poses resolved from the L152 owner layout and reused for both " "versions. Each view is at eye height at one of the three tunnel landing " "chambers. Before uses the L151 head build; after uses L152 source stamp " f"`{VERSIONS['after-l152']['stamp']}`.\n\n") for view in views: index.write(f"## {view['passage']} / {view['side']} landing\n\n" f"- Before: [image](before-l151-{view['name']}.jpg)\n" f"- After: [image](after-l152-{view['name']}.jpg)\n\n") index.write(f"Capture quality failures: {len(failures)}/{len(manifest)}.\n") print(f"[L152CaptureSummary] chamber_views={len(views)} " f"images={len(manifest)} quality_failures={len(failures)}", flush=True) return 0 if not failures else 1 if __name__ == "__main__": try: raise SystemExit(main()) except Exception as error: print(f"[L152CaptureFailure] {error}", file=sys.stderr, flush=True) raise SystemExit(1)