#!/usr/bin/env python3 """Capture the seed 0/2 cave-mouth joints before and after L149.""" 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/l149/captures" ARTIFACTS = ROOT / "Standalone/Research/WorldAssemblerM3/Linux149/captures" LIMIT = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--"] VERSIONS = { "before-l147": { "headless": ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-l147-clang", "renderer": ROOT / "Standalone/bin/linux/VoxelForgeOffscreenCapture-l147-clang", "stamp": "c6b8a100355e271ddc40eb5973569d2ff4e2ba8d9c628c72f8e0af2493f0609f", }, "after-l149": { "headless": ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-l149-clang", "renderer": ROOT / "Standalone/bin/linux/VoxelForgeOffscreenCapture-l149-clang", "stamp": "19dbe26ab76a629baadc54bb0ab93eaa22d6f878e6946df2b71509c12c9524e4", }, } JOINTS = ( (0, "tunnel-network", "tunnel-network/strate-2/route-edge-0"), (0, "tunnel-network", "tunnel-network/strate-2/departure-edge"), (2, "vertical-shafts", "vertical-shafts/strate-3/route-edge-0"), ) STAMP_RE = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=([0-9a-f]{64})$", re.M) def fmt(values: tuple[float, ...]) -> str: return ",".join(f"{value:.6f}" for value in values) def angles(camera: tuple[float, ...], target: tuple[float, ...]) -> tuple[float, float]: dx, dy, dz = (target[i] - camera[i] for i in range(3)) return math.degrees(math.atan2(dx, dy)), math.degrees( math.atan2(dz, math.hypot(dx, dy))) def 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 != 0 or not match or match.group(1) != expected_stamp: raise RuntimeError(f"{name}: executable 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 != 0 or not emitted or emitted.group(1) != expected_stamp: raise RuntimeError(f"{name}: process failed or source changed; see {log}") return output def q_to_metres(q: int, core_voxel_um: int) -> float: return q * core_voxel_um / (256.0 * 1_000_000.0) def camera_for(layout: dict, edge: dict) -> tuple[tuple[float, ...], tuple[float, ...]]: anchor = edge["start_anchor_stable_id"] if anchor.endswith("/lower-mouth"): side = "lower" mouth_q = next(p["lower_mouth_q"] for p in layout["passages"] if p["stable_id"] == anchor.removesuffix("/lower-mouth")) elif anchor.endswith("/upper-mouth"): side = "upper" mouth_q = next(p["upper_mouth_q"] for p in layout["passages"] if p["stable_id"] == anchor.removesuffix("/upper-mouth")) else: raise ValueError(f"edge has no passage mouth anchor: {anchor}") passage = next(p for p in layout["passages"] if p["stable_id"] == anchor.rsplit("/", 1)[0]) controls = passage["control_points"] mouth_index = len(controls) - 1 if side == "lower" else 0 neighbor_index = mouth_index - 1 if side == "lower" else 1 a, b = controls[neighbor_index]["position_q"], controls[mouth_index]["position_q"] sign = 1.0 if side == "lower" else -1.0 tx, ty = sign * (b[0] - a[0]), sign * (b[1] - a[1]) tangent_length = math.hypot(tx, ty) if tangent_length <= 1.0e-9: raise ValueError(f"passage has a degenerate {side} mouth tangent: {anchor}") tx, ty = tx / tangent_length, ty / tangent_length scale = layout["core_voxel_size_um"] mx, my, mz = (q_to_metres(value, scale) for value in mouth_q) camera = (mx - tx * 2.0, my - ty * 2.0, mz + 0.47) target = (mx + tx * 10.0, my + ty * 10.0, mz + 0.47) return camera, target def main() -> int: if not DATA.is_file(): raise SystemExit("missing composed 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") quality = ROOT / "Standalone/Tools/verify/capture_quality.py" manifest = [] failures = [] for version, config in VERSIONS.items(): phase = "before" if version.startswith("before") else "after" for seed, kind, edge_id in JOINTS: name = f"seed{seed}-{kind}-{edge_id.rsplit('/', 1)[-1]}" reference_path = ROOT / f"Saved/Linux/l147/seeds/{kind}/seed{seed}.json" reference_layout = json.loads(reference_path.read_text(encoding="utf-8")) reference_edge = next(row for row in reference_layout["cavern_edges"] if row["stable_id"] == edge_id) camera, target = camera_for(reference_layout, reference_edge) yaw, pitch = angles(camera, target) mesh = OUT / "meshes" / phase / f"{name}.vfmesh" layout_path = OUT / "meshes" / phase / f"{name}-layout.json" export_log = OUT / f"{phase}-{name}-export.log" export = [config["headless"], DATA, "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", "--workers=8", "--clip-radius=6", "--max-clip-level=0", "--max-level=0", "--force-generate", f"--seed={seed}", "--recipe=surface-world", "--render-distance-chunks=0", f"--camera={fmt((*camera, yaw, pitch))}", f"--resolved-layout-output={layout_path}", f"--mesh-export={mesh}"] run(f"{phase}-{name}-export", export, export_log, config["stamp"], env) if not mesh.is_file() or mesh.stat().st_size == 0: raise RuntimeError(f"{phase}-{name}: empty mesh export") layout = json.loads(layout_path.read_text(encoding="utf-8")) png = OUT / f"{phase}-{name}.png" jpg = ARTIFACTS / f"{phase}-{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((*camera[:3], 12.0))}"] run(f"{phase}-{name}-render", render, OUT / f"{phase}-{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, 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 quality_result = (check.stdout.strip().splitlines()[-1] if check.stdout.strip() else f"quality command exit={check.returncode}") if not quality_ok: failures.append(name) manifest.append({"phase": phase, "view": name, "file": str(jpg.relative_to(ROOT)), "bytes": jpg.stat().st_size, "camera_m": camera, "target_m": target, "yaw_deg": yaw, "pitch_deg": pitch, "quality_status": "PASS" if quality_ok else "FAIL", "quality": quality_result}) print(f"[L149MouthCapture] phase={phase} view={name} " f"file={jpg.relative_to(ROOT)} quality={'PASS' if quality_ok else 'FAIL'}", flush=True) (ARTIFACTS / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") print(f"[L149MouthCaptureSummary] views={len(JOINTS)} comparisons={len(manifest)//2} " f"quality_failures={len(failures)} status={'PASS' if not failures else 'FAIL'}", flush=True) return 0 if not failures else 1 if __name__ == "__main__": try: raise SystemExit(main()) except Exception as error: print(f"[L149MouthCaptureFailure] {error}", file=sys.stderr, flush=True) raise SystemExit(1)