VoxelForge/Standalone/Research/WorldAssemblerM3/Linux54/CaptureNativeCavesL54.py

#!/usr/bin/env python3
"""Export and capture the L54 native cave spine and chamber from reviewed views."""
from __future__ import annotations

import argparse
import json
import math
import os
from pathlib import Path
import struct
import subprocess
import sys
import time


ROOT = Path(__file__).resolve().parents[4]
LIMIT = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--"]
Q = 1_000_000.0

sys.path.insert(0, str(ROOT / "Standalone/Tools/verify"))
from runner import bounded_run, current_stamp, first_stamp


def bezier(points: list[list[float]], t: float) -> tuple[float, float, float]:
    one = 1.0 - t
    weights = (one ** 3, 3.0 * one * one * t, 3.0 * one * t * t, t ** 3)
    return tuple(sum(weights[i] * points[i][axis] for i in range(4)) for axis in range(3))


def tangent(points: list[list[float]], t: float) -> tuple[float, float, float]:
    one = 1.0 - t
    return tuple(3.0 * one ** 2 * (points[1][axis] - points[0][axis])
        + 6.0 * one * t * (points[2][axis] - points[1][axis])
        + 3.0 * t ** 2 * (points[3][axis] - points[2][axis]) for axis in range(3))


def camera_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 fmt(values: tuple[float, ...]) -> str:
    return ",".join(f"{value:.6f}" for value in values)


def filter_mesh_y_leq(source: Path, destination: Path, cut_y: float) -> int:
    """Keep triangles on the negative-Y side for an uncapped, inspectable cutaway."""
    kept = 0
    with source.open("rb") as src, destination.open("wb") as dst:
        prefix = src.read(60)
        if len(prefix) != 60 or prefix[:8] != b"VFMSH001":
            raise ValueError("cutaway requires a VFMSH001 Core mesh")
        version, voxel_um, tile_count = struct.unpack_from("<3I", prefix, 8)
        if version != 1 or voxel_um <= 0:
            raise ValueError("cutaway mesh version or voxel size is invalid")
        dst.write(prefix)
        for _ in range(tile_count):
            header = src.read(36)
            if len(header) != 36:
                raise ValueError("cutaway mesh tile header is truncated")
            key = struct.unpack_from("<4i", header, 0)
            surface_count, _skirt_count, vertex_count, index_count, mask_count = \
                struct.unpack_from("<5I", header, 16)
            positions = src.read(vertex_count * 12)
            normals = src.read(vertex_count * 12)
            index_bytes = src.read(index_count * 4)
            masks = src.read(mask_count)
            if (len(positions) != vertex_count * 12
                    or len(normals) != vertex_count * 12
                    or len(index_bytes) != index_count * 4
                    or len(masks) != mask_count):
                raise ValueError("cutaway mesh tile payload is truncated")
            coords = list(struct.iter_unpack("<3f", positions))
            indices = struct.unpack(f"<{index_count}I", index_bytes) if index_count else ()
            kept_indices = []
            kept_masks = bytearray()
            for triangle in range(index_count // 3):
                ids = indices[triangle * 3:triangle * 3 + 3]
                if max(ids) >= vertex_count:
                    raise ValueError("cutaway triangle references an invalid vertex")
                if sum(coords[index][1] for index in ids) / 3.0 > cut_y:
                    continue
                kept_indices.extend(ids)
                kept += 1
                if masks:
                    kept_masks.append(masks[triangle])
            skirt_count = sum(kept_masks)
            output_header = struct.pack("<4i5I", *key, surface_count, skirt_count,
                vertex_count, len(kept_indices), len(kept_masks))
            dst.write(output_header)
            dst.write(positions)
            dst.write(normals)
            if kept_indices:
                dst.write(struct.pack(f"<{len(kept_indices)}I", *kept_indices))
            dst.write(kept_masks)
        if src.read(1):
            raise ValueError("cutaway mesh contains trailing bytes")
    if kept == 0:
        destination.unlink(missing_ok=True)
        raise ValueError("cutaway plane retained no triangles")
    return kept


def run_step(name: str, command: list[str], log: Path, *, env: dict[str, str],
             stamped: bool = False) -> str:
    expected = current_stamp() if stamped else None
    if stamped:
        binary = Path(command[0])
        probe = bounded_run([str(binary), "--print-source-stamp"], timeout=15)
        if probe is None or probe.returncode != 0 or first_stamp(probe.stdout or "") != expected:
            raise RuntimeError(f"{name} refused an unstamped or stale executable: {binary}")
    started = time.monotonic()
    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)
    elapsed = time.monotonic() - started
    output = log.read_text(encoding="utf-8", errors="replace")
    if stamped and (first_stamp(output) != expected or current_stamp() != expected):
        raise RuntimeError(f"{name} source stamp mismatch; see {log}")
    print(f"[L54CavesCaptureRun] step={name} elapsed={elapsed:.1f}s "
          f"exit={result.returncode} log={log}", flush=True)
    if result.returncode != 0:
        raise RuntimeError(f"{name} failed; see {log}")
    return output


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--out", type=Path, required=True)
    parser.add_argument("--fixture", type=Path, required=True)
    parser.add_argument("--layout", type=Path, required=True)
    parser.add_argument("--headless", type=Path, required=True)
    parser.add_argument("--renderer", type=Path, required=True)
    parser.add_argument("--artifact-dir", type=Path, default=Path(__file__).resolve().parent / "captures")
    args = parser.parse_args()
    out, artifact_dir = args.out.resolve(), args.artifact_dir.resolve()
    capture_dir, mesh_dir = out / "capture-png", out / "caves-meshes"
    for directory in (capture_dir, mesh_dir, artifact_dir):
        directory.mkdir(parents=True, exist_ok=True)
    layout = json.loads(args.layout.read_text(encoding="utf-8"))
    edges = sorted(layout.get("cavern_edges", []), key=lambda row: row["stable_id"])
    rooms = layout.get("cavern_rooms", [])
    if len(edges) != 2 or len(rooms) != 1:
        raise RuntimeError("expected the resolved one-room, two-edge cave spine")
    controls = [[value / Q for value in point] for point in edges[0]["control_points_um"]]
    room = rooms[0]
    room_center = tuple(value / Q for value in room["center_um"])
    room_floor = room["floor_um"] / Q
    mesh = mesh_dir / "caves-whole-view.vfmesh"
    mesh_layout = out / "layouts" / "caves-capture-layout.json"
    env = os.environ.copy()
    env["LP_NUM_THREADS"] = "12"
    env["XDG_CACHE_HOME"] = str(out / "xdg-cache")
    run_step("whole-cave-export", [args.headless, args.fixture,
        "--workers=12", "--world-assembler-prototype", "--stream-benchmark",
        "--stream-pipeline", "--force-generate", "--clip-radius=4",
        "--max-clip-level=0", "--max-level=0", "--render-distance-chunks=0",
        "--recipe=caves", "--camera=41,-4,-160,0,0",
        f"--resolved-layout-output={mesh_layout}", f"--mesh-export={mesh}"],
        out / "capture-caves-export.log", env=env, stamped=True)
    if not mesh.is_file() or mesh.stat().st_size == 0:
        raise RuntimeError("whole-cave export produced no mesh")
    cutaway_mesh = mesh_dir / "caves-cutaway-y-minus5.vfmesh"
    cutaway_triangles = filter_mesh_y_leq(mesh, cutaway_mesh, -5.0)
    print(f"[L54CavesCutaway] plane_y_m=-5.000 retained_triangles={cutaway_triangles} "
          f"mesh={cutaway_mesh}", flush=True)

    eye_t = 0.50
    eye_floor = bezier(controls, eye_t)
    eye_direction = tangent(controls, eye_t)
    eye = (eye_floor[0], eye_floor[1], eye_floor[2] + 1.45)
    eye_yaw = math.degrees(math.atan2(eye_direction[0], eye_direction[1]))
    eye_target = (eye[0] + 10.0 * math.sin(math.radians(eye_yaw)),
                  eye[1] + 10.0 * math.cos(math.radians(eye_yaw)), eye[2])

    arrival_portal = [value / Q for value in edges[0]["control_points_um"][-1]]
    entrance = (arrival_portal[0], arrival_portal[1], room_floor + 1.45)
    entrance_target = (room_center[0], room_center[1], room_floor + 1.45)

    overview_target = (room_center[0] - 8.0, room_center[1], room_floor + 1.5)
    overview = (room_center[0] - 23.0, room_center[1] + 27.0, room_floor + 19.0)

    views = (
        ("spine-eye", cutaway_mesh, eye, eye_target, 20.0, 80.0),
        ("room-from-entrance", mesh, entrance, entrance_target, 18.0, 90.0),
        ("cutaway-overview", cutaway_mesh, overview, overview_target, 48.0, 65.0),
    )
    quality = ROOT / "Standalone/Tools/verify/capture_quality.py"
    for name, render_mesh, camera, target, focus_radius, fov in views:
        yaw, pitch = camera_angles(camera, target)
        if name == "spine-eye":
            yaw, pitch = eye_yaw, 0.0
        png = capture_dir / f"caves-{name}.png"
        image = artifact_dir / f"caves-{name}.jpg"
        command = [args.renderer, render_mesh, png, "--size=1600x900",
            f"--fov-deg={fov:.1f}",
            f"--camera-m={fmt(camera)}", f"--yaw-deg={yaw:.6f}",
            f"--pitch-deg={pitch - 4.0 if name == 'spine-eye' else pitch:.6f}",
            "--tile-skirts=off", "--shade=readable", "--headlamp=off",
            "--light-direction=1,-0.6,0.8", "--height-tint=on",
            f"--height-center-m={room_floor:.6f}",
            "--height-range-m=12", "--tile-edges=off",
            f"--focus-region={fmt((*target, focus_radius))}"]
        run_step(f"{name}-render", command, out / f"capture-{name}-render.log", env=env,
                 stamped=True)
        run_step(f"{name}-jpeg", ["ffmpeg", "-nostdin", "-hide_banner", "-loglevel",
            "error", "-y", "-i", png, "-frames:v", "1", "-q:v", "4", image],
            out / f"capture-{name}-jpeg.log", env=env)
        quality_output = run_step(f"{name}-quality", [sys.executable, quality, image],
            out / f"capture-{name}-quality.log", env=env)
        quality_lines = [line for line in quality_output.splitlines()
                         if line.startswith("[CaptureQuality]")]
        if not quality_lines or not quality_lines[-1].endswith("status=PASS"):
            raise RuntimeError(f"capture quality gate failed for {name}")
        print(f"[L54CavesCapture] view={name} file={image} camera_m=({fmt(camera)}) "
              f"target_m=({fmt(target)}) bytes={image.stat().st_size} status=PASS", flush=True)
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as error:
        print(f"[L54CavesCapture] status=FAIL error={error}", file=sys.stderr, flush=True)
        raise SystemExit(1)