#!/usr/bin/env python3 """Capture comparable player-height and wide views of owner S1 before and after L140.""" from __future__ import annotations import json import argparse import math import os import re import subprocess import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parents[4] DATA = ROOT / "Standalone/data/owner_world.json" LAYOUT = ROOT / "Saved/Linux/l140/seeds/seed0-composed-layout.json" OUT = ROOT / "Saved/Linux/l140/captures" ARTIFACTS = ROOT / "Standalone/Research/WorldAssemblerM3/Linux140/captures" LIMIT = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--"] STAMP_RE = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=([0-9a-f]{64})$", re.M) VERSIONS = { "before-head": { "headless": ROOT / "Saved/Linux/l139/bin/VoxelForgeCoreHeadless-l139-clang", "renderer": ROOT / "Saved/Linux/l139/bin/VoxelForgeOffscreenCapture-l139-clang", "stamp": "b8603b6cb3975ca0f9669759e4e2464561c557fdb7aa6836acef123ecf0b99d1", "native": False, }, "after-l140": { "headless": ROOT / "Saved/Linux/l140/bin/VoxelForgeCoreHeadless-l140-clang", "renderer": ROOT / "Saved/Linux/l140/bin/VoxelForgeOffscreenCapture-l140-clang", "stamp": "6205e6c180ef32c41792dc3e0d8bf443a96df41974f40334af4bf3a157ab88da", "native": True, }, } def fmt(values: tuple[float, ...]) -> str: return ",".join(f"{value:.6f}" for value in values) def angles(camera: tuple[float, float, float], target: tuple[float, float, 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 stamped_run(name: str, command: list[str], log: Path, expected: str, env: dict[str, str]) -> str: binary = Path(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: raise RuntimeError(f"{name}: executable stamp mismatch for {binary}") log.parent.mkdir(parents=True, exist_ok=True) started = time.monotonic() 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: raise RuntimeError(f"{name}: process failed or source changed; see {log}") print(f"[L140CaptureRun] step={name} elapsed_s={time.monotonic()-started:.1f} " f"log={log.relative_to(ROOT)}", flush=True) return output def qpoint(values: list[int], scale_m: float) -> tuple[float, float, float]: return tuple(value * scale_m for value in values[:3]) def make_views(layout: dict) -> list[dict[str, object]]: scale_m = layout["core_voxel_size_um"] / 1_000_000.0 / 256.0 strate = next(row for row in layout["strates"] if row["strate_index"] == 1) anchors = strate["anchors"] route = [qpoint(row["values"], scale_m) for row in anchors if row["kind"] == "route_point"] islands = [row for row in anchors if row["kind"] == "island_anchor"] if len(route) < 8 or len(islands) < 5: raise ValueError("seed-0 resolved layout lacks a complete S1 route or island chain") def tangent(index: int) -> tuple[float, float]: a, b = route[max(0, index - 1)], route[min(len(route) - 1, index + 1)] dx, dy = b[0] - a[0], b[1] - a[1] length = max(1e-9, math.hypot(dx, dy)) return dx / length, dy / length def eye(point: tuple[float, float, float]) -> tuple[float, float, float]: return point[0], point[1], point[2] + 1.45 i = len(route) // 3 tx, ty = tangent(i) arrival_t = tangent(0) departure_t = tangent(len(route) - 1) arrival, departure = route[0], route[-1] arrival_camera = eye((arrival[0] - 2.5 * arrival_t[0], arrival[1] - 2.5 * arrival_t[1], arrival[2])) arrival_target = eye((arrival[0] + 8.0 * arrival_t[0], arrival[1] + 8.0 * arrival_t[1], arrival[2])) departure_camera = eye((departure[0] - 2.5 * departure_t[0], departure[1] - 2.5 * departure_t[1], departure[2])) departure_target = eye((departure[0] + 8.0 * departure_t[0], departure[1] + 8.0 * departure_t[1], departure[2])) edge_point = route[i] edge_camera = eye((edge_point[0] + 6.0 * ty, edge_point[1] - 6.0 * tx, edge_point[2])) edge_target = (edge_point[0] + 4.0 * tx, edge_point[1] + 4.0 * ty, edge_point[2] + 0.35) centers = [qpoint(row["values"], scale_m) for row in islands] middle_index = len(route) // 2 midpoint = route[middle_index] mid_t = tangent(middle_index) walk_camera = eye((midpoint[0] - 2.0 * mid_t[0], midpoint[1] - 2.0 * mid_t[1], midpoint[2])) next_island_index = min(len(centers) - 1, 5) next_island_top = centers[next_island_index][2] + islands[next_island_index]["values"][4] * scale_m walk_target = (centers[next_island_index][0], centers[next_island_index][1], next_island_top + 0.5) walk_focus = tuple((centers[4][axis] + centers[5][axis]) * 0.5 for axis in range(3)) + (18.0,) cx = sum(p[0] for p in centers) / len(centers) cy = sum(p[1] for p in centers) / len(centers) cz = sum(p[2] + row["values"][4] * scale_m for p, row in zip(centers, islands)) / len(centers) wide_camera = (cx, cy + 23.0, cz + 10.0) wide_target = (cx, cy, cz) return [ {"name": "arrival-mouth", "camera": arrival_camera, "target": arrival_target, "focus": (*arrival, 14.0), "note": "player eye at the fixed 0-1 arrival mouth"}, {"name": "island-edge", "camera": edge_camera, "target": edge_target, "focus": (*edge_point, 18.0), "note": "player eye beside the route on an island edge"}, {"name": "walk-between-islands", "camera": walk_camera, "target": walk_target, "focus": walk_focus, "note": "player eye looking from one route landing toward the next island"}, {"name": "departure-mouth", "camera": departure_camera, "target": departure_target, "focus": (*departure, 14.0), "note": "player eye approaching the fixed 1-2 departure mouth"}, {"name": "wide-silhouette", "camera": wide_camera, "target": wide_target, "focus": (cx, cy, cz, 48.0), "fov": 72.0, "note": "wide side view showing the island chain silhouette"}, ] def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--views", nargs="*", default=None, help="optional view names to rerender in both phases") args = parser.parse_args() if not DATA.is_file() or not LAYOUT.is_file(): raise SystemExit("missing owner fixture or final seed-0 resolved layout") layout = json.loads(LAYOUT.read_text(encoding="utf-8")) views = make_views(layout) if args.views: unknown = set(args.views) - {str(view["name"]) for view in views} if unknown: raise SystemExit(f"unknown capture views: {sorted(unknown)}") views = [view for view in views if view["name"] in set(args.views)] 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: list[dict[str, object]] = [] old_manifest_path = ARTIFACTS / "manifest.json" existing = json.loads(old_manifest_path.read_text(encoding="utf-8")) \ if old_manifest_path.is_file() else {"captures": []} rerendered = {str(view["name"]) for view in views} for row in existing.get("captures", []): if row.get("view") not in rerendered: manifest.append(row) for version, config in VERSIONS.items(): phase = "before" if version.startswith("before") else "after" for view in views: name = str(view["name"]) camera, target = view["camera"], view["target"] assert isinstance(camera, tuple) and isinstance(target, tuple) yaw, pitch = angles(camera, target) mesh_dir = OUT / "meshes" / phase mesh_dir.mkdir(parents=True, exist_ok=True) mesh = mesh_dir / f"{name}.vfmesh" capture_layout = mesh_dir / f"{name}-layout.json" flags = ["--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}"] if config["native"]: flags.extend(["--native-owner-fi", "--density-dispatch-report"]) export = [config["headless"], DATA, *flags] stamped_run(f"{phase}-{name}-export", list(map(str, export)), OUT / f"{phase}-{name}-export.log", config["stamp"], env) if not mesh.is_file() or mesh.stat().st_size == 0: raise RuntimeError(f"{phase}-{name}: empty mesh export") png = OUT / f"{phase}-{name}.png" jpg = ARTIFACTS / f"{phase}-{name}.jpg" focus = view["focus"] assert isinstance(focus, tuple) fov = float(view.get("fov", 78.0)) render = [config["renderer"], mesh, png, "--size=1280x720", f"--fov-deg={fov:.1f}", 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=-740", "--height-range-m=70", "--tile-edges=off", f"--focus-region={fmt(focus)}"] stamped_run(f"{phase}-{name}-render", list(map(str, 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) result = check.stdout.strip().splitlines()[-1] if check.stdout.strip() else f"exit={check.returncode}" ok = check.returncode == 0 and "status=PASS" in check.stdout 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, "fov_deg": fov, "note": view["note"], "quality_status": "PASS" if ok else "FAIL", "quality": result}) print(f"[L140Capture] phase={phase} view={name} file={jpg.relative_to(ROOT)} " f"quality={'PASS' if ok else 'FAIL'} {result}", flush=True) manifest.sort(key=lambda row: (str(row["phase"]), str(row["view"]))) summary = {"seed": 0, "source_stamp": VERSIONS["after-l140"]["stamp"], "before_stamp": VERSIONS["before-head"]["stamp"], "views_per_phase": 5, "all_quality_pass": all(row["quality_status"] == "PASS" for row in manifest), "captures": manifest} (ARTIFACTS / "manifest.json").write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8") print(f"[L140CaptureSummary] views_per_phase={len(views)} captures={len(manifest)} " f"all_quality_pass={summary['all_quality_pass']}", flush=True) return 0 if summary["all_quality_pass"] else 1 if __name__ == "__main__": raise SystemExit(main())