VoxelForge/Standalone/Research/WorldAssemblerM3/Linux81/RunL81PassageSweep.py

#!/usr/bin/env python3
"""Interleaved exact passage-bound padding sweep on the L80 tile census."""
from __future__ import annotations

import json
import math
import re
import statistics
import subprocess
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l81/sweep"
DATA = ROOT / "Standalone/data/owner_world.json"
BINARY = ROOT / "Saved/Linux/l81/bin/VoxelForgeCoreHeadless-l81-clang"
CAMERAS = {
    0: (8.0, 16.0, -372.2626953125),
    1: (25.8369140625, 3.19921875, -737.0),
    2: (19.166015625, -14.2001953125, -753.3740234375),
    3: (-20.0625, -5.068359375, -900.353515625),
}
PADS = ("0", "1", "2", "4", "8", "16", "48", "inf")
REPS = 3
TILE = re.compile(r"^\[PassageComposerTile\] (.*)$", re.M)
FIELD = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")
KEY = re.compile(r"key=\((-?\d+),(-?\d+),(-?\d+),lod(\d+)\)")
HASH = re.compile(r"^\[PipelineTileHash\] lod=(\d+) x=(-?\d+) y=(-?\d+) z=(-?\d+) "
                  r"density=(\S+) mesh=(\S+) payload=(\S+) topology=(\S+)$", re.M)


def parse(log: Path, strate: int, pad: str, rep: int) -> dict:
    content = log.read_text(encoding="utf-8", errors="replace")
    tiles = []
    for match in TILE.finditer(content):
        values = dict(FIELD.findall(match.group(1)))
        key = KEY.search(match.group(1))
        if not key:
            continue
        row = {name: int(value) for name, value in values.items()
               if name not in ("key", "worker_ms")}
        row["worker_ms"] = float(values["worker_ms"])
        row["key"] = tuple(int(key.group(i)) for i in range(1, 5))
        tiles.append(row)
    hashes = {}
    for m in HASH.finditer(content):
        key = (int(m.group(2)), int(m.group(3)), int(m.group(4)), int(m.group(1)))
        hashes[key] = m.group(5, 6, 7, 8)
    if not tiles or not hashes:
        raise RuntimeError(f"missing passage rows or tile hashes in {log}")
    return {"strate": strate, "pad": pad, "replicate": rep, "log": str(log.relative_to(ROOT)),
            "tiles": tiles, "hashes": hashes}


def run(strate: int, pad: str, rep: int) -> dict:
    OUT.mkdir(parents=True, exist_ok=True)
    log = OUT / f"s{strate}-p{pad}-r{rep}.log"
    camera = ",".join(f"{v:.12f}" for v in CAMERAS[strate]) + ",0,0"
    command = [
        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
        "env", f"VOXELFORGE_L81_PASSAGE_BOUND_PADDING_VOXELS={pad}", "timeout", "1740s",
        str(BINARY), str(DATA), "--world-assembler-prototype", "--stream-benchmark",
        "--stream-pipeline", "--workers=8", "--clip-radius=3", "--max-clip-level=8",
        "--max-level=1", f"--camera={camera}", "--profile-lod-work",
        "--profile-passage-composer", "--stream-hash-tiles",
    ]
    start = time.monotonic()
    try:
        with log.open("w", encoding="utf-8") as stream:
            result = subprocess.run(command, cwd=ROOT, stdout=stream,
                                    stderr=subprocess.STDOUT, check=False, timeout=1740)
    except subprocess.TimeoutExpired as exc:
        raise RuntimeError(f"run exceeded 1,740 seconds: {log}") from exc
    if result.returncode:
        raise RuntimeError(f"run failed exit={result.returncode}: {log}")
    parsed = parse(log, strate, pad, rep)
    parsed["wall_s"] = round(time.monotonic() - start, 3)
    parsed["exit_code"] = result.returncode
    print("[L81PassageSweepRun] " + json.dumps(
        {k: v for k, v in parsed.items() if k not in ("tiles", "hashes")}, sort_keys=True),
        flush=True)
    return parsed


def main() -> int:
    if not BINARY.is_file() or not DATA.is_file():
        raise SystemExit(f"missing input: {BINARY} or {DATA}")
    old = json.loads((ROOT / "Saved/Linux/l80/passage-census/candidate-pad4.json")
                     .read_text(encoding="utf-8"))
    high = {int(run["strate"]): {tuple(key) for key in run["keys"]} for run in old}
    results = []
    ordinary: dict[int, set[tuple[int, int, int, int]]] = {}
    for strate in range(4):
        for rep in range(1, REPS + 1):
            offset = (strate + rep - 1) % len(PADS)
            order = PADS[offset:] + PADS[:offset]
            for pad in order:
                run_data = run(strate, pad, rep)
                if strate not in ordinary:
                    candidates = [row for row in run_data["tiles"]
                                  if row["key"] not in high[strate]
                                  and row.get("features", 0) == 0
                                  and row.get("tile_bound_checks", 0) > 0]
                    candidates.sort(key=lambda row: (
                        row.get("tile_bound_rejects", 0), row.get("worker_ms", 0)),
                        reverse=True)
                    ordinary[strate] = {row["key"] for row in candidates[:4]}
                results.append(run_data)

    indexed = {(r["strate"], r["pad"], r["replicate"]): r for r in results}
    pad_rows = []
    failures = []
    for strate in range(4):
        high_keys = high[strate]
        ordinary_keys = ordinary[strate]
        for pad in PADS[:-1]:
            pair_deltas = {"high": [], "ordinary": []}
            sums = {name: {key: 0 for key in ("tile_bound_checks", "tile_bound_rejects",
                    "tile_bound_skipped_iterations", "passage_iterations", "samples")}
                    for name in pair_deltas}
            all_hashes_equal = True
            for rep in range(1, REPS + 1):
                candidate = indexed[(strate, pad, rep)]
                baseline = indexed[(strate, "inf", rep)]
                base_tiles = {row["key"]: row for row in baseline["tiles"]}
                for row in candidate["tiles"]:
                    key = row["key"]
                    if key in high_keys:
                        group = "high"
                    elif key in ordinary_keys:
                        group = "ordinary"
                    else:
                        continue
                    base = base_tiles[key]
                    pair_deltas[group].append(row["worker_ms"] - base["worker_ms"])
                    for metric in sums[group]:
                        sums[group][metric] += row.get(metric, 0)
                    if candidate["hashes"].get(key) != baseline["hashes"].get(key):
                        all_hashes_equal = False
                        failures.append({"strate": strate, "pad": pad, "replicate": rep,
                                         "key": key, "candidate_hash": candidate["hashes"].get(key),
                                         "baseline_hash": baseline["hashes"].get(key)})
            entry = {"strate": strate, "pad": pad, "replicates": REPS,
                     "high_keys": len(high_keys), "ordinary_keys": len(ordinary_keys),
                     "hashes_equal_to_full": all_hashes_equal}
            for group in ("high", "ordinary"):
                n = len(pair_deltas[group])
                count_denominator = sums[group]["tile_bound_skipped_iterations"] \
                    + sums[group]["passage_iterations"]
                entry[group] = {
                    **sums[group],
                    "skipped_iteration_fraction": (
                        sums[group]["tile_bound_skipped_iterations"] / count_denominator
                        if count_denominator else 0.0),
                    "paired_tile_ms_delta_median": statistics.median(pair_deltas[group])
                        if n else 0.0,
                    "paired_tile_ms_delta_mean": statistics.mean(pair_deltas[group])
                        if n else 0.0,
                }
            pad_rows.append(entry)

    payload = {
        "pads_voxels_extra": list(PADS[:-1]), "baseline": "infinite extra pad (fail-open full passage path)",
        "high_tile_keys_from_l80": {str(k): [list(v) for v in sorted(keys)]
                                    for k, keys in high.items()},
        "ordinary_tile_keys": {str(k): [list(v) for v in sorted(keys)]
                               for k, keys in ordinary.items()},
        "sweep_rows": pad_rows,
        "hash_mismatches": failures,
        "runs": [{k: v for k, v in r.items() if k not in ("tiles", "hashes")}
                 for r in results],
    }
    (OUT / "summary.json").write_text(json.dumps(payload, indent=2) + "\n",
                                      encoding="utf-8")
    print(f"[L81PassageSweepSummary] runs={len(results)} hash_mismatches={len(failures)} "
          f"summary={OUT / 'summary.json'}", flush=True)
    return 1 if failures else 0


if __name__ == "__main__":
    raise SystemExit(main())