#!/usr/bin/env python3 """Capture L109/L110 passage views, open-air shell close-ups, and image checks.""" from __future__ import annotations import json import math import re from array import array import struct import subprocess import time from pathlib import Path ROOT = Path(__file__).resolve().parents[4] RESEARCH = Path(__file__).resolve().parent CAPTURES = RESEARCH / "captures" SAVED = ROOT / "Saved/Linux/l110/captures" DATA = ROOT / "Standalone/data/owner_world.json" BINARIES = { "before": ROOT / "Saved/Linux/l109/bin-final/VoxelForgeCoreHeadless-l109-clang", "after": ROOT / "Saved/Linux/l110/bin-clang-v257/VoxelForgeCoreHeadless-l110-clang-v257", } LAYOUTS = { "before": ROOT / "Saved/Linux/l109/audit-layout-final-stamp.json", "after": ROOT / "Saved/Linux/l109/audit-layout-final-stamp.json", } AUDIT_LOG = ROOT / "Saved/Linux/l109/audit-owner4-final-stamp.log" RENDERERS = { "before": ROOT / "Saved/Linux/l110/bin-clang-v257/VoxelForgeOffscreenCapture-l110-clang-v257", "after": ROOT / "Saved/Linux/l110/bin-clang-v257/VoxelForgeOffscreenCapture-l110-clang-v257", } SECTION = re.compile( r"^\[PassageOpenAirSection\] id=(\S+) section=(\d+) start_m=([\d.]+) " r"end_m=([\d.]+) samples=(\d+) midpoint_xyz_m=\(([^)]*)\) " r"midpoint_radius_m=([\d.]+)", re.M) DRAWN = re.compile(r"drawn_triangles=(\d+)") QUALITY_METRICS = re.compile(r"stddev_luma=([\d.]+).*coverage_luma_gt12=([\d.]+)") def vector(text: str) -> tuple[float, float, float]: return tuple(float(value) for value in text.split(",")) 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: return (0.0, 1.0, 0.0) return tuple(value / length for value in a) def angles(camera, target): delta = sub(target, camera) horizontal = math.hypot(delta[0], delta[1]) return (math.degrees(math.atan2(delta[0], delta[1])), math.degrees(math.atan2(delta[2], horizontal))) def fmt(values) -> str: return ",".join(f"{value:.6f}" for value in values) def mesh_triangle_count(mesh: Path) -> int: """Read only the VFMSH headers and count indexed triangles.""" with mesh.open("rb") as stream: magic = stream.read(8) materials = magic == b"VFMSH002" if magic not in (b"VFMSH001", b"VFMSH002"): raise RuntimeError(f"unsupported mesh scene header: {mesh}") version, _voxel_um, tile_count = struct.unpack("<3I", stream.read(12)) if version != (2 if materials else 1): raise RuntimeError(f"mesh scene version mismatch: {mesh}") stream.seek(40, 1) triangles = 0 for _ in range(tile_count): stream.seek(16, 1) surface_count, _skirts, vertex_count, index_count, mask_count = struct.unpack( "<5I", stream.read(20)) triangles += index_count // 3 stream.seek(vertex_count * 24 + (10 * vertex_count if materials else 0) + index_count * 4 + mask_count, 1) if stream.read(1): raise RuntimeError(f"trailing data in mesh scene: {mesh}") return triangles def dot(a, b): return sum(x * y for x, y in zip(a, b)) def nearest_surface_anchor(mesh: Path, point, margin_m=200.0) -> dict: """Return a nearby VFMSH surface vertex and its normal.""" best_distance_sq = float("inf") best_position = best_normal = None with mesh.open("rb") as stream: magic = stream.read(8) materials = magic == b"VFMSH002" if magic not in (b"VFMSH001", b"VFMSH002"): raise RuntimeError(f"unsupported mesh scene header: {mesh}") version, voxel_um, tile_count = struct.unpack("<3I", stream.read(12)) if version != (2 if materials else 1): raise RuntimeError(f"mesh scene version mismatch: {mesh}") stream.seek(40, 1) for _ in range(tile_count): key = struct.unpack("<4i", stream.read(16)) surface_count, _skirts, vertex_count, index_count, mask_count = struct.unpack( "<5I", stream.read(20)) extent = (32 << key[3]) * voxel_um / 1_000_000.0 origin = (key[0] * extent, key[1] * extent, key[2] * extent) if any(point[axis] < origin[axis] - margin_m or point[axis] > origin[axis] + extent + margin_m for axis in range(3)): stream.seek(vertex_count * 24 + (10 * vertex_count if materials else 0) + index_count * 4 + mask_count, 1) continue positions = array("f") positions.frombytes(stream.read(vertex_count * 12)) normals = array("f") normals.frombytes(stream.read(vertex_count * 12)) if materials: stream.seek(vertex_count * 10, 1) stream.seek(index_count * 4 + mask_count, 1) if struct.pack("=I", 1) != struct.pack("= best_distance_sq: continue raw_normal = tuple(float(normals[offset + axis]) for axis in range(3)) if dot(raw_normal, raw_normal) < 1.0e-10: continue best_distance_sq = distance_sq best_position = position best_normal = norm(raw_normal) if best_position is None: raise RuntimeError(f"no nearby surface vertices in mesh scene: {mesh}") return {"position_m": best_position, "normal": best_normal, "distance_m": math.sqrt(best_distance_sq)} def sanitize(value: str) -> str: return re.sub(r"[^A-Za-z0-9_-]+", "-", value).strip("-").replace("/", "-") def route_for(layout: dict, passage: dict) -> list[tuple[tuple[float, float, float], float]]: scale = layout["core_voxel_size_um"] / 256.0 / 1_000_000.0 controls = [(tuple(value * scale for value in row["position_q"]), row["radius_q"] * scale) for row in passage["control_points"]] lower_q = passage["lower_mouth_q"] player_center_above_floor_q = round( 1.13 * 256.0 * 1_000_000.0 / layout["core_voxel_size_um"]) lower_index = next((index for index, row in enumerate(passage["control_points"]) if row["position_q"][0] == lower_q[0] and row["position_q"][1] == lower_q[1] and row["position_q"][2] - row["radius_q"] + player_center_above_floor_q == lower_q[2]), None) if lower_index is None: raise RuntimeError(f"{passage['stable_id']}: lower mouth control is unbound") result = controls[:lower_index + 1] if len(result) < 2: raise RuntimeError(f"{passage['stable_id']}: incomplete route controls") return result def route_lengths(route): lengths = [math.dist(a[0], b[0]) for a, b in zip(route, route[1:])] prefix = [0.0] for length in lengths: prefix.append(prefix[-1] + length) return lengths, prefix def at_distance(route, lengths, prefix, distance): distance = max(0.0, min(distance, prefix[-1])) for index, length in enumerate(lengths): if distance <= prefix[index + 1] or index + 1 == len(lengths): t = 0.0 if length <= 1.0e-9 else (distance - prefix[index]) / length a, b = route[index], route[index + 1] return (tuple(a[0][axis] + (b[0][axis] - a[0][axis]) * t for axis in range(3)), a[1] + (b[1] - a[1]) * t) return route[-1] def tangent_at(route, lengths, prefix, distance): before = at_distance(route, lengths, prefix, max(0.0, distance - 0.25))[0] after = at_distance(route, lengths, prefix, min(prefix[-1], distance + 0.25))[0] return norm(sub(after, before)) def derived_route(route, passage, layout): """Apply L109's fixed-depth floor offset to raw layout controls.""" lengths, prefix = route_lengths(route) total = prefix[-1] body_radii = [radius for index, (_, radius) in enumerate(route) if prefix[index] >= 10.0 and total - prefix[index] >= 10.0] body_depth = max(0.5, min(body_radii or [radius for _, radius in route]) - 0.23) scale = layout["core_voxel_size_um"] / 256.0 / 1_000_000.0 upper_z = passage["upper_mouth_q"][2] * scale lower_z = passage["lower_mouth_q"][2] * scale upper_target = max(0.0, route[0][0][2] - upper_z + 1.13) lower_target = max(0.0, route[-1][0][2] - lower_z + 1.13) transformed = [] for index, (point, radius) in enumerate(route): local_radius_depth = max(0.0, radius - 0.23) mouth_distance = max(0.0, min(prefix[index], total - prefix[index])) u = max(0.0, min(1.0, mouth_distance / 5.0)) ramp = u * u * (3.0 - 2.0 * u) target = upper_target if prefix[index] <= total * 0.5 else lower_target mouth_depth = max(0.0, min(radius, target)) route_depth = mouth_depth + (local_radius_depth - mouth_depth) * ramp tube_depth = min(body_depth, local_radius_depth) transformed.append(((point[0], point[1], point[2] - route_depth + tube_depth), radius)) return transformed, body_depth def make_views(layout: dict, audit_sections: list[dict], *, use_derived_shape: bool, include_open_air: bool = True) -> list[dict]: views = [] passages = sorted(layout["passages"], key=lambda item: item["stable_id"]) for passage in passages: raw_route = route_for(layout, passage) if use_derived_shape: route, body_depth = derived_route(raw_route, passage, layout) else: route = raw_route body_depth = max(point[1] for point in route) def floor_depth(radius: float) -> float: if not use_derived_shape: return radius return min(body_depth, max(0.0, radius - 0.23)) lengths, prefix = route_lengths(route) total = prefix[-1] passage_id = passage["stable_id"] safe_id = sanitize(passage_id) for end_name, distance, outward_sign in ( ("upper-mouth", 0.0, -1.0), ("lower-mouth", total, 1.0)): center, radius = at_distance(route, lengths, prefix, distance) tangent = tangent_at(route, lengths, prefix, distance) outward = mul(tangent, outward_sign) floor_z = center[2] - floor_depth(radius) eye = (center[0] + outward[0] * 4.5, center[1] + outward[1] * 4.5, floor_z + 1.55) if end_name == "lower-mouth": eye = add(eye, add(mul(outward, 5.0), (0.0, 0.0, 2.0))) target = (center[0] - outward[0] * 2.0, center[1] - outward[1] * 2.0, floor_z + 1.45) yaw, pitch = angles(eye, target) views.append({"name": f"{safe_id}-{end_name}", "passage": passage_id, "kind": "mouth", "station": end_name, "camera": eye, "target": target, "yaw": yaw, "pitch": pitch, "focus": center, "center_z": center[2]}) for fraction in (0.25, 0.50, 0.75): distance = total * fraction center, radius = at_distance(route, lengths, prefix, distance) tangent = tangent_at(route, lengths, prefix, distance) depth = floor_depth(radius) floor_z = center[2] - depth horizontal = norm((-tangent[1], tangent[0], 0.0)) eye = (center[0] + horizontal[0] * 0.75, center[1] + horizontal[1] * 0.75, floor_z + 1.60) target_center, target_radius = at_distance(route, lengths, prefix, min(total, distance + 14.0)) target_depth = floor_depth(target_radius) target = (target_center[0] + horizontal[0] * 2.35, target_center[1] + horizontal[1] * 2.35, target_center[2] - target_depth + 1.80) yaw, pitch = angles(eye, target) views.append({"name": f"{safe_id}-station-{int(fraction * 100)}", "passage": passage_id, "kind": "station", "station": f"{int(fraction * 100)}%", "camera": eye, "target": target, "yaw": yaw, "pitch": pitch, "focus": center, "center_z": center[2], "route_distance_m": distance, "route_length_m": total, "requested_route_distance_m": distance}) if not include_open_air: continue for section in (item for item in audit_sections if item["passage"] == passage_id): distance = 0.5 * (section["start_m"] + section["end_m"]) center, radius = (section["midpoint"], section["radius_m"]) tangent = tangent_at(route, lengths, prefix, distance) side = norm((-tangent[1], tangent[0], 0.0)) depth = floor_depth(radius) section_id = section["section"] interior_eye = (center[0], center[1], center[2] - depth + 1.25) interior_target = add(interior_eye, mul(tangent, 7.0)) side_eye = add(center, add(mul(side, radius + 5.0), (0.0, 0.0, 1.2))) oblique_eye = add(center, add(mul(side, -(radius + 3.5)), (0.0, 0.0, 4.0))) targets = (add(center, (0.0, 0.0, 0.1)), add(center, mul(tangent, 2.0)), interior_target) for view_index, (eye, target) in enumerate(zip( (side_eye, oblique_eye, interior_eye), targets), start=1): yaw, pitch = angles(eye, target) views.append({"name": f"{safe_id}-openair-{section_id}-view-{view_index}", "passage": passage_id, "kind": "open-air", "station": f"section-{section_id}/view-{view_index}", "camera": eye, "target": target, "yaw": yaw, "pitch": pitch, "focus": center, "center_z": center[2], "open_air_section": section_id, "route_distance_m": distance, "requested_route_distance_m": distance, "route_length_m": total}) shell_radius = radius + 0.75 closeup_directions = [side] if passage_id == "passage/0-1/0" and section_id == 4: closeup_directions.extend([ norm(add(mul(side, -1.0), (0.0, 0.0, 0.65))), norm(add(mul(side, 0.55), (0.0, 0.0, 0.84))), ]) for closeup_index, direction in enumerate(closeup_directions, start=1): shell_target = add(center, mul(direction, shell_radius)) closeup_camera = add(shell_target, mul(direction, 4.0)) yaw, pitch = angles(closeup_camera, shell_target) views.append({"name": f"{safe_id}-openair-{section_id}-closeup-" f"{closeup_index}", "passage": passage_id, "kind": "open-air-closeup", "station": f"section-{section_id}/exterior-{closeup_index}", "camera": closeup_camera, "target": shell_target, "yaw": yaw, "pitch": pitch, "focus": center, "center_z": center[2], "open_air_section": section_id, "route_distance_m": distance, "requested_route_distance_m": distance, "route_length_m": total, "shell_standoff_m": 4.0}) return views def run_process(name: str, command: list[str], log: Path, *, allow_failure=False) -> str: wrapped = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--", "timeout", "1740s", *map(str, command)] started = time.monotonic() with log.open("w", encoding="utf-8") as stream: result = subprocess.run(wrapped, cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1740) if result.returncode and not allow_failure: raise RuntimeError(f"{name}: exit={result.returncode} log={log}") print(f"[L110CaptureProcess] name={name} elapsed_s={time.monotonic()-started:.2f} " f"exit={result.returncode} log={log.relative_to(ROOT)}", flush=True) return log.read_text(encoding="utf-8", errors="replace") def capture(variant: str, view: dict, layout: dict) -> dict: variant_saved = SAVED / variant variant_saved.mkdir(parents=True, exist_ok=True) variant_capture = CAPTURES / variant variant_capture.mkdir(parents=True, exist_ok=True) scene = variant_saved / f"{view['name']}.vfmesh" image = variant_capture / f"{view['name']}.png" candidates = [(view["camera"], view["target"], view["focus"], view.get("route_distance_m"), 0.0, "requested")] if variant == "before" and view["kind"] != "mouth": passage = next(item for item in layout["passages"] if item["stable_id"] == view["passage"]) route = route_for(layout, passage) lengths, prefix = route_lengths(route) total = prefix[-1] eye_offset = sub(view["camera"], view["focus"]) target_offset = sub(view["target"], view["focus"]) requested = view["route_distance_m"] for delta in (5, -5, 10, -10, 20, -20, 35, -35, 50, -50, 75, -75, 100, -100, 150, -150, 200, -200): distance = max(0.0, min(total, requested + delta)) if abs(distance - requested) < 0.01: continue center, _radius = at_distance(route, lengths, prefix, distance) candidates.append((add(center, eye_offset), add(center, target_offset), center, distance, distance - requested, "route-offset")) attempt_errors = [] for attempt_index, (probe_camera, probe_target, probe_focus, route_distance, route_adjustment, attempt_kind) in enumerate(candidates): suffix = "" if attempt_index == 0 else f"-attempt-{attempt_index:02d}" headless_log = variant_saved / f"{view['name']}{suffix}-mesh.log" render_log = variant_saved / f"{view['name']}{suffix}-render.log" quality_log = variant_saved / f"{view['name']}{suffix}-quality.log" yaw, pitch = angles(probe_camera, probe_target) camera = (*probe_camera, yaw, pitch) export_command = [ str(BINARIES[variant]), str(DATA), "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", "--workers=12", "--clip-radius=3", "--max-clip-level=0", "--max-level=0", "--force-generate", "--recipe=surface-world", f"--camera={fmt(camera)}", "--render-distance-chunks=1", f"--mesh-export={scene}", ] export_output = run_process(f"{variant}-{view['name']}{suffix}-mesh", export_command, headless_log) if "status=PASS" not in export_output or not scene.is_file(): raise RuntimeError(f"{variant}-{view['name']}: missing successful scene export") source_triangles = mesh_triangle_count(scene) if source_triangles <= 0: attempt_errors.append(f"{attempt_kind}@{route_distance}: empty mesh") if variant == "after" or view["kind"] == "mouth": break continue actual_camera, actual_target = probe_camera, probe_target actual_focus = probe_focus surface_distance = None shell_standoff_actual = None if attempt_kind == "route-offset" or view["kind"] == "open-air-closeup": anchor_point = probe_focus if attempt_kind == "route-offset" else probe_target anchor = nearest_surface_anchor(scene, anchor_point, margin_m=20.0) surface_distance = anchor["distance_m"] if view["kind"] == "open-air-closeup": radial = norm(sub(anchor["position_m"], view["focus"])) normal = anchor["normal"] if dot(normal, radial) < 0.0: normal = mul(normal, -1.0) actual_camera = add(anchor["position_m"], mul(normal, 4.0)) actual_target = anchor["position_m"] actual_focus = anchor["position_m"] shell_standoff_actual = math.dist(actual_camera, actual_target) else: actual_camera = add(anchor["position_m"], mul(anchor["normal"], 12.0)) actual_target = anchor["position_m"] actual_focus = anchor["position_m"] actual_yaw, actual_pitch = angles(actual_camera, actual_target) render_command = [ str(RENDERERS[variant]), str(scene), str(image), "--size=1920x1080", f"--camera-m={fmt(actual_camera)}", f"--yaw-deg={actual_yaw:.6f}", f"--pitch-deg={actual_pitch:.6f}", "--shade=readable", "--headlamp=on", "--height-tint=on", f"--height-center-m={actual_focus[2]:.6f}", "--height-range-m=32", "--tile-skirts=off", ] render_output = run_process(f"{variant}-{view['name']}{suffix}-render", render_command, render_log) if "[OffscreenMeshDisplay] tile_skirts=off" not in render_output: raise RuntimeError(f"{variant}-{view['name']}: skirt-free surface render missing") drawn = DRAWN.search(render_output) if drawn is None or int(drawn.group(1)) <= 0 or not image.is_file(): attempt_errors.append(f"{attempt_kind}@{route_distance}: no drawn geometry") if variant == "after" or view["kind"] == "mouth": break continue quality = run_process(f"{variant}-{view['name']}{suffix}-quality", [ "python3", "-B", str(ROOT / "Standalone/Tools/verify/capture_quality.py"), str(image)], quality_log, allow_failure=True) quality_status = "pass" if "status=PASS" not in quality: metrics = QUALITY_METRICS.search(quality) coverage = float(metrics.group(2)) if metrics else 0.0 if coverage >= 0.001: quality_status = "low_contrast_geometry_visible" else: attempt_errors.append(f"{attempt_kind}@{route_distance}: no visible pixel coverage") if variant == "after" or view["kind"] == "mouth": break continue row = {**view, "variant": variant, "capture": str(image.relative_to(RESEARCH)), "scene": str(scene.relative_to(ROOT)), "drawn_triangles": int(drawn.group(1)), "source_triangles": source_triangles, "verified": True, "render_tile_skirts": "off", "pixel_quality_status": quality_status, "route_distance_actual_m": route_distance, "route_station_adjustment_m": route_adjustment, "fallback_kind": attempt_kind if attempt_kind != "requested" else None, "fallback_surface_distance_m": surface_distance, "shell_standoff_actual_m": shell_standoff_actual, "camera_actual_m": actual_camera, "target_actual_m": actual_target, "logs": [str(headless_log.relative_to(ROOT)), str(render_log.relative_to(ROOT)), str(quality_log.relative_to(ROOT))]} print("[L110Capture] " + json.dumps(row, sort_keys=True), flush=True) return row raise RuntimeError(f"{variant}-{view['name']}: no verified image; " + "; ".join(attempt_errors)) def main() -> int: if any(not path.is_file() for path in (*BINARIES.values(), *RENDERERS.values(), *LAYOUTS.values(), AUDIT_LOG)): raise SystemExit("missing L109 capture binary, layout, or audit log") sections = [] for passage, section, start, end, samples, midpoint, radius in SECTION.findall( AUDIT_LOG.read_text(encoding="utf-8", errors="replace")): sections.append({"passage": passage, "section": int(section), "start_m": float(start), "end_m": float(end), "samples": int(samples), "midpoint": vector(midpoint), "radius_m": float(radius)}) if not sections: raise SystemExit("no open-air sections found in the final passage audit") all_rows = [] for variant in ("before", "after"): layout = json.loads(LAYOUTS[variant].read_text(encoding="utf-8")) include_open_air = True views = make_views(layout, sections, use_derived_shape=True, include_open_air=include_open_air) expected = 5 * len(layout["passages"]) if include_open_air: expected += 4 * len(sections) + 2 if len(views) != expected: raise RuntimeError(f"unexpected camera count: {len(views)} != {expected}") print(f"[L110CapturePlan] variant={variant} passage_views={5*len(layout['passages'])} " f"open_air_sections={len(sections) if include_open_air else 0} " f"open_air_views={3*len(sections) if include_open_air else 0} " f"exterior_closeups={len(sections)+2 if include_open_air else 0} " f"total={len(views)}", flush=True) for view in views: all_rows.append(capture(variant, view, layout)) CAPTURES.mkdir(parents=True, exist_ok=True) (CAPTURES / "capture-index.json").write_text( json.dumps({"open_air_sections": sections, "captures": all_rows}, indent=2) + "\n", encoding="utf-8") lines = ["# L110 passage captures", "", "Each PNG was rendered from its own exported mesh. The renderer reported visible " "triangles with LOD seam skirts disabled for surface review, and each decoded " "image met the pixel coverage floor. " "`low_contrast_geometry_visible` marks images that met this floor below the " "preferred contrast target.", ""] for variant in ("before", "after"): lines.extend([f"## {variant.title()}", ""]) for row in all_rows: if row["variant"] != variant: continue flag = " **OPEN AIR**" if row["kind"].startswith("open-air") else "" quality_note = (f"; {row['pixel_quality_status']}" if row["pixel_quality_status"] != "pass" else "") lines.append(f"- [{row['station']}]({variant}/{Path(row['capture']).name}) — " f"{row['passage']}{flag}; {row['drawn_triangles']} visible triangles" f"{quality_note}") lines.append("") (CAPTURES / "INDEX.md").write_text("\n".join(lines), encoding="utf-8") failures = [row for row in all_rows if not row["verified"]] print(f"[L110CaptureSummary] captures={len(all_rows)} verified={len(all_rows)-len(failures)} " f"failed={len(failures)} open_air_sections={len(sections)}", flush=True) return 1 if failures else 0 if __name__ == "__main__": raise SystemExit(main())