VoxelForge/Standalone/Research/WorldAssemblerM3/Linux74/RunCeilingL74.py

#!/usr/bin/env python3
"""Collect full-grid sign-uniformity ceilings for the four owner arrivals."""
from __future__ import annotations

import json
import re
import subprocess
import sys
import time
from collections import defaultdict
from pathlib import Path


ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l74/logs"
DATA = ROOT / "Standalone/data/owner_world.json"
BIN = ROOT / "Saved/Linux/l74/bin/VoxelForgeCoreHeadless-l74-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),
}
ROW = re.compile(r"\[UniformGridCeiling\] (.*)")
PAIR = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")


def parse_log(path: Path, view: int) -> list[dict]:
    rows = []
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        match = ROW.search(line)
        if not match:
            continue
        row = dict(PAIR.findall(match.group(1)))
        key = tuple(int(value) for value in row["key"].strip("()").split(","))
        row.update(view=view, lod=int(row["lod"]), strate=int(row["strate"]),
                   key=key, samples=int(row["samples"]), uniform=int(row["uniform"]),
                   air=int(row["air"]), blocks8=int(row["blocks8"]),
                   samples8=int(row["samples8"]), blocks4=int(row["blocks4"]),
                   samples4=int(row["samples4"]))
        rows.append(row)
    return rows


def run(view: int) -> list[dict]:
    camera = ",".join(f"{value:.12f}" for value in CAMERAS[view]) + ",0,0"
    log = OUT / f"ceiling-view{view}.log"
    command = [
        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
        "timeout", "1740s", str(BIN), str(DATA), "--world-assembler-prototype",
        "--stream-benchmark", "--stream-pipeline", "--stream-hash-tiles",
        "--force-generate", "--workers=8", "--clip-radius=3", "--max-clip-level=8",
        "--max-level=8", f"--camera={camera}",
    ]
    start = time.monotonic()
    with log.open("w", encoding="utf-8") as output:
        result = subprocess.run(command, cwd=ROOT, stdout=output, stderr=subprocess.STDOUT,
                                check=False, timeout=1740)
    rows = parse_log(log, view)
    print(json.dumps({"view": view, "exit_code": result.returncode,
                      "wall_s": round(time.monotonic() - start, 3),
                      "tiles": len(rows), "log": str(log.relative_to(ROOT))}, sort_keys=True),
          flush=True)
    if result.returncode != 0 or not rows:
        raise SystemExit(f"ceiling view {view} failed or emitted no rows; see {log}")
    return rows


def main() -> int:
    if not DATA.is_file() or not BIN.is_file():
        raise SystemExit("missing owner world or L74 headless binary")
    OUT.mkdir(parents=True, exist_ok=True)
    if sys.argv[1:] == ["--analyze-only"]:
        rows_path = OUT / "ceiling-rows.json"
        if not rows_path.is_file():
            raise SystemExit("no recorded ceiling rows to analyze")
        all_rows = json.loads(rows_path.read_text(encoding="utf-8"))
    elif len(sys.argv) == 1:
        all_rows = [row for view in range(4) for row in run(view)]
    else:
        raise SystemExit("usage: RunCeilingL74.py [--analyze-only]")
    unique = {}
    for row in all_rows:
        row["key"] = tuple(row["key"])
        # A tile's compiled field can depend on whether the whole-view build contains it
        # in its fine region, so retain each of the four camera-context observations.
        identity = (row["view"], row["strate"], row["lod"], row["key"])
        previous = unique.get(identity)
        if previous is None or (previous["samples"] == 0 and row["samples"] > 0):
            unique[identity] = row
        elif previous["samples"] > 0 and row["samples"] > 0:
            signature = ("samples", "uniform", "air", "samples8", "samples4")
            if any(previous[field] != row[field] for field in signature):
                raise SystemExit(f"conflicting full-grid ceiling rows for tile {identity}")
    groups: dict[tuple[int, int], list[dict]] = defaultdict(list)
    for row in unique.values():
        groups[(row["strate"], row["lod"])].append(row)
    summary = []
    for (strate, lod), rows in sorted(groups.items()):
        tiles = len(rows)
        sampled_rows = [row for row in rows if row["samples"] > 0]
        sampled_tiles = len(sampled_rows)
        samples = sum(row["samples"] for row in sampled_rows)
        uniform_tiles = sum(row["uniform"] for row in sampled_rows)
        uniform_samples = sum(row["samples"] for row in rows if row["uniform"])
        row = {
            "strate": strate, "lod": lod, "tiles_visited": tiles,
            "sampled_tiles": sampled_tiles, "grid_samples": samples,
            "uniform_tiles": uniform_tiles,
            "uniform_tile_pct": round(100 * uniform_tiles / sampled_tiles, 2)
                if sampled_tiles else 0,
            "uniform_tile_pct_of_visited": round(100 * uniform_tiles / tiles, 2)
                if tiles else 0,
            "uniform_air_tiles": sum(row["uniform"] and row["air"] for row in sampled_rows),
            "uniform_solid_tiles": sum(row["uniform"] and not row["air"]
                                        for row in sampled_rows),
            "samples_in_uniform_tiles_pct": round(100 * uniform_samples / samples, 2)
                if samples else 0,
            "uniform_8cube_samples_pct": round(100 * sum(row["samples8"] for row in rows)
                / samples, 2) if samples else 0,
            "uniform_4cube_samples_pct": round(100 * sum(row["samples4"] for row in rows)
                / samples, 2) if samples else 0,
        }
        summary.append(row)
        print("[L74Ceiling] " + json.dumps(row, sort_keys=True), flush=True)
    (OUT / "ceiling-rows.json").write_text(json.dumps(all_rows, indent=2) + "\n",
                                             encoding="utf-8")
    (OUT / "ceiling-summary.json").write_text(json.dumps(summary, indent=2) + "\n",
                                                encoding="utf-8")
    return 0


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