VoxelForge/Standalone/Research/WorldAssemblerM3/Linux136/RunL136Census.py

#!/usr/bin/env python3
"""Compare assembler/world and v236 composer/classifier phase counters."""
from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l136/census"
DATA = ROOT / "Standalone/data/owner_world.json"
BINARIES = {
    "assembler-world": ROOT / "Saved/Linux/l130/bin/VoxelForgeCoreHeadless-l130-clang",
    "v236": ROOT / "Saved/Linux/l135/bin/VoxelForgeCoreHeadless-clang-l135",
    "v239": ROOT / "Saved/Linux/l136/bin/VoxelForgeCoreHeadless-l136-clang",
}
STAMPS = {
    "assembler-world": "8b3e7bfb4cde2b408ab940925eab973d8375d6b719774d6f72f64ee72b526c35",
    "v236": "8dc4294ce0250a591ee4515471c3c547e796ffa4346ab44f0ae1740bbc3797f5",
    "v239": "ebe6b32a7cd96c5aa69c6dc64ae6b9c6ff4bc18bc484efc5148611c48723c28f",
}
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),
}
FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")


def field(line: str) -> dict[str, str]:
    return dict(FIELDS.findall(line))


def run(strate: int, version: str) -> dict:
    OUT.mkdir(parents=True, exist_ok=True)
    log = OUT / f"{version}-s{strate}.log"
    camera = ",".join(f"{value:.12f}" for value in CAMERAS[strate]) + ",0,0"
    command = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
        "timeout", "1740s", str(BINARIES[version]), str(DATA),
        "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
        "--workers=8", "--clip-radius=3", "--max-clip-level=8", "--max-level=8",
        f"--camera={camera}", "--profile-lod-work", "--profile-passage-composer",
        "--stream-scheduler-policy=3"]
    started = 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)
    content = log.read_text(encoding="utf-8", errors="replace")
    stamp_match = re.search(r"^\[VoxelForgeSourceStamp\] sha256=([0-9a-f]+)$", content, re.M)
    if result.returncode or stamp_match is None or stamp_match.group(1) != STAMPS[version]:
        raise SystemExit(f"census failed/stamp mismatch: {log}, exit={result.returncode}, stamp={stamp_match}")
    lod: dict[int, dict] = defaultdict(lambda: defaultdict(float))
    for line in content.splitlines():
        if line.startswith("[WorldAssemblerLODClassification] "):
            values = field(line)
            target = lod[int(values["lod"])]
            for name in ("tiles", "all_air", "all_solid", "mixed", "unclassified",
                         "tile_skips", "density_point_calls", "field_eval_sample_points",
                         "proof_air_cells", "proof_solid_cells"):
                target[name] = int(values.get(name, "0"))
            target["field_eval_ms_est"] = float(values.get("field_eval_ms_est", "0"))
        elif line.startswith("[PassageComposerTile] "):
            values = field(line)
            level = int(re.search(r"lod(\d+)\)", values["key"]).group(1))
            target = lod[level]
            target["composer_tiles"] += 1
            for name in ("samples", "field_calls", "native_cave_samples", "native_field_calls",
                         "room_candidates", "edge_candidates", "features", "route_candidates",
                         "passage_iterations", "carve_segment_visits"):
                target[name] += int(values.get(name, "0"))
            target["composer_worker_ms"] += float(values.get("worker_ms", "0"))
    plan = re.search(r"^\[WorldAssemblerPlanTiming\] total_ms=([0-9.]+)", content, re.M)
    cave = re.search(r"^\[CavernPlannerV1\].* plan_ms=([0-9.]+)", content, re.M)
    pipeline = re.search(r"^\[WorldAssemblerTilePipeline\] (.*)$", content, re.M)
    pipeline_fields = field(pipeline.group(1)) if pipeline else {}
    result_row = {
        "strate": strate, "version": version, "source_stamp": stamp_match.group(1),
        "plan_ms": float(plan.group(1)) if plan else None,
        "cave_plan_ms": float(cave.group(1)) if cave else None,
        "tile_pipeline": {key: (float(value) if "." in value else int(value))
                          for key, value in pipeline_fields.items()
                          if re.fullmatch(r"-?\d+(?:\.\d+)?", value)},
        "lod": {str(key): dict(values) for key, values in sorted(lod.items())},
        "wall_s": round(time.monotonic() - started, 3),
        "log": str(log.relative_to(ROOT)),
    }
    print("[L136CensusRun] " + json.dumps(result_row, sort_keys=True), flush=True)
    return result_row


def main() -> int:
    if any(not path.is_file() for path in (DATA, *BINARIES.values())):
        raise SystemExit("missing census fixture or binary")
    rows = []
    for strate in range(4):
        for version in BINARIES:
            rows.append(run(strate, version))
            (OUT / "runs-partial.json").write_text(json.dumps(rows, indent=2) + "\n")
    summary = {"status": "PASS", "rows": rows,
        "notes": [
            "classifier milliseconds and composer worker milliseconds are cumulative counters, not exclusive wall time",
            "assembler/world predates native cave field counters; field sample fields are only comparable where emitted",
        ]}
    (OUT / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
    return 0


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