Files
VoxelForge/Standalone/Research/WorldAssemblerM3/Linux128/RunL128DensityProbe.py
T
Fr0zkaandClaude Opus 5.5 7022d02e6e research: slab ownership for passage samples (L128, correct but slower; kept as patch)
Exact one-owner slabs (33,813 samples audited, controls fire), floor/seams/determinism green, routes already
satisfy R > S (S up to 7.9 m). But near-ready 8-21% behind 4041d4b and UE 30 m/s near 40.2 -> 31.5%.

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

89 lines
4.3 KiB
Python

#!/usr/bin/env python3
"""Probe reconstructed tunnel floor roots at the nine L119 stations."""
from __future__ import annotations
import csv
import json
import math
import os
import re
import statistics
import subprocess
from pathlib import Path
ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / os.environ.get("L128_PROBE_DIR", "Saved/Linux/l128/floor-probe")
BINARY = ROOT / os.environ.get("L128_PROBE_BINARY",
"Saved/Linux/l128/bin/VoxelForgeCoreHeadless-clang19")
DATA = ROOT / "Standalone/data/owner_world.json"
SLOPE = re.compile(
r"^\[L115DensitySlopeSample\] passage=(\S+) station=(\S+) s_m=([\d.]+) "
r".*?planned_floor_m=(-?[\d.]+) root_floor_m=(-?[\d.]+|nan) "
r"residual_m=(-?[\d.]+|nan)", re.M)
def main() -> int:
OUT.mkdir(parents=True, exist_ok=True)
log, layout = OUT / "owner4-clang.log", OUT / "layout.json"
command = [
"taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
"timeout", "1740s", str(BINARY), str(DATA), "--world-assembler-prototype",
"--stream-benchmark", "--stream-pipeline", "--workers=12", "--clip-radius=3",
"--max-clip-level=8", "--max-level=8",
"--camera=8,16,-372.2626953125,0,0", "--audit-resolved-layout-passages",
f"--resolved-layout-output={layout}",
]
with log.open("w", encoding="utf-8") as stream:
result = subprocess.run(command, cwd=ROOT,
env={**os.environ, "VOXELFORGE_L115_DENSITY_PROFILE": "1"},
stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1740)
if result.returncode:
raise SystemExit(f"density probe failed: {log}; exit={result.returncode}")
text = log.read_text(encoding="utf-8", errors="replace")
groups: dict[tuple[str, str], list[dict]] = {}
for passage, station, distance, planned, root, residual in SLOPE.findall(text):
root_value = float(root) if root != "nan" else math.nan
residual_value = float(residual) if residual != "nan" else math.nan
groups.setdefault((passage, station), []).append({
"route_s_m": float(distance), "planned_floor_m": float(planned),
"root_floor_m": root_value, "offset_m": residual_value})
summary = []
with (OUT / "zero-crossing-offsets.csv").open("w", newline="", encoding="utf-8") as stream:
writer = csv.DictWriter(stream, fieldnames=("passage", "station", "route_s_m",
"planned_floor_m", "root_floor_m", "offset_m"))
writer.writeheader()
for (passage, station), samples in sorted(groups.items()):
samples.sort(key=lambda sample: sample["route_s_m"])
for sample in samples:
writer.writerow({"passage": passage, "station": station, **sample})
finite = [sample["offset_m"] for sample in samples
if math.isfinite(sample["offset_m"])]
adjacent_steps = [abs(right["offset_m"] - left["offset_m"])
for left, right in zip(samples, samples[1:])
if math.isfinite(left["offset_m"])
and math.isfinite(right["offset_m"])]
summary.append({"passage": passage, "station": station,
"samples": len(samples), "finite_roots": len(finite),
"max_abs_offset_m": max(map(abs, finite)) if finite else None,
"max_adjacent_root_step_m": max(adjacent_steps)
if adjacent_steps else None,
"median_abs_offset_m": statistics.median(map(abs, finite))
if finite else None,
"p95_abs_offset_m": sorted(map(abs, finite))[
min(len(finite)-1, math.ceil(0.95*len(finite))-1)]
if finite else None})
if not summary or any(item["samples"] != 51 for item in summary):
raise RuntimeError("density profile is missing 51-sample route stations")
worst_step = max(item["max_adjacent_root_step_m"] or 0.0 for item in summary)
if worst_step > 0.003:
raise RuntimeError(f"floor root jumps by {worst_step:.6f} m between 0.1 m probes")
(OUT / "zero-crossing-summary.json").write_text(
json.dumps(summary, indent=2) + "\n", encoding="utf-8")
print(f"[L128DensityProbe] exit={result.returncode} stations={len(summary)} "
f"log={log.relative_to(ROOT)}", flush=True)
return result.returncode
if __name__ == "__main__":
raise SystemExit(main())