Stamp vocabulary v7 (opt-in fillBox, carveBox, fillSteps; carveArch reused); flat-field liminal stamp fixtures; capsule walk audit with doorway-width gate (1.2 m control fails as expected). Clang/GCC W1/W8/W12 identical; owner canonical hashes unchanged; fast tier 273/273. Terraced wide view not yet legible. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016dhxM43dnWvUu2B5Ph1iG1
321 lines
16 KiB
Python
321 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Run L161 stamp-fixture determinism, walk audits and two-view captures."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[4]
|
|
OUT_DEFAULT = Path(os.environ.get("VOXELFORGE_VERIFY_OUT", ROOT / "Saved/Linux/l161"))
|
|
CAPTURE_DEFAULT = Path(__file__).resolve().parent / "captures"
|
|
LIMIT = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--"]
|
|
Q_PER_METRE = 1024.0
|
|
FOG_COLOR = "0.78,0.84,0.87"
|
|
|
|
sys.path.insert(0, str(ROOT / "Standalone/Tools/verify"))
|
|
from runner import current_stamp, first_stamp
|
|
|
|
|
|
def fmt(values) -> str:
|
|
return ",".join(f"{value:.6f}" for value in values)
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(block)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def run_step(name: str, command: list[object], log: Path, *, env: dict[str, str],
|
|
stamped: bool = False, expected_exit: int = 0,
|
|
timeout_seconds: int = 1740) -> str:
|
|
expected_stamp = current_stamp() if stamped else None
|
|
if stamped:
|
|
probe = subprocess.run([str(command[0]), "--print-source-stamp"], cwd=ROOT,
|
|
capture_output=True, text=True, timeout=15, check=False)
|
|
if probe.returncode != 0 or first_stamp(probe.stdout) != expected_stamp:
|
|
raise RuntimeError(f"{name} refused a stale executable: {command[0]}")
|
|
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", f"{timeout_seconds}s",
|
|
*map(str, command)], cwd=ROOT, env=env, stdout=stream,
|
|
stderr=subprocess.STDOUT, timeout=timeout_seconds + 5, check=False)
|
|
elapsed = time.monotonic() - started
|
|
output = log.read_text(encoding="utf-8", errors="replace")
|
|
print(f"[L161Run] step={name} elapsed_s={elapsed:.3f} exit={result.returncode} log={log}",
|
|
flush=True)
|
|
if stamped and (first_stamp(output) != expected_stamp or current_stamp() != expected_stamp):
|
|
raise RuntimeError(f"{name} source stamp mismatch; see {log}")
|
|
if result.returncode != expected_exit:
|
|
raise RuntimeError(f"{name} exit={result.returncode}, expected={expected_exit}; see {log}")
|
|
return output
|
|
|
|
|
|
def camera_for(view: dict, floor: float, name: str, kind: str) -> tuple[float, float, float]:
|
|
center_x, center_y = view["center_xy_m"]
|
|
offset = view[f"{kind}_camera_offset_m"]
|
|
z = floor + offset[2]
|
|
if name == "backrooms-grid" or name == "backrooms-narrow-control":
|
|
z = floor + offset[2]
|
|
return center_x + offset[0], center_y + offset[1], z
|
|
|
|
|
|
def target_for(view: dict, floor: float, name: str, kind: str) -> tuple[float, float, float]:
|
|
center_x, center_y = view["center_xy_m"]
|
|
offset = view[f"{kind}_target_offset_m"]
|
|
return center_x + offset[0], center_y + offset[1], floor + offset[2]
|
|
|
|
|
|
def headless_command(binary: Path, fixture: Path, camera, mesh: Path,
|
|
manifest: Path, layout: Path, workers: int, clip_radius: int,
|
|
*, max_level: int = 0, hash_tiles: bool = False,
|
|
audit_path: Path | None = None) -> list[object]:
|
|
command: list[object] = [binary, fixture, f"--workers={workers}", "--seed=0",
|
|
"--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
|
|
"--force-generate", f"--clip-radius={clip_radius}",
|
|
f"--max-clip-level={max_level}", f"--max-level={max_level}",
|
|
"--render-distance-chunks=0",
|
|
f"--camera={fmt((*camera, 0.0, 0.0))}", f"--mesh-export={mesh}",
|
|
f"--world-assembler-manifest-output={manifest}",
|
|
f"--resolved-layout-output={layout}"]
|
|
if hash_tiles:
|
|
command.append("--stream-hash-tiles")
|
|
if audit_path is not None:
|
|
command.append(f"--audit-liminal-stamp-walk={audit_path}")
|
|
return command
|
|
|
|
|
|
def digest(output: str, manifest_path: Path, layout_path: Path) -> tuple[str, str, str]:
|
|
tile_hashes = sorted(line for line in output.splitlines()
|
|
if line.startswith("[PipelineTileHash] "))
|
|
if not manifest_path.is_file() or not layout_path.is_file() or not tile_hashes:
|
|
raise RuntimeError("determinism run omitted manifest, resolved layout or tile hashes")
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
layout = json.loads(layout_path.read_text(encoding="utf-8"))
|
|
manifest_digest = "|".join((manifest["content_hash"],
|
|
manifest["recipe"]["content_hash"], manifest["plan"]["content_hash"]))
|
|
return manifest_digest, layout["stable_identity"], "\n".join(tile_hashes)
|
|
|
|
|
|
def write_walk_audit(spec_path: Path, manifest_path: Path, audit_path: Path) -> None:
|
|
spec = json.loads(spec_path.read_text(encoding="utf-8"))
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
surface = manifest["recipe"]["surface_profile"]
|
|
floor = surface["h0_q"] / Q_PER_METRE
|
|
lines = ["# P records are world metres; doorway width axes use the resolved stamp's local extents."]
|
|
for gate in spec["door_prefixes"]:
|
|
lines.append(f"D {gate['prefix']} {gate['width_axis']} {gate['minimum_width_m']:.6f}")
|
|
for x, y, z_offset in spec["path_points_offset_m"]:
|
|
lines.append(f"P {x:.6f} {y:.6f} {floor + z_offset:.6f}")
|
|
audit_path.parent.mkdir(parents=True, exist_ok=True)
|
|
audit_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def camera_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 build_plan(binary: Path, name: str, fixture: Path, view: dict,
|
|
out: Path, env: dict[str, str]) -> tuple[Path, Path, dict, dict]:
|
|
run_dir = out / "plan-probes" / name
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
manifest, layout = run_dir / "manifest.json", run_dir / "layout.json"
|
|
mesh = run_dir / "probe.vfmesh"
|
|
camera = (view["center_xy_m"][0], view["center_xy_m"][1], -500.0)
|
|
output = run_step(f"plan-{name}", headless_command(binary, fixture, camera,
|
|
mesh, manifest, layout, 12, 1), out / "logs" / f"plan-{name}.log",
|
|
env=env, stamped=True)
|
|
if "[LiminalStampFixtureResolved] vocabulary=7" not in output:
|
|
raise RuntimeError(f"{name} did not resolve with stamp vocabulary 7")
|
|
if not manifest.is_file() or not layout.is_file():
|
|
raise RuntimeError(f"{name} plan probe omitted its outputs")
|
|
manifest_value = json.loads(manifest.read_text(encoding="utf-8"))
|
|
layout_value = json.loads(layout.read_text(encoding="utf-8"))
|
|
return manifest, layout, manifest_value, layout_value
|
|
|
|
|
|
def run_determinism(name: str, fixture: Path, view: dict, floor: float,
|
|
binaries: dict[str, Path], out: Path, env: dict[str, str]) -> None:
|
|
camera = camera_for(view, floor, name, "eye")
|
|
baseline = None
|
|
for compiler in ("clang", "gcc"):
|
|
for workers in (1, 8, 12):
|
|
tag = f"{name}-{compiler}-w{workers}"
|
|
run_dir = out / "determinism" / tag
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
mesh, manifest, layout = (run_dir / "fixture.vfmesh",
|
|
run_dir / "manifest.json", run_dir / "layout.json")
|
|
output = run_step(tag, headless_command(binaries[compiler], fixture,
|
|
camera, mesh, manifest, layout, workers, 1,
|
|
hash_tiles=True), out / "logs" / f"determinism-{tag}.log",
|
|
env=env, stamped=True)
|
|
result = digest(output, manifest, layout)
|
|
if baseline is None:
|
|
baseline = result
|
|
elif result != baseline:
|
|
raise RuntimeError(f"{name} diverged at {compiler} W{workers}; see determinism logs")
|
|
print(f"[L161Determinism] fixture={name} compiler_worker_cells=6 identical=yes status=PASS",
|
|
flush=True)
|
|
|
|
|
|
def run_walk_audit(name: str, fixture: Path, view: dict, floor: float,
|
|
spec: Path, binary: Path, out: Path, env: dict[str, str],
|
|
*, expected_exit: int = 0) -> str:
|
|
camera = camera_for(view, floor, name, "eye")
|
|
run_dir = out / "walk-audits" / name
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
manifest = run_dir / "manifest.json"
|
|
layout = run_dir / "layout.json"
|
|
mesh = run_dir / "audit.vfmesh"
|
|
audit_path = run_dir / "walk.path"
|
|
# Use a fixture-specific manifest, so the path uses the exact authored floor height.
|
|
plan = out / "plan-probes" / name / "manifest.json"
|
|
if not plan.is_file():
|
|
raise RuntimeError(f"{name} has no resolved manifest for its walk audit")
|
|
write_walk_audit(spec, plan, audit_path)
|
|
output = run_step(f"walk-{name}", headless_command(binary, fixture, camera,
|
|
mesh, manifest, layout, 12, 1, audit_path=audit_path),
|
|
out / "logs" / f"walk-{name}.log", env=env, stamped=True,
|
|
expected_exit=expected_exit)
|
|
expected_status = "FAIL" if expected_exit else "PASS"
|
|
if not re.search(rf"\[LiminalStampWalkAudit\].*status={expected_status}\b", output):
|
|
raise RuntimeError(f"{name} walk audit did not report expected {expected_status}")
|
|
return output
|
|
|
|
|
|
def capture(name: str, fixture: Path, view: dict, floor: float, *, kind: str,
|
|
binary: Path, renderer: Path, out: Path, artifact_dir: Path,
|
|
env: dict[str, str]) -> None:
|
|
camera = camera_for(view, floor, name, kind)
|
|
target = target_for(view, floor, name, kind)
|
|
clip_radius = int(view[f"{kind}_clip_radius"])
|
|
max_level = 1 if name == "terraced-hills" and kind == "wide" else 0
|
|
stem = f"{name}-{kind}"
|
|
run_dir = out / "captures" / stem
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
mesh, manifest, layout = (run_dir / "capture.vfmesh",
|
|
run_dir / "manifest.json", run_dir / "layout.json")
|
|
png = run_dir / "capture.png"
|
|
image = artifact_dir / f"{stem}.jpg"
|
|
output = run_step(f"capture-export-{stem}", headless_command(binary,
|
|
fixture, camera, mesh, manifest, layout, 12, clip_radius,
|
|
max_level=max_level), out / "logs" / f"capture-export-{stem}.log",
|
|
env=env, stamped=True)
|
|
if not mesh.is_file() or mesh.stat().st_size == 0:
|
|
raise RuntimeError(f"{stem} produced an empty mesh")
|
|
yaw, pitch = camera_angles(camera, target)
|
|
if name == "backrooms-grid":
|
|
fog_start, fog_end, fog_color = 90, 500, "0.82,0.83,0.76"
|
|
elif name == "pool-rooms":
|
|
fog_start, fog_end, fog_color = 90, 480, "0.76,0.84,0.87"
|
|
elif name == "staircase-to-nowhere":
|
|
fog_start, fog_end, fog_color = 110, 700, "0.78,0.84,0.87"
|
|
else:
|
|
fog_start, fog_end, fog_color = 160, 1050, "0.78,0.84,0.89"
|
|
renderer_command = [renderer, mesh, png, "--size=1600x900", "--fov-deg=72",
|
|
f"--camera-m={fmt(camera)}", f"--yaw-deg={yaw:.6f}",
|
|
f"--pitch-deg={pitch:.6f}", "--tile-skirts=on", "--shade=readable",
|
|
"--headlamp=off", "--height-tint=on",
|
|
f"--height-center-m={floor + 3.0:.6f}", "--height-range-m=30.0",
|
|
"--light-direction=-0.42,-0.58,0.70", "--fog=on",
|
|
f"--fog-start-m={fog_start}", f"--fog-end-m={fog_end}",
|
|
f"--fog-color={fog_color}"]
|
|
run_step(f"capture-render-{stem}", renderer_command,
|
|
out / "logs" / f"capture-render-{stem}.log", env=env, stamped=True)
|
|
run_step(f"capture-jpeg-{stem}", ["ffmpeg", "-nostdin", "-hide_banner",
|
|
"-loglevel", "error", "-y", "-i", png, "-frames:v", "1", "-q:v", "4", image],
|
|
out / "logs" / f"capture-jpeg-{stem}.log", env=env)
|
|
quality = run_step(f"capture-quality-{stem}", [sys.executable,
|
|
ROOT / "Standalone/Tools/verify/capture_quality.py", image],
|
|
out / "logs" / f"capture-quality-{stem}.log", env=env)
|
|
if "[CaptureQuality]" not in quality or not image.is_file() or image.stat().st_size > 300_000:
|
|
raise RuntimeError(f"{stem} failed image quality or size checks")
|
|
print(f"[L161Capture] fixture={name} view={kind} camera_m=({fmt(camera)}) "
|
|
f"target_m=({fmt(target)}) bytes={image.stat().st_size} "
|
|
f"sha256={sha256(image)} status=PASS", flush=True)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--out", type=Path, default=OUT_DEFAULT)
|
|
parser.add_argument("--artifact-dir", type=Path, default=CAPTURE_DEFAULT)
|
|
parser.add_argument("--headless-clang", type=Path, required=True)
|
|
parser.add_argument("--headless-gcc", type=Path, required=True)
|
|
parser.add_argument("--renderer", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
out, artifact_dir = args.out.resolve(), args.artifact_dir.resolve()
|
|
fixture_dir = out / "fixtures"
|
|
for directory in (out, fixture_dir, artifact_dir):
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
env = os.environ.copy()
|
|
env["VOXELFORGE_VERIFY_OUT"] = str(out)
|
|
env["VOXELFORGE_OWNER_DATA"] = str(ROOT / "Standalone/data/owner_world.json")
|
|
env["LP_NUM_THREADS"] = "12"
|
|
env["XDG_CACHE_HOME"] = str(out / "xdg-cache")
|
|
run_step("build-fixtures", [sys.executable,
|
|
Path(__file__).resolve().parent / "BuildL161Fixtures.py",
|
|
f"--out={fixture_dir}"], out / "logs" / "build-fixtures.log", env=env)
|
|
|
|
binaries = {"clang": args.headless_clang.resolve(), "gcc": args.headless_gcc.resolve()}
|
|
fixtures = {name: fixture_dir / f"liminal-{name}.json" for name in
|
|
("pool-rooms", "backrooms-grid", "staircase-to-nowhere", "terraced-hills")}
|
|
views = {}
|
|
floors = {}
|
|
for name, fixture_path in fixtures.items():
|
|
view = json.loads((fixture_dir / "views" / f"{name}.json").read_text(encoding="utf-8"))
|
|
manifest_path, _, manifest, layout = build_plan(binaries["clang"], name,
|
|
fixture_path, view, out, env)
|
|
view["center_xy_m"] = view["center_xy_m"]
|
|
floors[name] = manifest["recipe"]["surface_profile"]["h0_q"] / Q_PER_METRE
|
|
if layout["stamp_vocabulary_version"] != 7 or not layout["stamps"]:
|
|
raise RuntimeError(f"{name} plan lacks resolved vocabulary-7 stamps")
|
|
write_walk_audit(fixture_dir / "walk-specs" / f"{name}.json",
|
|
manifest_path, out / "walk-audits" / name / "walk.path")
|
|
views[name] = view
|
|
print(f"[L161Plan] fixture={name} stamps={len(layout['stamps'])} "
|
|
f"floor_m={floors[name]:.3f} layout={layout['stable_identity']} status=PASS")
|
|
|
|
for name, fixture_path in fixtures.items():
|
|
run_walk_audit(name, fixture_path, views[name], floors[name],
|
|
fixture_dir / "walk-specs" / f"{name}.json", binaries["clang"], out, env)
|
|
run_determinism(name, fixture_path, views[name], floors[name], binaries, out, env)
|
|
|
|
narrow_name = "backrooms-narrow-control"
|
|
narrow_fixture = fixture_dir / f"liminal-{narrow_name}.json"
|
|
narrow_view = json.loads((fixture_dir / "views" / f"{narrow_name}.json").read_text(encoding="utf-8"))
|
|
narrow_manifest, _, narrow_manifest_value, _ = build_plan(binaries["clang"],
|
|
narrow_name, narrow_fixture, narrow_view, out, env)
|
|
narrow_floor = narrow_manifest_value["recipe"]["surface_profile"]["h0_q"] / Q_PER_METRE
|
|
run_walk_audit(narrow_name, narrow_fixture, narrow_view, narrow_floor,
|
|
fixture_dir / "walk-specs" / f"{narrow_name}.json", binaries["clang"],
|
|
out, env, expected_exit=14)
|
|
print("[L161DoorwayNegativeControl] width_m=1.200 minimum_m=1.500 expected=FAIL status=PASS")
|
|
|
|
for name, fixture_path in fixtures.items():
|
|
for kind in ("eye", "wide"):
|
|
capture(name, fixture_path, views[name], floors[name], kind=kind,
|
|
binary=binaries["clang"], renderer=args.renderer.resolve(), out=out,
|
|
artifact_dir=artifact_dir, env=env)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except Exception as error:
|
|
print(f"[L161Capture] status=FAIL error={error}", file=sys.stderr, flush=True)
|
|
raise SystemExit(1)
|