VoxelForge/Standalone/Research/WorldAssemblerM3/Linux105/AnalyzeL105Timeline.py

#!/usr/bin/env python3
"""Analyze L105 worker-idle phases and nested shape-build timers."""
from __future__ import annotations

import csv
import json
import re
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l105/timeline"
EVENT = re.compile(
    r"^\[CpuTileTimeline\] event=(\S+) key=\(([^,]+),([^,]+),([^,]+),lod(-?\d+)\) "
    r"at_ms=([0-9.]+).*?worker_id=(-?\d+) queue_depth=(\d+) active_workers=(\d+) "
    r"priority_rank=(\d+)", re.M)
COLD = re.compile(r"^\[ColdTimeline\] event=(\S+) edge=(\S+) at_ms=([0-9.]+)", re.M)
PIPELINE = re.compile(r"^\[StreamPipeline\] (.*)$", re.M)
NEAR = re.compile(r"^\[WorldAssemblerNearLODReady\] lods=0-2 tiles=(\d+)/(\d+) cold_ready_s=([0-9.]+)$", re.M)
FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")


def run(path: Path) -> dict:
    content = path.read_text(encoding="utf-8", errors="replace")
    events = []
    for match in EVENT.finditer(content):
        event, x, y, z, lod, ms, worker, queue, active, rank = match.groups()
        events.append({"event": event, "lod": int(lod), "at_ms": float(ms),
                       "worker": int(worker), "queue": int(queue),
                       "active": int(active), "rank": int(rank)})
    cold = {f"{event}_{edge}": float(ms) for event, edge, ms in COLD.findall(content)}
    idle_start: dict[int, float] = {}
    idle_intervals = []
    for event in sorted((item for item in events if item["event"] in ("idle_begin", "idle_end")),
                        key=lambda item: item["at_ms"]):
        if event["event"] == "idle_begin":
            idle_start[event["worker"]] = event["at_ms"]
        elif event["worker"] in idle_start:
            idle_intervals.append((idle_start.pop(event["worker"]), event["at_ms"]))
    near = NEAR.search(content)
    pipeline = PIPELINE.search(content)
    config = re.search(r"^\[HeadlessConfig\] (.*)$", content, re.M)
    if not events or near is None or pipeline is None or config is None:
        raise RuntimeError(f"incomplete timeline markers: {path}")
    config_fields = dict(FIELDS.findall(config.group(1)))
    workers = int(config_fields["workers"])
    whole_fields = dict(FIELDS.findall(pipeline.group(1)))
    cold_events = {f"{event}_{edge}": float(ms)
                   for event, edge, ms in COLD.findall(content)}
    phases = (
        ("full_pipeline", "desired_tile_set_requested", "lod0_2_done_pipeline_drain"),
        ("base_shape_compile", "base_compile_begin", "base_compile_end"),
        ("lod0_1_tile_drain", "lod0_1_admission_end", "lod0_1_done_publish_drained"),
        ("far_parent_build", "far_parent_build_begin", "far_parent_build_end"),
        ("lod2_queue_and_drain", "lod2_queue_begin", "lod2_done_pipeline_drain"),
    )
    phase_rows = []
    for label, begin, end in phases:
        start, finish = cold_events[begin], cold_events[end]
        idle_ms = sum(max(0.0, min(b, finish) - max(a, start))
                      for a, b in idle_intervals)
        capacity_ms = max(0.0, finish - start) * workers
        phase_rows.append({"phase": label, "wall_ms": finish - start,
                           "idle_worker_ms": idle_ms,
                           "idle_pct": 100.0 * idle_ms / capacity_ms if capacity_ms else 0.0,
                           "busy_pct": 100.0 * (1.0 - idle_ms / capacity_ms)
                               if capacity_ms else 0.0})

    def timer(prefix: str) -> dict:
        match = re.search(rf"^\[{prefix}\] (.*)$", content, re.M)
        return dict(FIELDS.findall(match.group(1))) if match else {}

    requested = [item for item in events if item["event"] == "requested"]
    admitted = [item for item in events if item["event"] == "admitted"]
    lod01_started = [item for item in events if item["event"] == "started" and item["lod"] <= 1]
    lod01_finished = [item for item in events if item["event"] == "finished" and item["lod"] <= 1]
    return {
        "log": str(path.relative_to(ROOT)), "workers": workers,
        "source_stamp": re.search(r"^\[VoxelForgeSourceStamp\] sha256=(\S+)$",
                                   content, re.M).group(1),
        "requested_lod01": sum(item["lod"] <= 1 for item in requested),
        "admitted_lod01": sum(item["lod"] <= 1 for item in admitted),
        "started_lod01": len(lod01_started), "finished_lod01": len(lod01_finished),
        "near_tiles": int(near.group(1)), "whole_tiles": whole_fields["whole_tiles"],
        "phase_timers": timer("ShapeBreakdown"),
        "far_phase_timers": timer("FarShapeBreakdown"),
        "worker_idle_phases": phase_rows,
    }


def main() -> int:
    logs = sorted(OUT.glob("*-w*-direct-measured-r1.log"))
    if len(logs) != 4:
        raise SystemExit(f"expected FI/TN W8/W12 traces, found {len(logs)}")
    rows = [run(path) for path in logs]
    (OUT / "analysis.json").write_text(json.dumps(rows, indent=2) + "\n", encoding="utf-8")
    with (OUT / "utilization.csv").open("w", newline="", encoding="utf-8") as stream:
        fields = ["log", "workers", "phase", "wall_ms", "idle_worker_ms", "idle_pct", "busy_pct"]
        writer = csv.DictWriter(stream, fieldnames=fields)
        writer.writeheader()
        for row in rows:
            for phase in row["worker_idle_phases"]:
                writer.writerow({"log": row["log"], "workers": row["workers"], **phase})
    print("[L105TimelineAnalysis] " + json.dumps(rows, sort_keys=True), flush=True)
    return 0


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