Files
VoxelForge/Standalone/Research/WorldAssemblerM3/Linux129/CountReachCandidates.py
T
Fr0zkaandClaude Opus 5.5 0d1140c117 research: tunnel reach study (L129, no-go; found a units mismatch in v225 bounds)
BuildPassageSpatialIndices adds the 4.335 m reach in core-voxel units, so the broadphase reaches ~1.08 m while
the sweep cutoff says 4 m. Physically-correct 2 m bounds are larger than v225's and 4.9-9.9% slower; captures at
all L125 poses pixel-identical. Patch kept.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016dhxM43dnWvUu2B5Ph1iG1
2026-10-30 23:09:27 +01:00

311 lines
15 KiB
Python

#!/usr/bin/env python3
"""Count reach-bound work from the saved v225 layout and L125 tile profiles."""
from __future__ import annotations
import json
import math
import re
import subprocess
from pathlib import Path
ROOT = Path(__file__).resolve().parents[4]
HERE = Path(__file__).resolve().parent
OUT = ROOT / "Saved/Linux/l129/precount"
LAYOUT = ROOT / "Saved/Linux/l125/hashes/owner4-surface-world-w8-clang.layout.json"
BASELINE_LOGS = [ROOT / f"Saved/Linux/l125/counts/s{s}-w8-profile1.log" for s in range(4)]
CORE_VOXEL_METRES = 0.25
SUBVOXELS_PER_METRE = 1024.0
PROFILE = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")
TILE_KEY = re.compile(r"\((-?\d+),(-?\d+),(-?\d+),lod(\d+)\)")
def knot_span(a: list[float], b: list[float], units_per_metre: float) -> float:
dx, dy = (b[0] - a[0]) / units_per_metre, (b[1] - a[1]) / units_per_metre
return max(1.0e-6, math.sqrt(math.sqrt(dx * dx + dy * dy)))
def evaluate_raw(controls: list[list[float]], span: int, t: float,
units_per_metre: float) -> list[float]:
p1, p2 = controls[span], controls[span + 1]
p0, p3 = controls[max(0, span - 1)], controls[min(span + 2, len(controls) - 1)]
dt0, dt1, dt2 = (knot_span(p0, p1, units_per_metre),
knot_span(p1, p2, units_per_metre),
knot_span(p2, p3, units_per_metre))
t2, t3 = t * t, t * t * t
result = [0.0, 0.0, 0.0, 0.0]
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.0 * t3 - 3.0 * t2 + 1.0) * p1[axis]
+ (-2.0 * t3 + 3.0 * t2) * p2[axis]
+ (t3 - 2.0 * 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 build_raw_segments(controls: list[list[float]], units_per_metre: float,
spacing_metres: float = 8.0,
chord_error_metres: float = 0.1) -> list[dict]:
result: list[dict] = []
along_metres = 0.0
for span in range(len(controls) - 1):
horizontal_arc = [0.0] * 33
previous = evaluate_raw(controls, span, 0.0, units_per_metre)
for step in range(1, 33):
current = evaluate_raw(controls, span, step / 32.0, units_per_metre)
dx = (current[0] - previous[0]) / units_per_metre
dy = (current[1] - previous[1]) / units_per_metre
horizontal_arc[step] = horizontal_arc[step - 1] + math.hypot(dx, dy)
previous = current
span_length = horizontal_arc[-1]
base_count = max(1, math.ceil(span_length / spacing_metres))
def parameter_at_arc(target: float) -> float:
upper = 1
while upper < len(horizontal_arc) and horizontal_arc[upper] < target:
upper += 1
upper = min(upper, 32)
lower = upper - 1
arc_span = horizontal_arc[upper] - horizontal_arc[lower]
arc_t = ((target - horizontal_arc[lower]) / arc_span
if arc_span > 1.0e-12 else 0.0)
return (lower + min(max(arc_t, 0.0), 1.0)) / 32.0
parameters = [0.0]
def append_adaptive(t_a: float, t_b: float, a: list[float], b: list[float],
depth: int) -> None:
fractions = (0.25, 0.5, 0.75)
maximum_error, polyline_length = 0.0, 0.0
previous_point = a
for fraction in fractions:
current = evaluate_raw(controls, span, t_a + (t_b - t_a) * fraction,
units_per_metre)
dx = (current[0] - (a[0] + (b[0] - a[0]) * fraction)) / units_per_metre
dy = (current[1] - (a[1] + (b[1] - a[1]) * fraction)) / units_per_metre
dr = (current[3] - (a[3] + (b[3] - a[3]) * fraction)) / units_per_metre
maximum_error = max(maximum_error, math.sqrt(dx * dx + dy * dy + dr * dr))
sx = (current[0] - previous_point[0]) / units_per_metre
sy = (current[1] - previous_point[1]) / units_per_metre
polyline_length += math.hypot(sx, sy)
previous_point = current
sx = (b[0] - previous_point[0]) / units_per_metre
sy = (b[1] - previous_point[1]) / units_per_metre
polyline_length += math.hypot(sx, sy)
split = ((polyline_length > spacing_metres * (1.0 + 1.0e-9)
or maximum_error > chord_error_metres) and depth < 12)
if split:
t_mid = (t_a + t_b) * 0.5
middle = evaluate_raw(controls, span, t_mid, units_per_metre)
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, base_count + 1):
target_a = span_length * (segment - 1) / base_count
target_b = span_length * segment / base_count
t_a, t_b = parameter_at_arc(target_a), parameter_at_arc(target_b)
append_adaptive(t_a, t_b,
evaluate_raw(controls, span, t_a, units_per_metre),
evaluate_raw(controls, span, t_b, units_per_metre), 0)
points = [evaluate_raw(controls, span, t, units_per_metre) for t in parameters]
sampled_arc = [0.0] * len(points)
for index in range(1, len(points)):
dx = (points[index][0] - points[index - 1][0]) / units_per_metre
dy = (points[index][1] - points[index - 1][1]) / units_per_metre
sampled_arc[index] = sampled_arc[index - 1] + math.hypot(dx, dy)
sampled_span_length = sampled_arc[-1]
for index, point in enumerate(points):
progress = (sampled_arc[index] / sampled_span_length
if sampled_span_length > 1.0e-12 else parameters[index])
point[2] = controls[span][2] + (controls[span + 1][2] - controls[span][2]) * progress
for index in range(1, len(points)):
dx = (points[index][0] - points[index - 1][0]) / units_per_metre
dy = (points[index][1] - points[index - 1][1]) / units_per_metre
dz = (points[index][2] - points[index - 1][2]) / units_per_metre
length = math.sqrt(dx * dx + dy * dy + dz * dz)
progress_a = (sampled_arc[index - 1] / sampled_span_length
if sampled_span_length > 1.0e-12 else parameters[index - 1])
progress_b = (sampled_arc[index] / sampled_span_length
if sampled_span_length > 1.0e-12 else parameters[index])
floor_a = controls[span][2] - controls[span][3]
floor_b = controls[span + 1][2] - controls[span + 1][3]
result.append({"a": points[index - 1], "b": points[index],
"along": along_metres, "length": length, "span": span,
"floor_a": floor_a + (floor_b - floor_a) * progress_a,
"floor_b": floor_a + (floor_b - floor_a) * progress_b})
along_metres += length
return result
def closest_route_height(route: list[list[float]], x: float, y: float) -> tuple[float, float]:
best_height, best_distance_sq = 0.0, float("inf")
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(max(((x - a[0]) * dx + (y - a[1]) * dy) / length_sq, 0.0), 1.0) \
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_distance_sq:
best_distance_sq = distance_sq
best_height = a[2] + (b[2] - a[2]) * t
return best_height, best_distance_sq
def prepared_segments(layout: dict) -> list[list[dict]]:
qpm = SUBVOXELS_PER_METRE
strata = {item["strate_index"]: item for item in layout["strates"]}
all_passages: list[list[dict]] = []
for passage in layout["passages"]:
controls_q = [[float(value) for value in point["position_q"]]
+ [float(point["radius_q"])] for point in passage["control_points"]]
controls = [point.copy() for point in controls_q]
upper_mouth = [value / qpm for value in passage["upper_mouth_q"]]
upper_depth = max(0.0, (controls[0][2] / qpm) - upper_mouth[2] + 1.13)
upper_stratum = strata.get(passage["upper_strate_index"], {})
if upper_stratum.get("generator_kind") == "FloatingIslands":
route = []
for anchor in upper_stratum.get("anchors", []):
if anchor.get("kind") == "route_point" and len(anchor.get("values", [])) >= 3:
route.append([value / qpm for value in anchor["values"][:3]])
if route:
route_height, distance_sq = closest_route_height(
route, controls[0][0] / qpm, controls[0][1] / qpm)
if distance_sq <= (controls[0][3] / qpm + 2.0) ** 2:
upper_depth = max(upper_depth,
controls[0][2] / qpm - route_height + 1.13)
controls[0][3] = max(controls[0][3], upper_depth * qpm)
lower_mouth = [value for value in passage["lower_mouth_q"]]
player_center_q = round(1.13 * qpm)
route_end = len(controls) - 1
for index in range(1, len(controls_q)):
p = controls_q[index]
if (int(p[0]) == lower_mouth[0] and int(p[1]) == lower_mouth[1]
and int(p[2] - p[3] + player_center_q) == lower_mouth[2]):
route_end = index
raw = build_raw_segments(controls, qpm)
total_path = 0.0
for segment in raw:
if segment["span"] + 1 == route_end:
total_path = segment["along"] + segment["length"]
radii = [(0.0, controls[0][3] / qpm)]
for segment in raw:
if segment["span"] < route_end:
radii.append((segment["along"] + segment["length"],
segment["b"][3] / qpm))
body_radii = [radius for along, radius in radii
if along >= 10.0 and total_path - along >= 10.0]
minimum_body = min(body_radii) if body_radii else min(radius for _, radius in radii)
floor_depth = max(0.5, minimum_body - 0.23)
prepared = []
for segment in raw:
route_segment = segment["span"] < route_end
along_b = segment["along"] + segment["length"]
def effective_z(raw_z: float, radius: float, planned_floor: float,
along: float) -> float:
raw_z /= qpm
radius /= qpm
planned_floor /= qpm
local_depth = min(floor_depth, max(0.0, radius - 0.23))
if route_segment or abs(along - total_path) <= 1.0e-9:
return planned_floor + local_depth
route_depth = max(0.0, radius - 0.23)
return raw_z - route_depth + local_depth
a = [segment["a"][0] / qpm, segment["a"][1] / qpm,
effective_z(segment["a"][2], segment["a"][3],
segment["floor_a"], segment["along"])]
b = [segment["b"][0] / qpm, segment["b"][1] / qpm,
effective_z(segment["b"][2], segment["b"][3],
segment["floor_b"], along_b)]
prepared.append({"a": a, "b": b, "radius_a": segment["a"][3] / qpm,
"radius_b": segment["b"][3] / qpm})
all_passages.append(prepared)
return all_passages
def profile_tiles() -> list[tuple[int, int, int, int, int, int, int, int]]:
result = []
for strate, path in enumerate(BASELINE_LOGS):
if not path.is_file():
raise SystemExit(f"missing baseline profile: {path}")
seen = set()
for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
if not line.startswith("[PassageComposerTile] "):
continue
row = dict(PROFILE.findall(line))
match = TILE_KEY.search(row.get("key", ""))
if not match:
continue
x, y, z, lod = map(int, match.groups())
if lod > 2:
continue
key = (x, y, z, lod)
if key in seen:
continue
seen.add(key)
result.append((strate, x, y, z, lod, int(row.get("features", 0)),
int(row.get("route_candidates", 0)),
int(row.get("sweep_pairs", 0))))
return result
def main() -> int:
if not LAYOUT.is_file():
raise SystemExit(f"missing resolved layout: {LAYOUT}")
OUT.mkdir(parents=True, exist_ok=True)
layout = json.loads(LAYOUT.read_text(encoding="utf-8"))
passages = prepared_segments(layout)
tiles = profile_tiles()
input_path = OUT / "reach-input.tsv"
with input_path.open("w", encoding="utf-8") as output:
for passage_index, segments in enumerate(passages):
for segment_index, segment in enumerate(segments):
values = [*segment["a"], *segment["b"],
segment["radius_a"], segment["radius_b"]]
output.write("S {} {} {}\n".format(passage_index, segment_index,
" ".join(f"{value:.15g}" for value in values)))
for tile in tiles:
output.write("T {}\n".format(" ".join(str(value) for value in tile)))
binary = OUT / "CountReachCandidates"
compile_log = OUT / "compile.log"
with compile_log.open("w", encoding="utf-8") as log:
built = subprocess.run(["g++", "-std=c++20", "-O3", "-DNDEBUG",
str(HERE / "CountReachCandidates.cpp"), "-o", str(binary)],
cwd=ROOT, stdout=log, stderr=subprocess.STDOUT,
check=False)
if built.returncode:
raise SystemExit(f"counter compile failed: {compile_log}")
result_path = OUT / "results.csv"
run_log = OUT / "counter.log"
command = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
str(binary), str(input_path), str(result_path)]
with run_log.open("w", encoding="utf-8") as log:
result = subprocess.run(command, cwd=ROOT, stdout=log, stderr=subprocess.STDOUT,
check=False, timeout=1740)
if result.returncode:
raise SystemExit(f"counter failed: {run_log}; exit={result.returncode}")
print(f"[L129Precount] segments={[len(x) for x in passages]} "
f"profile_tiles={len(tiles)} input={input_path.relative_to(ROOT)} "
f"results={result_path.relative_to(ROOT)}")
return 0
if __name__ == "__main__":
raise SystemExit(main())