VoxelForge/Standalone/Research/WorldAssemblerM3/Linux70/SummarizeArrivalTracesL70.py

#!/usr/bin/env python3
"""Summarize L70 Unreal spawn traces beside exact-plan headless arrivals."""
from __future__ import annotations

import json
import re
import statistics
from pathlib import Path


ROOT = Path(__file__).resolve().parents[4]
SAVED = ROOT / "Saved/Linux/l70"
TRACES = SAVED / "traces"
HEADLESS = SAVED / "headless-arrivals"
OUT = SAVED / "arrival-metrics.json"


def parse_trace(strate: int, trial: int) -> dict:
    path = TRACES / f"strate{strate}-trial{trial}.trace.jsonl"
    events = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()
              if line.strip()]
    spawn = next(row for row in events if row.get("type") == "trace_spawn_location")
    ready_gate = next(row for row in events if row.get("type") == "walkable_gate"
                      and row.get("ready") == 1)
    ring = next(row for row in events if row.get("type") == "lod0_ring_ready")
    full = next(row for row in events if row.get("type") == "stream_trace_milestone"
                and row.get("name") == "initial_set_ready")
    summary = next(row for row in events if row.get("type") == "summary")
    tiles = [row for row in events if row.get("type") == "tile"]
    lod0 = summary["tile_percentiles"]["lod0"]["worker_task_s"]
    return {
        "label": f"strate{strate}-trial{trial}",
        "strate": strate,
        "trial": trial,
        "walkable_ring_s": ready_gate["t_s"] - spawn["t_s"],
        "walkable_ready_tiles": ready_gate.get("ready_tiles"),
        "walkable_required_tiles": ready_gate.get("required_tiles"),
        "lod0_ring_s": ring["t_s"] - spawn["t_s"],
        "lod0_tiles": ring.get("lod0_tiles"),
        "initial_set_s": full["t_s"] - spawn["t_s"],
        "initial_set_tiles": full.get("loaded"),
        "worker_seconds": sum(row.get("worker_task_s", 0.0) for row in tiles),
        "generate_tile_seconds": sum(row.get("core_generate_s", 0.0) for row in tiles),
        "mesh_copy_seconds": sum(row.get("mesh_copy_s", 0.0) for row in tiles),
        "rmc_seconds": sum(row.get("rmc_stream_build_s", 0.0) for row in tiles),
        "lod0_worker_p50_s": lod0["p50"],
        "lod0_worker_p95_s": lod0["p95"],
        "tile_events": len(tiles),
        "summary_tiles": summary["tile_percentiles"]["all"]["count"],
    }


def parse_headless(strate: int) -> dict:
    path = HEADLESS / f"strate{strate}.log"
    text = path.read_text(encoding="utf-8", errors="replace")
    milestone = re.search(r"\[PipelineMilestone\] first_tile_ready=([0-9.]+)s", text)
    stream = re.search(r"\[StreamPipeline\] ([^\n]+)", text)
    if not milestone or not stream:
        raise ValueError(f"missing pipeline timings in {path}")
    fields = dict(re.findall(r"([A-Za-z0-9_]+)=([^ ]+)", stream.group(1)))
    worker_rows = re.findall(
        r"\[WorldAssemblerLODWorker\] lod=(\d+) tiles=(\d+) worker_sum_s=([0-9.]+)"
        r".*?mc_exclusive_ms=([0-9.]+).*?mesh_assembly_ms=([0-9.]+)", text)
    return {
        "strate": strate,
        "first_tile_s": float(milestone.group(1)),
        "neighborhood_ready_s": float(fields["neighborhood_ready"].rstrip("s")),
        "full_detail_ready_s": float(fields["full_detail_ready"].rstrip("s")),
        "whole_view_ready_s": float(fields["whole_view_ready"].rstrip("s")),
        "neighborhood_tiles": fields["neighborhood_tiles"],
        "full_detail_tiles": fields["full_detail_tiles"],
        "whole_tiles": fields["whole_tiles"],
        "worker_seconds": sum(float(row[2]) for row in worker_rows),
        "worker_lods": len(worker_rows),
        "mc_exclusive_seconds": sum(float(row[3]) for row in worker_rows) / 1000.0,
        "mesh_assembly_seconds": sum(float(row[4]) for row in worker_rows) / 1000.0,
    }


def median(rows: list[dict], key: str) -> float:
    return statistics.median(float(row[key]) for row in rows)


def main() -> int:
    results = []
    for strate in range(4):
        trials = [parse_trace(strate, trial) for trial in range(1, 4)]
        headless = parse_headless(strate)
        engine = {key: median(trials, key) for key in (
            "walkable_ring_s", "lod0_ring_s", "initial_set_s", "worker_seconds",
            "generate_tile_seconds", "mesh_copy_seconds", "rmc_seconds",
            "lod0_worker_p50_s", "lod0_worker_p95_s")}
        engine.update({"walkable_ready_tiles": trials[0]["walkable_ready_tiles"],
                       "walkable_required_tiles": trials[0]["walkable_required_tiles"],
                       "lod0_tiles": trials[0]["lod0_tiles"],
                       "initial_set_tiles": trials[0]["initial_set_tiles"],
                       "trace_tile_events": trials[0]["tile_events"],
                       "trace_summary_tiles": trials[0]["summary_tiles"]})
        row = {"strate": strate, "engine_median": engine, "engine_trials": trials,
               "headless": headless,
               "initial_set_time_gap_percent":
                   100.0 * (engine["initial_set_s"] / headless["whole_view_ready_s"] - 1.0),
               "worker_sum_gap_percent":
                   100.0 * (engine["worker_seconds"] / headless["worker_seconds"] - 1.0)}
        results.append(row)
        print(
            f"strate={strate} walk={engine['walkable_ring_s']:.3f}s "
            f"lod0={engine['lod0_ring_s']:.3f}s/{engine['lod0_tiles']} tiles "
            f"full=UE {engine['initial_set_s']:.3f}s/{engine['initial_set_tiles']} "
            f"vs headless {headless['whole_view_ready_s']:.3f}s/{headless['whole_tiles']} "
            f"workers=UE {engine['worker_seconds']:.3f}s/{engine['trace_tile_events']} "
            f"vs headless {headless['worker_seconds']:.3f}s "
            f"LOD0 p50/p95={engine['lod0_worker_p50_s']:.6f}/{engine['lod0_worker_p95_s']:.6f}s "
            f"GenerateTile/mesh-copy/RMC={engine['generate_tile_seconds']:.3f}/"
            f"{engine['mesh_copy_seconds']:.3f}/{engine['rmc_seconds']:.3f}s "
            f"gaps={row['initial_set_time_gap_percent']:+.1f}% time,"
            f"{row['worker_sum_gap_percent']:+.1f}% worker")
    OUT.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8")
    print(f"[L70ArrivalMetrics] wrote {OUT}")
    return 0


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