VoxelForge/Standalone/Research/WorldAssemblerM3/Linux123/RunL123Pairs.py

#!/usr/bin/env python3
"""Five interleaved 30 m/s UE pairs for the Core region lookahead experiment."""
from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[4]
HOST = Path("/home/jahni/VFHost")
EDITOR = Path("/home/jahni/UnrealEngine/Engine/Binaries/Linux/UnrealEditor-Cmd")
SPAWN_TN_M = (19.166015625, -14.2001953125, -753.3740234375)
SPAWN_FI_M = (25.8369140625, 3.19921875, -737.0)


def percentile(values: list[float], fraction: float) -> float | None:
    if not values:
        return None
    ordered = sorted(values)
    return ordered[max(0, min(len(ordered) - 1, math.ceil(len(ordered) * fraction) - 1))]


def read_events(path: Path) -> list[dict]:
    rows = []
    if path.is_file():
        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
            if not line.startswith("{"):
                continue
            try:
                rows.append(json.loads(line))
            except json.JSONDecodeError:
                pass
    return rows


def run(scenario: str, condition: str, repetition: int, turn: bool = False) -> dict:
    lookahead = int(condition in ("lookahead", "turn_lookahead"))
    if scenario == "movement":
        speed, start, duration, exit_s, slot, spawn = (
            3000.0, 6.0, 12.0, 23.0, 0, SPAWN_TN_M)
    else:
        speed, start, duration, exit_s, slot, spawn = (
            140.0, 20.0, 12.0, 37.0, 1, SPAWN_FI_M)

    label = f"{scenario}-{condition}-r{repetition}"
    output_group = "turn" if turn else scenario
    out = ROOT / "Saved/Linux/l123" / output_group / condition / "unreal"
    host_out = HOST / "Plugins/VoxelForge/Saved/Linux/l123" / output_group / condition
    host_saved = HOST / "Saved/Linux/l123"
    trace = out / f"{label}.trace.jsonl"
    abslog = out / f"{label}.log"
    stdout = out / f"{label}.stdout.log"
    export = host_out / "output" / label
    userdir = host_saved / "user" / label
    zendata = host_saved / "zen" / label
    ddc = host_saved / "DDC"
    for path in (out, export, userdir, zendata, ddc):
        path.mkdir(parents=True, exist_ok=True)
    trace.unlink(missing_ok=True)
    spawn_cm = ",".join(f"{value * 100.0:.3f}" for value in spawn)
    command = [
        str(EDITOR), str(HOST / "VFHost.uproject"), "-nullrhi", "-unattended",
        "-nop4", "-nosplash", "-NoSound", "-corelimit=12", "-NoZenAutoLaunch",
        "-notraceserver", "-nocrashreports", "-DisablePlugins=AndroidFileServer",
        "-ddc=NoZenLocalFallback", f"-userdir={userdir}", f"-ZenDataPath={zendata}",
        f"-LocalDataCachePath={ddc}", "-run=VoxelForgeExplore", "-seed=0",
        "-archetype=TunnelNetwork", f"-slot={slot}", "-modes=export", "-gameconfig=1",
        "-coreworldtest=1", "-ownerfourstrateplan=1", "-corevoxelsizecm=25",
        "-coreislandtest=0", "-corefallbacktest=0", "-coreofftest=0", "-coreplantest=0",
        "-settingsref=/Game/VoxelForge/DA_Settings.DA_Settings", "-corehybridmesher=0",
        f"-out={export}", "-voxel.StreamingCoarseFirst=1",
        "-voxel.StreamingDynamicNearReserve=1", "-voxel.StreamingLegacyPendingResultCap=0",
        "-voxel.CoreRegionConcurrentAdmission=1", "-voxel.StreamingResultBacklogCap=256",
        "-voxel.StreamingReadyQueueDepth=8", "-voxel.StreamingBackgroundWorkerCap=1",
        "-voxel.CoreColumnCacheDirectMapped=0", "-voxel.StreamingFarApplySupplement=1",
        "-voxel.CoreMCGradientMemoization=1", "-voxel.StreamingCollisionLookahead=1",
        f"-voxel.CoreRegionLookahead={lookahead}", "-voxel.TestMove=1",
        f"-voxel.TestMoveSpeedCmPerSecond={speed:.1f}",
        f"-voxel.TestMoveStartSeconds={start:.1f}",
        f"-voxel.TestMoveDurationSeconds={duration:.1f}",
        f"-voxel.TestExitSeconds={exit_s:.1f}", "-voxel.TestMoveDirection=1,0,0",
        "-voxel.StreamingSchedulerPolicy=3", "-voxel.TestExitOnSteady=0",
        "-coreworldstreamtrace=1", f"-voxel.StartupTraceSpawnCm={spawn_cm}",
        "-voxel.StartupTraceUntilEndPlay=1", "-voxel.StartupTraceThroughCrossing=1",
        "-voxel.StartupTraceMaxTiles=16384", f"-voxel.StartupTraceFile={trace}",
        f"-abslog={abslog}",
    ]
    if turn:
        command.extend(["-voxel.TestMoveTurnAfterSeconds=6.0",
                        "-voxel.TestMoveTurnDirection=0,1,0"])
    started = time.monotonic()
    wrapped = ["taskset", "-c", "0-11", "timeout", "1740s", *command]
    with stdout.open("w", encoding="utf-8") as stream:
        process = subprocess.run(wrapped, cwd=ROOT, stdout=stream,
                                  stderr=subprocess.STDOUT, check=False, timeout=1740)
    events = read_events(trace)
    samples = [event for event in events if event.get("type") == "streaming_sample"]
    moving = [event for event in samples if float(event.get("speed_cm_s", 0.0)) >= 50.0]
    stop_event = next((event for event in events
                       if event.get("type") == "streaming_test_move_stopped"), {})
    stop_s = float(stop_event.get("t_s", 0.0))
    move_start = min((float(event.get("t_s", 0.0)) for event in moving), default=float("inf"))
    dwell = [event for event in samples if float(event.get("t_s", 0.0)) >= stop_s
             and float(event.get("speed_cm_s", 0.0)) < 50.0]
    backend = next((event for event in events if event.get("type") == "worker_backend_config"), {})
    cap = next((event for event in events if event.get("type") == "streaming_worker_cap_readback"), {})
    if process.returncode or len(moving) < 40 or len(dwell) < 16:
        raise RuntimeError(f"incomplete UE run {label}: exit={process.returncode}, "
                           f"move={len(moving)}, dwell={len(dwell)}, trace={trace}")
    if (int(backend.get("core_region_lookahead", -1)) != lookahead
            or int(backend.get("streaming_far_apply_supplement", -1)) != 1
            or int(cap.get("streaming_background_worker_cap", -1)) != 1
            or int(backend.get("effective_tile_slots", -1)) != 8):
        raise RuntimeError(f"configuration readback mismatch: {backend}; {cap}")

    moving_rows = []
    for event in moving:
        desired = event.get("desired_by_lod", [0] * 9)
        resident = event.get("resident_by_lod", [0] * 9)
        covered = event.get("covered_by_lod", resident)
        moving_rows.append({
            "near_ready_pct": 100.0 * sum(resident[:3]) / max(1, sum(desired[:3])),
            "lod0_ready_pct": 100.0 * resident[0] / max(1, desired[0]),
            "lod1_ready_pct": 100.0 * resident[1] / max(1, desired[1]),
            "lod2_ready_pct": 100.0 * resident[2] / max(1, desired[2]),
            "far_hole_pct": 100.0 * (sum(desired[3:]) - sum(covered[3:]))
            / max(1, sum(desired[3:])),
        })
    work = [event for event in events if event.get("type") == "game_thread_work_frame"
            and move_start <= float(event.get("t_s", 0.0)) < stop_s]
    ticks = [float(event.get("tick_ms", 0.0)) for event in work]
    updates = [float(event.get("update_chunks_ms", 0.0)) for event in work]
    arrival = [event for event in samples if float(event.get("t_s", 0.0)) < start
               and float(event.get("speed_cm_s", 0.0)) < 50.0]

    def settled(event: dict, end: int) -> bool:
        desired = event.get("desired_by_lod", [0] * 9)
        actual = event.get("settled_by_lod", [0] * 9)
        return sum(desired[:end]) > 0 and sum(actual[:end]) >= sum(desired[:end])

    def far_holes(event: dict) -> float | None:
        desired = event.get("desired_by_lod", [0] * 9)[3:]
        covered = event.get("covered_by_lod", [0] * 9)[3:]
        return 100.0 * (sum(desired) - sum(covered)) / sum(desired) if sum(desired) else None

    near_settled = next((event for event in arrival if settled(event, 3)), None)
    whole_settled = next((event for event in arrival if settled(event, 9)), None)
    far_clear = next((event for event in arrival if far_holes(event) == 0.0), None)
    compile_done = [event for event in events if event.get("type") == "core_region_compile_complete"
                    and move_start <= float(event.get("t_s", 0.0)) < stop_s]
    compile_requests = [event for event in events
                        if event.get("type") == "core_region_compile_request"
                        and move_start <= float(event.get("t_s", 0.0)) < stop_s]
    lookahead_turns = [event for event in events
                       if event.get("type") == "core_region_lookahead_turn"]
    row = {
        "scenario": scenario, "condition": condition, "repetition": repetition,
        "speed_mps": speed / 100.0, "core_region_lookahead": lookahead,
        "exit_code": process.returncode, "wall_s": round(time.monotonic() - started, 3),
        "moving_samples": len(moving), "dwell_samples": len(dwell),
        "near_ready_moving_mean_pct": statistics.mean(r["near_ready_pct"] for r in moving_rows),
        "lod0_ready_moving_mean_pct": statistics.mean(r["lod0_ready_pct"] for r in moving_rows),
        "lod1_ready_moving_mean_pct": statistics.mean(r["lod1_ready_pct"] for r in moving_rows),
        "lod2_ready_moving_mean_pct": statistics.mean(r["lod2_ready_pct"] for r in moving_rows),
        "far_holes_moving_mean_pct": statistics.mean(r["far_hole_pct"] for r in moving_rows),
        "game_thread_tick_p99_ms": percentile(ticks, 0.99),
        "game_thread_tick_max_ms": max(ticks, default=None),
        "stream_update_p99_ms": percentile(updates, 0.99),
        "stream_update_max_ms": max(updates, default=None),
        "lod02_settled_t_s": float(near_settled["t_s"]) if near_settled else None,
        "whole_view_settled_t_s": float(whole_settled["t_s"]) if whole_settled else None,
        "far_clear_t_s": float(far_clear["t_s"]) if far_clear else None,
        "far_holes_at_pre_move_pct": far_holes(arrival[-1]) if arrival else None,
        "pending_gate_p50_ms": percentile(
            [float(e.get("pending_gate_elapsed_ms", 0.0)) for e in compile_done], 0.50),
        "compile_queue_p95_ms": percentile(
            [float(e.get("compile_queue_ms", 0.0)) for e in compile_done], 0.95),
        "compile_task_p95_ms": percentile(
            [float(e.get("compile_task_ms", 0.0)) for e in compile_done], 0.95),
        "compiled_tiles_total": sum(int(e.get("compiled_tiles", 0)) for e in compile_done),
        "lookahead_tiles_peak": max((int(e.get("lookahead_region_tiles", 0))
                                      for e in compile_requests), default=0),
        "wasted_prefetch_tiles": sum(int(e.get("wasted_prefetch_tiles", 0))
                                      for e in compile_done),
        "lookahead_turn_events": lookahead_turns,
        "compile_completion_count": len(compile_done),
        "trace": str(trace.relative_to(ROOT)), "log": str(abslog.relative_to(ROOT)),
    }
    with (ROOT / "Saved/Linux/l123" / "runs.jsonl").open("a", encoding="utf-8") as stream:
        stream.write(json.dumps(row, sort_keys=True) + "\n")
    print("[L123Pair] " + json.dumps(row, sort_keys=True), flush=True)
    return row


def pairs_for(runs: list[dict], scenario: str) -> list[dict]:
    pairs = []
    for repetition in range(1, 6):
        baseline = next(row for row in runs if row["scenario"] == scenario
                        and row["condition"] == "baseline" and row["repetition"] == repetition)
        candidate = next(row for row in runs if row["scenario"] == scenario
                         and row["condition"] == "lookahead" and row["repetition"] == repetition)
        pairs.append({"repetition": repetition,
                      "near_ready_delta_pct": candidate["near_ready_moving_mean_pct"]
                      - baseline["near_ready_moving_mean_pct"],
                      "far_holes_delta_pct": candidate["far_holes_moving_mean_pct"]
                      - baseline["far_holes_moving_mean_pct"],
                      "tick_p99_delta_ms": candidate["game_thread_tick_p99_ms"]
                      - baseline["game_thread_tick_p99_ms"],
                      "tick_max_delta_ms": candidate["game_thread_tick_max_ms"]
                      - baseline["game_thread_tick_max_ms"]})
    return pairs


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--scenario", choices=("movement", "arrival"), required=True)
    parser.add_argument("--turn-diagnostic", action="store_true")
    args = parser.parse_args()
    if not EDITOR.is_file():
        raise SystemExit(f"missing Unreal editor: {EDITOR}")
    root_out = ROOT / "Saved/Linux/l123" / args.scenario
    root_out.mkdir(parents=True, exist_ok=True)
    (ROOT / "Saved/Linux/l123").mkdir(parents=True, exist_ok=True)
    if args.turn_diagnostic:
        row = run("movement", "turn_lookahead", 1, turn=True)
        turn_out = ROOT / "Saved/Linux/l123/turn"
        turn_out.mkdir(parents=True, exist_ok=True)
        (turn_out / "summary.json").write_text(
            json.dumps({"run": row}, indent=2) + "\n", encoding="utf-8")
        return 0
    runs = []
    for repetition in range(1, 6):
        order = ("baseline", "lookahead") if repetition % 2 else ("lookahead", "baseline")
        for condition in order:
            runs.append(run(args.scenario, condition, repetition))
            (root_out / "runs-partial.json").write_text(
                json.dumps(runs, indent=2) + "\n", encoding="utf-8")
    pairs = pairs_for(runs, args.scenario)
    summary = {
        "design": "Compile/retain the current near region set plus one step along velocity; "
                  "return completion through an Epoch-tagged MPSC queue.",
        "scenario": args.scenario, "runs": runs, "pairs": pairs,
        "paired_median_near_ready_delta_pct": statistics.median(
            pair["near_ready_delta_pct"] for pair in pairs),
        "paired_median_far_holes_delta_pct": statistics.median(
            pair["far_holes_delta_pct"] for pair in pairs),
        "paired_median_tick_p99_delta_ms": statistics.median(
            pair["tick_p99_delta_ms"] for pair in pairs),
        "paired_median_tick_max_delta_ms": statistics.median(
            pair["tick_max_delta_ms"] for pair in pairs),
    }
    (root_out / "summary.json").write_text(json.dumps(summary, indent=2) + "\n",
                                           encoding="utf-8")
    return 0


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