Files
VoxelForge/Standalone/Research/WorldAssemblerM3/Linux113/RunL113Captures.py
T
Fr0zkaandClaude Opus 5.5 36d954cf2f passages: rounded half-pipe crest (L113, v220); floor bands diagnosed, speed not recovered
Open-air walls: 0.75 m rounded shell, relief fades over the top 0.25 m - the needle spikes are
gone. Every owner plan passes the 15 deg forward-grade check (max ~10.7 deg) and all walk audits.
Floor bands: median 1.25 m interval, ~0.21 m drop on a ~10.4 deg floor = one 0.25 m voxel row
projected along the grade (not control points, not slab thickness) - not yet fixed. Exact
segment-to-tile filter did not reduce candidates; removed. Timings this round are unreliable
(a gitea process used ~43% CPU throughout). Fast tier PASS; Clang/GCC deterministic; UE builds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016dhxM43dnWvUu2B5Ph1iG1
2026-10-29 21:57:13 +01:00

893 lines
44 KiB
Python

#!/usr/bin/env python3
"""Compare copied L112 captures against L113 routes and verified 4 m wall views."""
from __future__ import annotations
import hashlib
import json
import math
import re
from array import array
import shutil
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/l113/captures"
DATA = ROOT / "Standalone/data/owner_world.json"
BINARIES = {
"after": ROOT / "Saved/Linux/l113/bin/VoxelForgeCoreHeadless-l113-clang",
}
LAYOUTS = {
"before": ROOT / "Saved/Linux/l112/audit-layout-candidate.json",
"after": ROOT / "Saved/Linux/l113/audit-layout-candidate.json",
}
AUDIT_LOGS = {
"before": ROOT / "Saved/Linux/l112/audit-owner-clang.log",
"after": ROOT / "Saved/Linux/l113/audit-owner-clang.log",
}
RENDERERS = {
"after": ROOT / "Saved/Linux/l113/bin/VoxelForgeOffscreenCapture-l113-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("<I", 1):
positions.byteswap()
normals.byteswap()
indices.byteswap()
def vertex(index):
offset = index * 3
return tuple(float(positions[offset + axis]) for axis in range(3))
def vertex_normal(index):
offset = index * 3
return tuple(float(normals[offset + axis]) for axis in range(3))
for index in range(0, index_count - 2, 3):
triangle = (indices[index], indices[index + 1], indices[index + 2])
if any(vertex_index >= 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 the runtime adaptive PassageSpline::BuildSegments 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
maximum_spacing = 8.0
maximum_chord_error = 0.025
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
span_length = arc[-1]
segment_count = max(1, math.ceil(span_length / maximum_spacing))
def parameter_at_arc(target):
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
return (lower + fraction) / arc_steps
parameters = [0.0]
def append_adaptive(t_a, t_b, a, b, depth):
maximum_error = 0.0
polyline_length = 0.0
previous_point = a
for fraction in (0.25, 0.5, 0.75):
current = spline_point(controls, span, t_a + (t_b - t_a) * fraction)
dx = current[0] - (a[0] + (b[0] - a[0]) * fraction)
dy = current[1] - (a[1] + (b[1] - a[1]) * fraction)
dr = current[3] - (a[3] + (b[3] - a[3]) * fraction)
maximum_error = max(maximum_error, math.sqrt(dx * dx + dy * dy + dr * dr))
polyline_length += math.hypot(current[0] - previous_point[0],
current[1] - previous_point[1])
previous_point = current
polyline_length += math.hypot(b[0] - previous_point[0],
b[1] - previous_point[1])
split = (polyline_length > maximum_spacing * (1.0 + 1.0e-9)
or maximum_error > maximum_chord_error) and depth < 12
if split:
t_mid = (t_a + t_b) * 0.5
middle = spline_point(controls, span, t_mid)
append_adaptive(t_a, t_mid, a, middle, depth + 1)
append_adaptive(t_mid, t_b, middle, b, depth + 1)
else:
parameters.append(t_b)
for segment in range(1, segment_count + 1):
target_a = span_length * (segment - 1) / segment_count
target_b = span_length * segment / segment_count
t_a, t_b = parameter_at_arc(target_a), parameter_at_arc(target_b)
append_adaptive(t_a, t_b,
spline_point(controls, span, t_a),
spline_point(controls, span, t_b), 0)
points = [spline_point(controls, span, t) for t in parameters]
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"[L113CaptureProcess] 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("[L113Capture] " + 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:
before_index = ROOT / "Standalone/Research/WorldAssemblerM3/Linux112/captures/capture-index.json"
required = (*BINARIES.values(), *RENDERERS.values(), *LAYOUTS.values(),
*AUDIT_LOGS.values(), before_index)
if any(not path.is_file() for path in required):
raise SystemExit("missing L112 before captures or L113 capture inputs")
(SAVED / "before").mkdir(parents=True, exist_ok=True)
before_document = json.loads(before_index.read_text(encoding="utf-8"))
before_rows = []
before_source = ROOT / "Standalone/Research/WorldAssemblerM3/Linux112"
for source_row in before_document.get("captures", []):
if source_row.get("variant") != "after":
continue
source_image = before_source / source_row["capture"]
target_image = CAPTURES / "before" / Path(source_row["capture"]).name
target_image.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source_image, target_image)
source_hash = hashlib.sha256(source_image.read_bytes()).hexdigest()
target_hash = hashlib.sha256(target_image.read_bytes()).hexdigest()
if source_hash != target_hash:
raise RuntimeError(f"L112 before capture copy changed bytes: {source_image}")
quality_log = SAVED / "before" / f"{target_image.stem}-quality.log"
quality = run_process(f"before-{target_image.stem}-quality", [
"python3", "-B", str(ROOT / "Standalone/Tools/verify/capture_quality.py"),
str(target_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:
raise RuntimeError(f"L112 before capture failed image quality: {source_image}")
quality_status = "low_contrast_geometry_visible"
row = dict(source_row)
row["variant"] = "before"
row["capture"] = str(target_image.relative_to(RESEARCH))
row["source_capture"] = str(source_image.relative_to(ROOT))
row["source_sha256"] = source_hash
row["verified"] = True
row["pixel_quality_status"] = quality_status
row["logs"] = [*row.get("logs", []), str(quality_log.relative_to(ROOT))]
before_rows.append(row)
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
layout = json.loads(LAYOUTS["after"].read_text(encoding="utf-8"))
views = make_views(layout, sections_by_variant["after"], use_derived_shape=True,
use_spline=True, include_open_air=True)
expected = 5 * len(layout["passages"]) + 5 * len(sections_by_variant["after"])
if len(views) != expected:
raise RuntimeError(f"unexpected L113 camera count: {len(views)} != {expected}")
print(f"[L113CapturePlan] passages={len(layout['passages'])} "
f"open_air_sections={len(sections_by_variant['after'])} "
f"after_views={len(views)} before_copies={len(before_rows)}", flush=True)
all_rows = before_rows
for view in views:
all_rows.append(capture("after", view, layout))
if not before_rows:
raise RuntimeError("no L112 captures were copied to the L113 before set")
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 = ["# L113 passage captures", "",
"Before images are byte-identical copies of the verified L112 captures. "
"After images use L113 meshes. Every image has nonzero visible triangles, "
"a passing pixel coverage check, and the 4 m close-ups are surface-snapped.", ""]
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"]
open_air_views = [row for row in all_rows
if row["kind"] == "open-air" and row["station"].endswith("view-1")]
closeups_ok = (len(closeups) == 2 * len(open_air_views)
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"[L113CaptureSummary] captures={len(all_rows)} "
f"verified={len(all_rows)-len(failures)} failed={len(failures)} "
f"4m_closeups={len(closeups)} closeups_verified={str(closeups_ok).lower()}",
flush=True)
return 1 if failures or not closeups_ok else 0
if __name__ == "__main__":
raise SystemExit(main())