#!/usr/bin/env python3
"""Capture comparable player-height views of the L138 and L139 owner shaft layer."""
from __future__ import annotations
import json
import math
import os
from pathlib import Path
import re
import subprocess
import sys
import time
ROOT = Path(__file__).resolve().parents[4]
DATA = ROOT / "Standalone/data/owner_world.json"
LAYOUT = ROOT / "Saved/Linux/l139/seeds/seed0.json"
OUT = ROOT / "Saved/Linux/l139/captures"
ARTIFACTS = ROOT / "Standalone/Research/WorldAssemblerM3/Linux139/captures"
LIMIT = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--"]
VERSIONS = {
"before-l138": {
"headless": ROOT / "Saved/Linux/l138/bin/VoxelForgeCoreHeadless-l138-clang",
"renderer": ROOT / "Saved/Linux/l138/bin/VoxelForgeOffscreenCapture-l138-clang",
"stamp": "9aade4e6cb33e86c36b89440644d9ef5d440288fb6ff0af03af128291c95f93f",
},
"after-l139": {
"headless": ROOT / "Saved/Linux/l139/bin/VoxelForgeCoreHeadless-l139-clang",
"renderer": ROOT / "Saved/Linux/l139/bin/VoxelForgeOffscreenCapture-l139-clang",
"stamp": "b8603b6cb3975ca0f9669759e4e2464561c557fdb7aa6836acef123ecf0b99d1",
},
}
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, 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"[L139CaptureRun] step={name} elapsed_s={time.monotonic()-started:.1f} "
f"log={log.relative_to(ROOT)}", flush=True)
return output
def point_um(values: list[int]) -> tuple[float, float, float]:
return tuple(value / 1_000_000.0 for value in values)
def make_views(layout: dict) -> list[dict]:
edges = {row["stable_id"]: row for row in layout["cavern_edges"]}
rooms = {row["stable_id"]: row for row in layout["cavern_rooms"]}
shaft = rooms["vertical-shafts/strate-3/shaft-0/primitive-0"]
access = edges["vertical-shafts/strate-3/shaft-access-0"]["polyline_um"]
arrival = edges["vertical-shafts/strate-3/route-edge-0"]["polyline_um"]
basal = rooms["vertical-shafts/strate-3/basal-landing/primitive-0"]
center = point_um(shaft["center_um"])
half = point_um(shaft["half_extents_um"])
shaft_top = center[2] + half[2]
shaft_bottom = center[2] - half[2]
route_a, route_b = point_um(arrival[0]), point_um(arrival[1])
tx, ty = route_b[0] - route_a[0], route_b[1] - route_a[1]
tlen = max(1.0e-9, math.hypot(tx, ty))
tangent = (tx / tlen, ty / tlen)
mouth_camera = (route_a[0] - tangent[0] * 2.0,
route_a[1] - tangent[1] * 2.0, route_a[2] + 1.45)
mouth_target = (route_a[0] + tangent[0] * 8.0,
route_a[1] + tangent[1] * 8.0, route_a[2] + 1.45)
access_a, access_b = point_um(access[0]), point_um(access[-1])
access_camera = (access_a[0], access_a[1], access_a[2] + 1.45)
access_target = (access_b[0], access_b[1], access_b[2] + 1.45)
shaft_lip_camera = (access_b[0], access_b[1], access_b[2] + 1.45)
basal_center = point_um(basal["center_um"])
basal_camera = (basal_center[0], basal_center[1], basal["floor_um"] / 1.0e6 + 1.45)
basal_target = (basal_center[0] + tangent[0] * 8.0,
basal_center[1] + tangent[1] * 8.0, basal_camera[2])
ledge_z = shaft_top - 12 * (shaft["ledge_spacing_um"] / 1.0e6)
return [
{"name": "arrival-mouth", "camera": mouth_camera, "target": mouth_target,
"focus": (route_a[0], route_a[1], route_a[2] + 1.4, 12.0),
"note": "player eye at the passage arrival, looking into the connected layer"},
{"name": "landing-at-shaft-access", "camera": access_camera,
"target": access_target,
"focus": (center[0], center[1], access_a[2], 14.0),
"note": "walkable route landing and short shaft access tunnel"},
{"name": "looking-down-shaft", "camera": shaft_lip_camera,
"target": (center[0], center[1], shaft_top - 30.0),
"focus": (center[0], center[1], shaft_top - 25.0, 14.0),
"fov": 35.0,
"note": "player eye at the upper access landing, looking down the open shaft"},
{"name": "looking-up-from-ledge", "camera": (center[0], center[1], ledge_z + 1.45),
"target": (center[0], center[1], shaft_top - 2.0),
"focus": (center[0], center[1], shaft_top - 22.0, 20.0),
"note": "player eye at an internal ledge, looking up; reaching this ledge is a climb tier"},
{"name": "basal-terminal-landing", "camera": basal_camera,
"target": basal_target,
"focus": (basal_center[0], basal_center[1], basal_center[2], 16.0),
"note": "walkable terminal landing above the basal seal"},
]
def write_overview(layout: dict, destination: Path) -> None:
shafts = [room for room in layout["cavern_rooms"] if room.get("shape_kind") == 1]
route_edges = sorted((edge for edge in layout["cavern_edges"]
if edge["stable_id"].startswith("vertical-shafts/strate-3/route-edge-")),
key=lambda edge: int(edge["stable_id"].rsplit("-", 1)[1]))
if not shafts or not route_edges:
raise ValueError("overview needs the resolved shaft primitives and walk route")
points = [point_um(row) for edge in route_edges for row in edge["polyline_um"]]
shaft_x_bounds = [(point_um(room["center_um"])[0]
-room["half_extents_um"][0]/1e6,
point_um(room["center_um"])[0]+room["half_extents_um"][0]/1e6)
for room in shafts]
x_min = min(min(bounds[0] for bounds in shaft_x_bounds),
min(row[0] for row in points))
x_max = max(x + room["half_extents_um"][0] / 1e6
for room in shafts for x in [point_um(room["center_um"])[0]])
x_max = max(x_max, max(row[0] for row in points))
z_min = min(point_um(room["center_um"])[2] - room["half_extents_um"][2] / 1e6
for room in shafts)
z_max = max(point_um(room["center_um"])[2] + room["half_extents_um"][2] / 1e6
for room in shafts)
z_min = min(z_min, min(row[2] for row in points))
z_max = max(z_max, max(row[2] for row in points))
width, height, margin = 1800.0, 760.0, 72.0
sx = (width - 2 * margin) / max(1.0, x_max - x_min)
sz = (height - 2 * margin) / max(1.0, z_max - z_min)
scale = min(sx, sz)
x_offset = margin + ((width - 2 * margin) - (x_max - x_min) * scale) * 0.5
z_offset = margin + ((height - 2 * margin) - (z_max - z_min) * scale) * 0.5
def xy(x: float, z: float) -> tuple[float, float]:
return x_offset + (x - x_min) * scale, height - (z_offset + (z - z_min) * scale)
route_xy = [xy(p[0], p[2]) for p in points]
route = " ".join(("M" if i == 0 else "L") + f"{x:.2f},{y:.2f}"
for i, (x, y) in enumerate(route_xy))
cylinders = []
for room in shafts:
x, _y, z = point_um(room["center_um"])
radius = room["half_extents_um"][0] / 1e6
half_z = room["half_extents_um"][2] / 1e6
left, top = xy(x - radius, z + half_z)
right, bottom = xy(x + radius, z - half_z)
cylinders.append(f'')
rx0, ry0 = xy(x_min, 0.0)
rx1, ry1 = xy(x_max, 0.0)
svg = f''''''
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_text(svg, encoding="utf-8")
def main() -> int:
if not DATA.is_file() or not LAYOUT.is_file():
raise SystemExit("missing owner fixture or passing seed-0 resolved layout")
layout = json.loads(LAYOUT.read_text(encoding="utf-8"))
views = make_views(layout)
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 = []
quality_failures = []
for version, config in VERSIONS.items():
phase = "before" if version.startswith("before") else "after"
for view in views:
name = view["name"]
camera, target = view["camera"], view["target"]
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"
export_log = OUT / f"{phase}-{name}-export.log"
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"{phase}-{name}-export", list(map(str, 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")
png = OUT / f"{phase}-{name}.png"
jpg = ARTIFACTS / f"{phase}-{name}.jpg"
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=-950",
"--height-range-m=60", "--tile-edges=off",
f"--focus-region={fmt(view['focus'])}"]
stamped_run(f"{phase}-{name}-render", list(map(str, render)),
OUT / f"{phase}-{name}-render.log", config["stamp"], env)
jpeg = ["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y",
"-i", png, "-frames:v", "1", "-q:v", "4", jpg]
subprocess.run([*LIMIT, "timeout", "120s", *map(str, jpeg)], 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:
quality_failures.append(f"{phase}-{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, "fov_deg": fov,
"note": view["note"],
"quality_status": "PASS" if quality_ok else "FAIL",
"quality": quality_result})
print(f"[L139Capture] phase={phase} view={name} file={jpg.relative_to(ROOT)} "
f"bytes={jpg.stat().st_size} quality={'PASS' if quality_ok else 'FAIL'}",
flush=True)
overview = ARTIFACTS / "after-whole-layer-side-elevation.svg"
write_overview(layout, overview)
manifest.append({"phase": "after", "view": "whole-layer-side-elevation",
"file": str(overview.relative_to(ROOT)), "shafts": sum(
room.get("shape_kind") == 1 for room in layout["cavern_rooms"]),
"walk_route_edges": len([edge for edge in layout["cavern_edges"]
if edge["stable_id"].startswith("vertical-shafts/strate-3/route-edge-")]),
"note": "resolved-plan side projection; not a density-mesh capture"})
(ARTIFACTS / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n",
encoding="utf-8")
status = "PASS" if not quality_failures else "FAIL"
print(f"[L139CaptureSummary] spot_views={len(views)} comparisons={len(views)*2} "
f"whole_layer_projection=1 quality_failures={len(quality_failures)} "
f"status={status}", flush=True)
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except Exception as error:
print(f"[L139CaptureFailure] {error}", file=sys.stderr, flush=True)
raise SystemExit(1)