VoxelForge/Standalone/Research/WorldAssemblerM3/Linux114/RunL114L77Census.py

#!/usr/bin/env python3
"""Compare L77 worker phase samples for L109 and L114 across four cameras."""
from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l114/l77-census"
DATA = ROOT / "Standalone/data/owner_world.json"
BINARIES = {
    "L109": ROOT / "Saved/Linux/l109/bin-final-profile/VoxelForgeCoreHeadless-l109-candidate-profile",
    "L114": ROOT / "Saved/Linux/l114/bin-profile/VoxelForgeCoreHeadless-l114-clang-l77",
}
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),
}
WORKER = re.compile(r"^\[WorldAssemblerLODWorker\] (.*)$", re.M)
SAMPLE = re.compile(r"^\[L77Sample\] (.*)$", re.M)
SAMPLER = re.compile(r"^\[L77Sampler\] action=stop (.*)$", re.M)
FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")


def run(version: str, strate: int, replicate: int) -> dict:
    OUT.mkdir(parents=True, exist_ok=True)
    label = f"{version.lower()}-s{strate}-r{replicate}"
    log = OUT / f"{label}.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", "--stream-scheduler-policy=3",
        "--linux77-profile",
    ]
    started = time.monotonic()
    with log.open("w", encoding="utf-8") as stream:
        result = subprocess.run(command, cwd=ROOT, stdout=stream,
                                stderr=subprocess.STDOUT, check=False, timeout=1740)
    content = log.read_text(encoding="utf-8", errors="replace")
    sampler_match = SAMPLER.search(content)
    worker_rows = [dict(FIELDS.findall(match.group(1))) for match in WORKER.finditer(content)]
    samples = [dict(FIELDS.findall(match.group(1))) for match in SAMPLE.finditer(content)]
    if result.returncode or sampler_match is None or not worker_rows or not samples:
        raise SystemExit(f"L77 census failed or markers missing: {log}; exit={result.returncode}")
    sampler = dict(FIELDS.findall(sampler_match.group(1)))
    if sampler.get("status") != "PASS":
        raise SystemExit(f"L77 sampler failed: {log}: {sampler}")

    scopes = {}
    for scope, lods in (("lod0_2", set(range(3))), ("whole_view", set(range(9)))):
        selected = [row for row in samples if int(row["lod"]) in lods]
        worker_cpu_s = sum(float(row["worker_sum_s"]) for row in worker_rows
                           if int(row["lod"]) in lods)
        phase_weights: dict[str, int] = {}
        stage_weights: dict[str, int] = {}
        for row in selected:
            weight = int(row.get("weight", "1"))
            phase = row.get("phase", "other")
            stage = row.get("stage", "other")
            phase_weights[phase] = phase_weights.get(phase, 0) + weight
            stage_weights[stage] = stage_weights.get(stage, 0) + weight
        total = sum(phase_weights.values())

        def estimate(weights: dict[str, int]) -> dict:
            return {key: {"share_pct": round(100.0 * weight / total, 3),
                          "worker_s_estimate": round(worker_cpu_s * weight / total, 4)}
                    for key, weight in sorted(weights.items(), key=lambda item: (-item[1], item[0]))}

        scopes[scope] = {
            "worker_cpu_s": round(worker_cpu_s, 4),
            "weighted_samples": total,
            "phase": estimate(phase_weights),
            "stage": estimate(stage_weights),
        }
    row = {
        "version": version, "strate": strate, "replicate": replicate,
        "exit_code": result.returncode, "sampler": sampler, "scopes": scopes,
        "log": str(log.relative_to(ROOT)),
        "runner_wall_s": round(time.monotonic() - started, 3),
    }
    print("[L114L77Run] " + json.dumps(row, sort_keys=True), flush=True)
    return row


def summarize(rows: list[dict]) -> dict:
    summary = []
    for strate in range(4):
        for scope in ("lod0_2", "whole_view"):
            groups = {version: [row for row in rows if row["strate"] == strate
                                and row["version"] == version]
                      for version in ("L109", "L114")}
            phases = sorted({phase for group in groups.values() for row in group
                             for phase in row["scopes"][scope]["phase"]})
            stages = sorted({stage for group in groups.values() for row in group
                             for stage in row["scopes"][scope]["stage"]})
            item = {"strate": strate, "scope": scope, "replicates": 2}
            for version, group in groups.items():
                item[f"{version}_worker_cpu_s_median"] = statistics.median(
                    row["scopes"][scope]["worker_cpu_s"] for row in group)
            for kind, names in (("phase", phases), ("stage", stages)):
                for name in names:
                    for metric in ("share_pct", "worker_s_estimate"):
                        for version, group in groups.items():
                            values = [row["scopes"][scope][kind].get(name, {})
                                      .get(metric, 0.0) for row in group]
                            item[f"{version}_{kind}_{name}_{metric}_median"] = \
                                statistics.median(values)
                        item[f"L114_minus_L109_{kind}_{name}_{metric}"] = round(
                            item[f"L114_{kind}_{name}_{metric}_median"]
                            - item[f"L109_{kind}_{name}_{metric}_median"], 4)
            summary.append(item)
    return {"runs": rows, "summary": summary}


def main() -> int:
    if any(not path.is_file() for path in (*BINARIES.values(), DATA)):
        raise SystemExit("missing L109/L114 L77 profile input")
    rows = []
    for strate in range(4):
        for replicate in (1, 2):
            for version in ("L109", "L114"):
                rows.append(run(version, strate, replicate))
                (OUT / "runs-partial.json").write_text(
                    json.dumps(rows, indent=2) + "\n", encoding="utf-8")
    (OUT / "summary.json").write_text(
        json.dumps(summarize(rows), indent=2) + "\n", encoding="utf-8")
    return 0


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