VoxelForge/Experiments/LivingPlane/check_repeatability.py

#!/usr/bin/env python3
"""Compare final field checksums across different query and capture settings."""

from __future__ import annotations

import argparse
import json
import pathlib
import shutil
import subprocess
import sys


PLUGIN_ROOT = pathlib.Path(__file__).resolve().parents[2]


def read_checksum(path: pathlib.Path) -> str:
    try:
        with path.open(encoding="utf-8") as source:
            result = json.load(source)
    except (OSError, json.JSONDecodeError) as error:
        raise RuntimeError(f"could not read {path}: {error}") from error
    checksum = result.get("field_checksum_fnv1a64")
    if not isinstance(checksum, str) or len(checksum) != 16:
        raise RuntimeError(f"{path} has no 16-digit field checksum")
    return checksum


def compare(left: pathlib.Path, right: pathlib.Path) -> bool:
    left_checksum = read_checksum(left)
    right_checksum = read_checksum(right)
    if left_checksum != right_checksum:
        print(
            "FAIL field checksum mismatch: "
            f"{left}={left_checksum}, {right}={right_checksum}",
            file=sys.stderr,
        )
        return False
    print(f"PASS field checksum matches: {left_checksum}")
    return True


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--binary",
        type=pathlib.Path,
        default=PLUGIN_ROOT / "Saved/Linux/l164/living-plane",
    )
    parser.add_argument(
        "--out",
        type=pathlib.Path,
        default=PLUGIN_ROOT / "Saved/Linux/l164/repeatability",
    )
    parser.add_argument("--device", choices=("auto", "cpu", "gpu"), default="cpu")
    parser.add_argument("--size", type=int, default=512)
    parser.add_argument("--steps", type=int, default=1800)
    parser.add_argument(
        "--compare-only",
        nargs=2,
        type=pathlib.Path,
        metavar=("LEFT_JSON", "RIGHT_JSON"),
        help="compare existing timing.json files; used to exercise the failure path",
    )
    options = parser.parse_args()

    if options.compare_only:
        return 0 if compare(*options.compare_only) else 1

    if not options.binary.is_file():
        parser.error(f"binary not found: {options.binary}; run Experiments/LivingPlane/build.sh")

    options.out.mkdir(parents=True, exist_ok=True)
    off_dir = options.out / "queries-off_capture-off"
    on_dir = options.out / "queries-on_capture-every-900"
    for directory in (off_dir, on_dir):
        if directory.exists():
            shutil.rmtree(directory)

    common = [
        f"--device={options.device}",
        "--scene=alive",
        f"--size={options.size}",
        f"--steps={options.steps}",
    ]
    runs = (
        ("--capture-every=0", "--queries=off", off_dir),
        ("--capture-every=900", "--queries=on", on_dir),
    )
    for capture, queries, output in runs:
        command = [
            "taskset",
            "-c",
            "0-11",
            "prlimit",
            "--as=6442450944",
            "--",
            "timeout",
            "1740s",
            str(options.binary.resolve()),
            *common,
            capture,
            queries,
            f"--out={output}",
        ]
        print("RUN " + " ".join(command), flush=True)
        log_path = options.out / f"{output.name}.log"
        with log_path.open("w", encoding="utf-8") as log_file:
            try:
                subprocess.run(
                    command,
                    cwd=PLUGIN_ROOT,
                    check=True,
                    stdout=log_file,
                    stderr=subprocess.STDOUT,
                )
            except subprocess.CalledProcessError:
                print(log_path.read_text(encoding="utf-8"), file=sys.stderr)
                raise
        print(log_path.read_text(encoding="utf-8").rstrip())

    left = off_dir / "timing.json"
    right = on_dir / "timing.json"
    return 0 if compare(left, right) else 1


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, RuntimeError, subprocess.CalledProcessError) as error:
        print(f"FAIL repeatability check: {error}", file=sys.stderr)
        raise SystemExit(1)