#!/usr/bin/env python3 """Capture L109 and L112 routes, every open-air section, and verified 4 m wall views.""" from __future__ import annotations import json import math import re from array import array import struct import subprocess import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parents[4] RESEARCH = Path(__file__).resolve().parent CAPTURES = RESEARCH / "captures" SAVED = ROOT / "Saved/Linux/l112/captures" DATA = ROOT / "Standalone/data/owner_world.json" BINARIES = { "before": ROOT / "Saved/Linux/l109/bin-final/VoxelForgeCoreHeadless-l109-clang", "after": ROOT / "Saved/Linux/l112/bin-clang/VoxelForgeCoreHeadless-l112-clang", } LAYOUTS = { "before": ROOT / "Saved/Linux/l109/audit-layout-final-stamp.json", "after": ROOT / "Saved/Linux/l112/audit-layout-candidate.json", } AUDIT_LOGS = { "before": ROOT / "Saved/Linux/l109/audit-owner4-final-stamp.log", "after": ROOT / "Saved/Linux/l112/audit-owner-clang.log", } RENDERERS = { "before": ROOT / "Saved/Linux/l109/bin-final/VoxelForgeOffscreenCapture-l109-clang", "after": ROOT / "Saved/Linux/l112/bin-clang/VoxelForgeOffscreenCapture-l112-clang", } 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 closest_point_triangle(point, a, b, c): def closest_segment(start, end, first_weight, second_weight): edge = sub(end, start) length_squared = dot(edge, edge) t = (min(1.0, max(0.0, dot(sub(point, start), edge) / length_squared)) if length_squared > 1.0e-16 else 0.0) position = add(start, mul(edge, t)) weights = [0.0, 0.0, 0.0] weights[first_weight], weights[second_weight] = 1.0 - t, t return position, tuple(weights) ab, ac = sub(b, a), sub(c, a) cross = ((ab[1] * ac[2] - ab[2] * ac[1]), (ab[2] * ac[0] - ab[0] * ac[2]), (ab[0] * ac[1] - ab[1] * ac[0])) if dot(cross, cross) <= 1.0e-16: candidates = [closest_segment(a, b, 0, 1), closest_segment(b, c, 1, 2), closest_segment(c, a, 2, 0)] return min(candidates, key=lambda candidate: dot(sub(candidate[0], point), sub(candidate[0], point))) ap = sub(point, a) d1, d2 = dot(ab, ap), dot(ac, ap) if d1 <= 0.0 and d2 <= 0.0: return a, (1.0, 0.0, 0.0) bp = sub(point, b) d3, d4 = dot(ab, bp), dot(ac, bp) if d3 >= 0.0 and d4 <= d3: return b, (0.0, 1.0, 0.0) vc = d1 * d4 - d3 * d2 if vc <= 0.0 and d1 >= 0.0 and d3 <= 0.0: denominator = d1 - d3 if denominator <= 1.0e-16: return a, (1.0, 0.0, 0.0) v = d1 / denominator return add(a, mul(ab, v)), (1.0 - v, v, 0.0) cp = sub(point, c) d5, d6 = dot(ab, cp), dot(ac, cp) if d6 >= 0.0 and d5 <= d6: return c, (0.0, 0.0, 1.0) vb = d5 * d2 - d1 * d6 if vb <= 0.0 and d2 >= 0.0 and d6 <= 0.0: denominator = d2 - d6 if denominator <= 1.0e-16: return a, (1.0, 0.0, 0.0) w = d2 / denominator return add(a, mul(ac, w)), (1.0 - w, 0.0, w) va = d3 * d6 - d5 * d4 if va <= 0.0 and d4 - d3 >= 0.0 and d5 - d6 >= 0.0: denominator = (d4 - d3) + (d5 - d6) if denominator <= 1.0e-16: return b, (0.0, 1.0, 0.0) w = (d4 - d3) / denominator edge = sub(c, b) return add(b, mul(edge, w)), (0.0, 1.0 - w, w) face_denominator = va + vb + vc if face_denominator <= 1.0e-16: candidates = [closest_segment(a, b, 0, 1), closest_segment(b, c, 1, 2), closest_segment(c, a, 2, 0)] return min(candidates, key=lambda candidate: dot(sub(candidate[0], point), sub(candidate[0], point))) denominator = 1.0 / face_denominator v, w = vb * denominator, vc * denominator weights = (1.0 - v - w, v, w) closest = tuple(weights[0] * a[axis] + weights[1] * b[axis] + weights[2] * c[axis] for axis in range(3)) return closest, weights def nearest_surface_anchor(mesh: Path, point, margin_m=200.0, *, preferred_normal=None, minimum_alignment=-1.0) -> dict: """Return the nearest point on a VFMSH surface triangle 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) indices = array("I") indices.frombytes(stream.read(index_count * 4)) stream.seek(mask_count, 1) if struct.pack("=I", 1) != struct.pack("= surface_count or vertex_index >= vertex_count for vertex_index in triangle): continue a, b, c = (vertex(vertex_index) for vertex_index in triangle) position, weights = closest_point_triangle(point, a, b, c) delta = sub(position, point) distance_sq = dot(delta, delta) if distance_sq >= best_distance_sq: continue raw_normals = [vertex_normal(vertex_index) for vertex_index in triangle] raw_normal = tuple(sum(weights[corner] * raw_normals[corner][axis] for corner in range(3)) for axis in range(3)) if dot(raw_normal, raw_normal) < 1.0e-10: raw_normal = ( (b[1] - a[1]) * (c[2] - a[2]) - (b[2] - a[2]) * (c[1] - a[1]), (b[2] - a[2]) * (c[0] - a[0]) - (b[0] - a[0]) * (c[2] - a[2]), (b[0] - a[0]) * (c[1] - a[1]) - (b[1] - a[1]) * (c[0] - a[0])) if dot(raw_normal, raw_normal) < 1.0e-10: continue candidate_normal = norm(raw_normal) if preferred_normal is not None: alignment = dot(candidate_normal, preferred_normal) if abs(alignment) < minimum_alignment: continue if alignment < 0.0: candidate_normal = mul(candidate_normal, -1.0) best_distance_sq = distance_sq best_position = position best_normal = candidate_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_height_at_xy(route, x, y): if not route: return None if len(route) == 1: point = route[0] return point[2], (x - point[0]) ** 2 + (y - point[1]) ** 2 best = (float("inf"), None) for a, b in zip(route, route[1:]): dx, dy = b[0] - a[0], b[1] - a[1] length_sq = dx * dx + dy * dy t = min(1.0, max(0.0, ((x - a[0]) * dx + (y - a[1]) * dy) / length_sq)) \ if length_sq > 1.0e-12 else 0.0 px, py = a[0] + t * dx, a[1] + t * dy distance_sq = (x - px) ** 2 + (y - py) ** 2 if distance_sq < best[0]: best = (distance_sq, a[2] + (b[2] - a[2]) * t) return best[1], best[0] def upper_mouth_floor_depth(layout: dict, passage: dict, controls) -> float: scale = layout["core_voxel_size_um"] / 256.0 / 1_000_000.0 authored = max(0.0, controls[0][0][2] - passage["upper_mouth_q"][2] * scale + 1.13) stratum = next((row for row in layout["strates"] if row["strate_index"] == passage["upper_strate_index"]), None) if stratum is None or stratum["generator_kind"] != "FloatingIslands": return authored route = [tuple(value * scale for value in anchor["values"][:3]) for anchor in stratum.get("anchors", []) if anchor.get("kind") == "route_point" and len(anchor.get("values", [])) >= 3] projected = route_height_at_xy(route, controls[0][0][0], controls[0][0][1]) if projected is None or projected[1] > (controls[0][1] + 2.0) ** 2: return authored route_depth = max(0.0, controls[0][0][2] - projected[0] + 1.13) return max(authored, route_depth) def route_controls(layout: dict, passage: dict, *, adjust_upper_mouth=False): 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") if lower_index < 1: raise RuntimeError(f"{passage['stable_id']}: incomplete route controls") if adjust_upper_mouth: controls[0] = (controls[0][0], max(controls[0][1], upper_mouth_floor_depth(layout, passage, controls))) return controls, lower_index def route_for(layout: dict, passage: dict) -> list[tuple[tuple[float, float, float], float]]: controls, lower_index = route_controls(layout, passage) return controls[:lower_index + 1] def spline_knot_span(a, b): return max(1.0e-6, math.sqrt(math.sqrt((b[0] - a[0]) ** 2 + (b[1] - a[1]) ** 2))) def spline_point(controls, span, t): as_point = lambda index: (*controls[index][0], controls[index][1]) p0, p1, p2, p3 = (as_point(max(0, span - 1)), as_point(span), as_point(span + 1), as_point(min(span + 2, len(controls) - 1))) dt0, dt1, dt2 = (spline_knot_span(p0, p1), spline_knot_span(p1, p2), spline_knot_span(p2, p3)) t2, t3 = t * t, t * t * t result = [0.0] * 4 for axis in (0, 1, 3): m1 = dt1 * ((p1[axis] - p0[axis]) / dt0 - (p2[axis] - p0[axis]) / (dt0 + dt1) + (p2[axis] - p1[axis]) / dt1) m2 = dt1 * ((p2[axis] - p1[axis]) / dt1 - (p3[axis] - p1[axis]) / (dt1 + dt2) + (p3[axis] - p2[axis]) / dt2) result[axis] = ((2 * t3 - 3 * t2 + 1) * p1[axis] + (-2 * t3 + 3 * t2) * p2[axis] + (t3 - 2 * t2 + t) * m1 + (t3 - t2) * m2) result[2] = p1[2] + (p2[2] - p1[2]) * t result[3] = min(max(result[3], min(p1[3], p2[3])), max(p1[3], p2[3])) return result def smooth_route(layout: dict, passage: dict): """Mirror PassageSpline::BuildSegments from the runtime shared sampler.""" scale = layout["core_voxel_size_um"] / 256.0 / 1_000_000.0 planned_floor_profile = [ (row["position_q"][2] - row["radius_q"]) * scale for row in passage["control_points"]] controls, route_end = route_controls(layout, passage, adjust_upper_mouth=True) result = [] along = 0.0 arc_steps = 32 for span in range(route_end): arc = [0.0] previous = spline_point(controls, span, 0.0) for step in range(1, arc_steps + 1): current = spline_point(controls, span, step / arc_steps) arc.append(arc[-1] + math.hypot(current[0] - previous[0], current[1] - previous[1])) previous = current segment_count = max(1, math.ceil(arc[-1])) points = [spline_point(controls, span, 0.0)] parameters = [0.0] for segment in range(1, segment_count + 1): target = min(arc[-1], arc[-1] * segment / segment_count) upper = next(index for index in range(1, len(arc)) if arc[index] >= target) lower = upper - 1 width = arc[upper] - arc[lower] fraction = min(1.0, max(0.0, (target - arc[lower]) / width)) if width > 1e-12 else 0.0 t = (lower + fraction) / arc_steps points.append(spline_point(controls, span, t)) parameters.append(t) sampled_arc = [0.0] for a, b in zip(points, points[1:]): sampled_arc.append(sampled_arc[-1] + math.hypot(b[0] - a[0], b[1] - a[1])) horizontal_length = sampled_arc[-1] for index, point in enumerate(points): progress = sampled_arc[index] / horizontal_length if horizontal_length > 1e-12 else parameters[index] point[2] = controls[span][0][2] + (controls[span + 1][0][2] - controls[span][0][2]) * progress point.append(planned_floor_profile[span] + (planned_floor_profile[span + 1] - planned_floor_profile[span]) * progress) if not result: result.append((tuple(points[0][:3]), points[0][3], 0.0, points[0][4])) for a, b in zip(points, points[1:]): length = math.dist(a[:3], b[:3]) along += length result.append((tuple(b[:3]), b[3], along, b[4])) return result def derived_smooth_route(route, passage, layout): total = route[-1][2] interior_radii = [radius for _, radius, distance, _ in route if distance >= 10.0 and total - distance >= 10.0] body_depth = max(0.5, min(interior_radii or [radius for _, radius, _, _ in route]) - 0.23) result = [] for point, radius, distance, planned_floor in route: local_radius_depth = max(0.0, radius - 0.23) tube_depth = min(body_depth, local_radius_depth) center = (point[0], point[1], planned_floor + tube_depth) result.append((center, radius, distance, planned_floor)) return result, body_depth def route_lengths(route): lengths = [((b[2] - a[2]) if len(a) > 2 and len(b) > 2 else 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, use_spline: 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 = smooth_route(layout, passage) if use_spline else route_for(layout, passage) if use_derived_shape: route, body_depth = (derived_smooth_route(raw_route, passage, layout) if use_spline else 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}) floor_z = center[2] - depth outside_target = add(center, add(mul(side, radius), (0.0, 0.0, -depth + 0.5))) outside_eye = add(outside_target, mul(side, 4.0)) interior_radius = math.sqrt(max(0.0, radius * radius - max(0.0, depth - 0.5) ** 2)) inside_target = add(center, add(mul(side, interior_radius), (0.0, 0.0, -depth + 0.5))) inside_eye = add(inside_target, add(mul(side, -1.0), mul(tangent, math.sqrt(15.0)))) for closeup, eye, target in (("outside", outside_eye, outside_target), ("interior-wall", inside_eye, inside_target)): yaw, pitch = angles(eye, target) views.append({"name": f"{safe_id}-openair-{section_id}-4m-{closeup}", "passage": passage_id, "kind": "open-air-close", "station": f"section-{section_id}/4m-{closeup}", "camera": eye, "target": target, "yaw": yaw, "pitch": pitch, "focus": center, "center_z": center[2], "open_air_section": section_id, "closeup_side": closeup, "closeup_preferred_normal": side if closeup == "outside" else mul(side, -1.0), "camera_target_distance_m": math.dist(eye, target), "route_distance_m": distance, "requested_route_distance_m": distance, "route_length_m": total}) 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"[L112CaptureProcess] 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"] not in ("mouth", "open-air-close"): 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 closeup_surface_distance = None closeup_anchor_offset = None closeup_normal_alignment = None closeup_distance = None if view["kind"] == "open-air-close": try: wall_anchor = nearest_surface_anchor(scene, probe_target, margin_m=5.0, preferred_normal=view["closeup_preferred_normal"], minimum_alignment=0.25) closeup_surface_distance = wall_anchor["distance_m"] closeup_anchor_offset = closeup_surface_distance closeup_normal_alignment = dot(wall_anchor["normal"], view["closeup_preferred_normal"]) if closeup_surface_distance > 5.0: raise RuntimeError(f"{variant}-{view['name']}: nearest wall surface is " f"{closeup_surface_distance:.3f} m from the route-local target") actual_target = wall_anchor["position_m"] actual_camera = add(actual_target, mul(wall_anchor["normal"], 4.0)) closeup_distance = math.dist(actual_camera, actual_target) if abs(closeup_distance - 4.0) > 0.02: raise RuntimeError(f"{variant}-{view['name']}: camera-to-wall distance is " f"{closeup_distance:.3f} m, expected 4 m") except RuntimeError as error: raise RuntimeError(f"{variant}-{view['name']}: could not verify the 4 m wall " f"view: {error}") from error if attempt_kind == "route-offset": anchor = nearest_surface_anchor(scene, probe_focus) surface_distance = anchor["distance_m"] 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", ] render_output = run_process(f"{variant}-{view['name']}{suffix}-render", render_command, render_log) 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, "pixel_quality_status": quality_status, "route_distance_actual_m": route_distance, "route_station_adjustment_m": route_adjustment, "closeup_camera_target_distance_m": closeup_distance, "closeup_target_surface_distance_m": closeup_surface_distance, "closeup_predicted_target_offset_m": closeup_anchor_offset, "closeup_wall_normal_alignment": closeup_normal_alignment, "fallback_kind": attempt_kind if attempt_kind != "requested" else None, "fallback_surface_distance_m": surface_distance, "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("[L112Capture] " + 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_LOGS.values())): raise SystemExit("missing L109 capture binary, layout, or audit log") all_rows = [] variants = ("before", "after") if "--after-only" in sys.argv[1:]: prior_log = SAVED.parent / "captures-before-first.log" if not prior_log.is_file(): raise SystemExit(f"missing reusable L109 capture rows: {prior_log}") for line in prior_log.read_text(encoding="utf-8", errors="replace").splitlines(): if line.startswith("[L112Capture] ") or line.startswith("[L109Capture] "): all_rows.append(json.loads(line.split(" ", 1)[1])) if not all_rows or any(row["variant"] != "before" for row in all_rows): raise SystemExit("saved L109 capture rows are incomplete") variants = ("after",) sections_by_variant = {} for variant in ("before", "after"): sections = [] for passage, section, start, end, samples, midpoint, radius in SECTION.findall( AUDIT_LOGS[variant].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(f"no open-air sections found in the {variant} passage audit") sections_by_variant[variant] = sections if variant not in variants: continue if variant == "after" and variants == ("after",): expected_before = 15 + 5 * len(sections_by_variant["before"]) if len(all_rows) != expected_before or any( row.get("variant") != "before" or not row.get("verified") for row in all_rows): raise RuntimeError("saved L109 capture rows do not cover every required view") layout = json.loads(LAYOUTS[variant].read_text(encoding="utf-8")) include_open_air = True views = make_views(layout, sections, use_derived_shape=True, use_spline=variant == "after", include_open_air=include_open_air) expected = 5 * len(layout["passages"]) if include_open_air: expected += 5 * len(sections) if len(views) != expected: raise RuntimeError(f"unexpected camera count: {len(views)} != {expected}") print(f"[L112CapturePlan] variant={variant} passage_views={5*len(layout['passages'])} " f"open_air_sections={len(sections) if include_open_air else 0} " f"open_air_views={5*len(sections) 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_by_variant": sections_by_variant, "captures": all_rows}, indent=2) + "\n", encoding="utf-8") lines = ["# L112 passage captures", "", "Each PNG was rendered from its own exported mesh. The renderer reported visible " "triangles, 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"] == "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"]] closeups = [row for row in all_rows if row["kind"] == "open-air-close"] closeups_ok = (len(closeups) == 2 * sum(1 for row in all_rows if row["kind"] == "open-air" and row["station"].endswith("view-1")) and all(row["verified"] and abs(row["closeup_camera_target_distance_m"] - 4.0) <= 0.02 and row["closeup_target_surface_distance_m"] <= 5.0 and row["closeup_wall_normal_alignment"] >= 0.25 and row["drawn_triangles"] > 0 and row["pixel_quality_status"] in ("pass", "low_contrast_geometry_visible") for row in closeups)) print(f"[L112CaptureSummary] captures={len(all_rows)} verified={len(all_rows)-len(failures)} " f"failed={len(failures)} 4m_closeups={len(closeups)} closeups_verified=" f"{str(closeups_ok).lower()}", flush=True) return 1 if failures or not closeups_ok else 0 if __name__ == "__main__": raise SystemExit(main())