VoxelForge/Standalone/Research/WorldAssemblerM3/Linux67/CaptureOwnerPassagesL67.py

#!/usr/bin/env python3
"""Render arrival, departure, and mid-tunnel views for the four-strate owner plan."""
import json
import importlib.util
import argparse
import math
import os
import re
import subprocess
import time
from pathlib import Path


ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l67/captures-run"
BIN = ROOT / "Saved/Linux/l67/bin-final"
CAPTURES = Path(__file__).resolve().parent / "captures"
HEADLESS = BIN / "VoxelForgeCoreHeadless-l67-final-clang"
RENDERER = BIN / "VoxelForgeOffscreenCapture-l67-final-clang"
PNG_DIR = OUT / "capture-png"
RECIPE_INPUTS = (
    ("passage-0-1", ROOT / "Standalone/data/owner_world.json",
     "surface-world", "SurfaceWorld", "upper"),
    ("passage-1-2", ROOT / "Standalone/data/owner_world.json",
     "surface-world", "FloatingIslands", "upper"),
    ("passage-2-3", ROOT / "Standalone/data/owner_world.json",
     "surface-world", "TunnelNetwork", "upper"),
)
SUBVOXELS_PER_VOXEL = 256.0
RING_AUDIT = ROOT / "Standalone/Research/WorldAssemblerM3/Linux32/AuditRingTerraceL32.py"
spec = importlib.util.spec_from_file_location("ring_profile_l32", RING_AUDIT)
ring_profile = importlib.util.module_from_spec(spec)
spec.loader.exec_module(ring_profile)

for directory in (OUT, BIN, PNG_DIR, CAPTURES):
    directory.mkdir(parents=True, exist_ok=True)
ENV = os.environ.copy()
ENV["LP_NUM_THREADS"] = "12"
ENV["XDG_CACHE_HOME"] = str(OUT / "xdg-cache")
PARSER = argparse.ArgumentParser(description=__doc__)
PARSER.add_argument("--scene", choices=[value[0] for value in RECIPE_INPUTS])
PARSER.add_argument("--view", choices=("arrival-mouth", "departure-mouth", "mid-tunnel"))
OPTIONS = PARSER.parse_args()


def run(name, command):
    log = OUT / f"capture-{name}.log"
    wrapped = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
               "timeout", "1800s", *map(str, command)]
    started = time.monotonic()
    with log.open("w", encoding="utf-8") as stream:
        result = subprocess.run(wrapped, cwd=ROOT, env=ENV, stdout=stream,
                                stderr=subprocess.STDOUT, timeout=1800, check=False)
    elapsed = time.monotonic() - started
    print(f"[L35CaptureRun] step={name} elapsed={elapsed:.1f}s exit={result.returncode} log={log}",
          flush=True)
    if result.returncode != 0:
        raise SystemExit(f"{name} failed; see {log}")
    return log


def add(a, b):
    return tuple(x + y for x, y in zip(a, b))


def sub(a, b):
    return tuple(x - y for x, y in zip(a, b))


def mul(a, scale):
    return tuple(x * scale for x in a)


def norm(a):
    length = math.sqrt(sum(x * x for x in a))
    if length <= 1.0e-9:
        raise ValueError("passage has a degenerate camera tangent")
    return tuple(x / length for x in a)


def interpolate_route(points, distance):
    remaining = max(0.0, distance)
    for a, b in zip(points, points[1:]):
        delta = sub(b[0], a[0])
        length = math.sqrt(sum(x * x for x in delta))
        if length <= 1.0e-9:
            continue
        if remaining <= length:
            t = remaining / length
            pos = tuple(a[0][axis] + delta[axis] * t for axis in range(3))
            radius = a[1] + (b[1] - a[1]) * t
            return pos, radius
        remaining -= length
    return points[-1]


def route_frame(points, distance):
    center, radius = interpolate_route(points, distance)
    before, _ = interpolate_route(points, max(0.0, distance - 0.25))
    after, _ = interpolate_route(points, distance + 0.25)
    tangent = norm(sub(after, before))
    horizontal = norm((tangent[0], tangent[1], 0.0))
    side = (-horizontal[1], horizontal[0], 0.0)
    return center, radius, tangent, side


def camera_angles(camera, target):
    delta = sub(target, camera)
    horizontal = math.hypot(delta[0], delta[1])
    yaw = math.degrees(math.atan2(delta[0], delta[1]))
    pitch = math.degrees(math.atan2(delta[2], horizontal))
    return yaw, pitch


def fmt(values):
    return ",".join(f"{value:.6f}" for value in values)


def load_reference_cameras():
    log = ROOT / "Saved/Linux/l34/run-capture-standard-set-final-radius3.log"
    pattern = re.compile(
        r"^\[L34CaptureCamera\] view=(\S+).*?camera_m=\(([^)]*)\) "
        r"target_m=\(([^)]*)\) clip=(\S+)", re.M)
    values = {}
    for name, camera, target, clip in pattern.findall(
            log.read_text(encoding="utf-8", errors="replace")):
        parse = lambda value: tuple(float(part) for part in value.split(","))
        values[name] = (parse(camera), parse(target), None if clip == "none" else parse(clip))
    return values


REFERENCE_CAMERAS = {}


def terrain_surface_near(mesh, x, y, expected_height):
    bounds = (x - 2.0, x + 2.0, y - 2.0, y + 2.0)
    _, triangles, _ = ring_profile.read_scene(
        str(mesh), None, -1.0e9, 1.0e9, bounds)
    buckets = ring_profile.build_buckets(triangles, True, bounds, cell_size=1.0)
    hits = []
    for dy in range(-4, 5):
        for dx in range(-4, 5):
            sample_x, sample_y = x + dx * 0.5, y + dy * 0.5
            hit = ring_profile.sample_surface(buckets, sample_x, sample_y, True, 1.0,
                                              preferred_height=expected_height)
            if hit is not None:
                hits.append((math.hypot(dx * 0.5, dy * 0.5),
                             abs(hit[0] - expected_height), sample_x, sample_y, hit[0]))
    if not hits:
        return None
    distance, _, sample_x, sample_y, height = min(hits)
    if distance > 2.0 or abs(height - expected_height) > 8.0:
        return None
    return sample_x, sample_y, height


def load_layout(name, source, recipe):
    path = OUT / f"layout-{name}.json"
    run(f"layout-{name}", [HEADLESS, source, "--workers=12", "--world-assembler-prototype",
        "--stream-benchmark", "--stream-pipeline", "--force-generate", "--clip-radius=2",
        "--max-clip-level=0", "--max-level=0", f"--recipe={recipe}",
        f"--resolved-layout-output={path}", "--resolved-layout-only"])
    return json.loads(path.read_text(encoding="utf-8"))


def select_passage(layout, mouth_generator, preferred_side):
    strata = {int(value["strate_index"]): value for value in layout["strates"]}
    preferred_candidates = []
    fallback_candidates = []
    for passage in layout["passages"]:
        upper = strata[int(passage["upper_strate_index"])]
        lower = strata[int(passage["lower_strate_index"])]
        sides = []
        if upper["generator_kind"] == mouth_generator:
            sides.append("upper")
        if lower["generator_kind"] == mouth_generator:
            sides.append("lower")
        if preferred_side in sides:
            preferred_candidates.append((passage, preferred_side == "lower"))
        elif sides:
            fallback_candidates.append((passage, sides[0] == "lower"))
    candidates = preferred_candidates or fallback_candidates
    if not candidates:
        raise RuntimeError(f"no passage mouth in generator {mouth_generator}")
    return sorted(candidates, key=lambda item: item[0]["stable_id"])[0]


def make_views(layout, passage, reverse):
    q_to_m = layout["core_voxel_size_um"] / (SUBVOXELS_PER_VOXEL * 1_000_000.0)
    controls = [
        (tuple(value * q_to_m for value in point["position_q"]),
         point["radius_q"] * q_to_m)
        for point in passage["control_points"]
    ]
    lower_mouth = tuple(value * q_to_m for value in passage["lower_mouth_q"])
    lower_index = min(range(len(controls)), key=lambda i: sum(
        (controls[i][0][axis] - lower_mouth[axis]) ** 2 for axis in range(3)))
    path = controls[:lower_index + 1]
    if len(path) < 2:
        raise RuntimeError(f"{passage['stable_id']} has no mouth-to-mouth route")
    if reverse:
        path.reverse()
    lengths = [math.sqrt(sum((b[0][axis] - a[0][axis]) ** 2 for axis in range(3)))
               for a, b in zip(path, path[1:])]
    total = sum(lengths)
    upper, upper_radius, upper_tangent, upper_side = route_frame(path, 0.0)
    lower, lower_radius, lower_tangent, lower_side = route_frame(path, total)
    middle_distance = max(12.0, min(total - 12.0, total * 0.4)) if total > 24.0 else total * 0.4
    entry_distance = min(6.0, total * 0.25)
    entry, entry_radius, entry_tangent, entry_side = route_frame(path, entry_distance)
    middle, middle_radius, middle_tangent, middle_side = route_frame(path, middle_distance)

    def exterior_view(prefix, center, radius, outward, target):
        horizontal_outward = norm((outward[0], outward[1], 0.0))
        eye_height = center[2] - radius + 1.6
        eye = (center[0] + horizontal_outward[0] * 5.5,
               center[1] + horizontal_outward[1] * 5.5,
               eye_height)
        yaw, pitch = camera_angles(eye, target)
        return {"name": prefix, "camera": eye, "yaw": yaw, "pitch": pitch,
                "target": target, "clip": None, "camera_xy": eye[:2],
                "focus": center, "mouth": center, "eye_height_z": eye_height}

    entrance, entrance_radius, entrance_tangent = upper, upper_radius, upper_tangent
    exit_point, exit_radius, exit_tangent = lower, lower_radius, lower_tangent
    ground_target_pos, ground_target_radius = interpolate_route(path, min(total, 1.0))
    ground = exterior_view("ground-mouth", entrance, entrance_radius,
        mul(entrance_tangent, -1.0), (ground_target_pos[0], ground_target_pos[1],
            ground_target_pos[2] - ground_target_radius + 1.0))
    ground["focus"] = entrance
    ground["outside_camera"] = True

    def cutaway_view(prefix, center, radius, tangent, side, distance, target_distance):
        camera = (center[0] - side[0] * 7.0, center[1] - side[1] * 7.0,
                  center[2] - radius + 1.6)
        target_pos, target_radius = interpolate_route(path, target_distance)
        target = (target_pos[0], target_pos[1],
                  target_pos[2] - target_radius + 1.6)
        yaw, pitch = camera_angles(camera, target)
        plane = (side[0], side[1], 0.0,
                 -(side[0] * center[0] + side[1] * center[1]))
        return {"name": prefix, "camera": camera, "yaw": yaw, "pitch": pitch,
                "target": target, "clip": plane, "camera_xy": camera[:2],
                "focus": center, "mouth": center}

    # Place each interface view at its true route endpoint. The endpoint camera
    # is a side cutaway so the solid terrain in front of an underground mouth
    # cannot hide the floor-to-tunnel connection.
    arrival_distance = 0.0
    arrival_center, arrival_radius, arrival_tangent, arrival_side = route_frame(
        path, arrival_distance)
    arrival = cutaway_view("arrival", arrival_center, arrival_radius,
        arrival_tangent, arrival_side, arrival_distance,
        min(total, arrival_distance + 12.0))
    arrival["camera_location"] = "passage-air"
    arrival["focus_radius"] = 20.0
    arrival["yaw"], arrival["pitch"] = camera_angles(
        arrival["camera"], arrival["target"])

    entry_camera_distance = min(8.0, total * 0.1)
    entry_center, entry_radius, entry_tangent, entry_side = route_frame(
        path, entry_camera_distance)
    # Player-height view from just inside the passage, aimed through its actual
    # interface mouth so the floor, bore, and opening share one frame.
    entry_camera = (entry_center[0], entry_center[1],
        entry_center[2] - entry_radius + 1.6)
    entry_target_pos, entry_target_radius = interpolate_route(
        path, min(total, entry_distance + 14.0))
    entry_target = (entry_target_pos[0], entry_target_pos[1],
        entry_camera[2] - 2.5)
    entry_yaw, entry_pitch = camera_angles(entry_camera, entry_target)
    entry_view = {"name": "rock-entry", "camera": entry_camera,
        "yaw": entry_yaw, "pitch": entry_pitch, "target": entry_target,
        "clip": None, "camera_xy": entry_camera[:2],
        "focus": entry_center, "mouth": entrance, "outside_camera": False,
        "camera_location": "passage-air", "focus_radius": 20.0,
        "eye_height_z": entry_camera[2]}
    middle_view = cutaway_view("mid-tunnel", middle, middle_radius, middle_tangent,
        middle_side, middle_distance, min(total, middle_distance + 2.0))
    ground["focus_radius"] = 20.0
    middle_view["focus_radius"] = 20.0
    middle_camera, _ = interpolate_route(path, max(0.0, middle_distance - 6.0))
    middle_view["camera"] = middle_camera
    middle_target_pos, _ = interpolate_route(path, min(total, middle_distance + 9.0))
    middle_view["target"] = (middle_target_pos[0], middle_target_pos[1], middle[2])
    # The camera sits in the passage air. Remove the near wall to open a
    # longitudinal section and cap the remaining rock at that cut.
    middle_cut_point = (middle[0] - middle_side[0] * 2.5,
        middle[1] - middle_side[1] * 2.5, middle[2])
    middle_view["clip"] = (*middle_side,
        -sum(middle_side[axis] * middle_cut_point[axis] for axis in range(3)))
    middle_view["camera_location"] = "passage-air"
    middle_view["yaw"], middle_view["pitch"] = camera_angles(
        middle_view["camera"], middle_view["target"])
    return (ground, entry_view, middle_view, arrival), total, lower_index


def export_capture(scene_name, source, recipe, passage, endpoint, view):
    stem = f"{passage['stable_id'].replace('/', '-')}-{endpoint}"
    view["name"] = stem
    mesh = OUT / f"{stem}.vfmesh"
    png = PNG_DIR / f"{stem}.png"
    jpg = CAPTURES / f"{stem}.jpg"
    yaw, pitch = view["yaw"], view["pitch"]
    camera = view["camera"]
    export_camera = view.get("focus", camera)
    run(f"{stem}-export", [HEADLESS, source, "--workers=12",
        "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
        "--force-generate", "--clip-radius=3", "--max-clip-level=0", "--max-level=0",
        "--render-distance-chunks=0", f"--recipe={recipe}",
        f"--camera={fmt((*export_camera, yaw, pitch))}", f"--mesh-export={mesh}"])
    render_clip = view["clip"]
    surface = None
    if view.get("outside_camera", False):
        surface = terrain_surface_near(mesh, *view["camera_xy"], view["target"][2])
        if surface is None:
            camera = (view["camera_xy"][0], view["camera_xy"][1], view["eye_height_z"])
        else:
            camera = (surface[0], surface[1], surface[2] + 1.6)
    elif render_clip is not None:
        camera = view["camera"]
    yaw, pitch = camera_angles(camera, view["target"])
    print(f"[L67CaptureCamera] view={stem} camera_m=({fmt(camera)}) "
        f"target_m=({fmt(view['target'])}) "
        f"clip={fmt(render_clip) if render_clip is not None else 'none'} fov_deg=80",
        flush=True)
    render = [RENDERER, mesh, png, "--size=1600x900", "--fov-deg=80",
        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",
        f"--height-center-m={view['target'][2]:.6f}", "--height-range-m=8", "--tile-edges=off"]
    if view.get("focus_radius", 24.0) is not None:
        render.append(f"--focus-region={fmt((*view['focus'], view.get('focus_radius', 24.0)))}")
    if render_clip is not None:
        render.append(f"--clip-plane={fmt(render_clip)}")
    run(f"{stem}-render", render)
    run(f"{stem}-jpeg", ["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error",
        "-y", "-i", png, "-frames:v", "1", "-q:v", "4", jpg])
    size = jpg.stat().st_size if jpg.exists() else 0
    print(f"[L67CaptureJPEG] file={jpg.relative_to(ROOT)} bytes={size} "
        f"status={'PASS' if 0 < size <= 300_000 else 'FAIL'}", flush=True)
    if not 0 < size <= 300_000:
        raise SystemExit(f"{jpg.name} is missing or exceeds 300 KB")


for scene_name, source, recipe, mouth_generator, preferred_side in RECIPE_INPUTS:
    if OPTIONS.scene and scene_name != OPTIONS.scene:
        continue
    layout = load_layout(scene_name, source, recipe)
    passage, _ = select_passage(layout, mouth_generator, preferred_side)
    forward_views, route_length, lower_index = make_views(layout, passage, False)
    reverse_views, _, _ = make_views(layout, passage, True)
    print(f"[L67CapturePassage] interface={scene_name.removeprefix('passage-')} "
        f"id={passage['stable_id']} route_m={route_length:.3f} "
        f"control_points={lower_index + 1}", flush=True)
    endpoint_views = (
        ("arrival-mouth", forward_views[0] if scene_name == "passage-0-1" else forward_views[1]),
        ("departure-mouth", reverse_views[3]),
    )
    for endpoint, view in endpoint_views:
        if OPTIONS.view and endpoint != OPTIONS.view:
            continue
        export_capture(scene_name, source, recipe, passage, endpoint, view)
    if scene_name == "passage-1-2" and (OPTIONS.view is None or OPTIONS.view == "mid-tunnel"):
        export_capture(scene_name, source, recipe, passage, "mid-tunnel", forward_views[2])