#!/usr/bin/env python3 """Tiny live dashboard for Gemma 4 baseline-vs-MTP benchmark runs.""" from __future__ import annotations import argparse import json import mimetypes import os import re import time import urllib.error import urllib.parse import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any METRIC_RE = re.compile(r"^(sglang:\w+)(?:\{([^}]*)\})?\s+([\d.eE+-]+)") HTML = r""" Gemma 4 MTP Derby

Gemma 4 MTP Derby

Baseline and MTP run the same matrix, one clean server at a time. The track shows live throughput while the steward table keeps the final JSON honest.

Waiting
Race Track No active cell
Live Board --
Photo Finish Waiting for results
Last Log Lines baseline / mtp 

      

    

  


  


"""


def load_json(path: Path) -> Any:
    try:
        with path.open() as f:
            return json.load(f)
    except Exception:
        return None


def tail_text(path: Path, lines: int = 90) -> str:
    try:
        with path.open(errors="replace") as f:
            data = f.readlines()
        return "".join(data[-lines:])
    except Exception:
        return ""


def scrape_metrics(sglang_url: str) -> dict[str, float]:
    metrics: dict[str, float] = {}
    if not sglang_url:
        return metrics

    url = urllib.parse.urljoin(sglang_url.rstrip("/") + "/", "metrics")
    try:
        with urllib.request.urlopen(url, timeout=1.5) as resp:
            body = resp.read().decode("utf-8", errors="replace")
    except (urllib.error.URLError, TimeoutError, OSError):
        return metrics

    for line in body.splitlines():
        if line.startswith("#"):
            continue
        match = METRIC_RE.match(line)
        if not match:
            continue
        name, labels, value = match.group(1), match.group(2) or "", float(match.group(3))
        if "tp_rank=" in labels and 'tp_rank="0"' not in labels:
            continue
        if name not in metrics:
            metrics[name] = value
    return metrics


def live_is_recent(doc: dict[str, Any] | None, max_age_seconds: float = 12.0) -> bool:
    if not doc or not doc.get("cell_running"):
        return False
    try:
        updated = doc.get("updated_at", "")
        timestamp = time.mktime(time.strptime(updated[:19], "%Y-%m-%dT%H:%M:%S"))
        return (time.time() - timestamp) <= max_age_seconds
    except Exception:
        return True


def build_snapshot(results_dir: Path, sglang_url: str) -> dict[str, Any]:
    baseline_live = load_json(results_dir / "baseline.live.json")
    mtp_live = load_json(results_dir / "mtp_topk1.live.json")
    baseline_result = load_json(results_dir / "baseline.json")
    mtp_result = load_json(results_dir / "mtp_topk1.json")
    summary = load_json(results_dir / "summary.json")

    live_docs = {
        "baseline": baseline_live,
        "mtp_topk1": mtp_live,
    }
    active_case = "mtp_topk1" if live_is_recent(mtp_live) else "baseline"
    if not live_is_recent(live_docs.get(active_case)):
        active_case = "baseline" if live_is_recent(baseline_live) else "mtp_topk1" if mtp_live else "baseline"

    snapshot = {
        "updated_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "results_dir": str(results_dir),
        "active_case": active_case,
        "active_live": live_docs.get(active_case) or {},
        "live": live_docs,
        "metrics": scrape_metrics(sglang_url),
        "results": {
            "baseline": baseline_result,
            "mtp_topk1": mtp_result,
        },
        "summary": summary,
        "logs": {
            "baseline": tail_text(results_dir / "logs" / "baseline.log"),
            "mtp_topk1": tail_text(results_dir / "logs" / "mtp_topk1.log"),
        },
    }
    return snapshot


class DashboardHandler(BaseHTTPRequestHandler):
    results_dir: Path
    sglang_url: str

    def log_message(self, fmt: str, *args: Any) -> None:
        return

    def send_body(self, status: int, body: bytes, content_type: str) -> None:
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path in {"/", "/index.html"}:
            self.send_body(200, HTML.encode("utf-8"), "text/html; charset=utf-8")
            return

        if parsed.path == "/api/snapshot":
            snapshot = build_snapshot(self.results_dir, self.sglang_url)
            self.send_body(
                200,
                json.dumps(snapshot).encode("utf-8"),
                "application/json; charset=utf-8",
            )
            return

        content_type = mimetypes.guess_type(parsed.path)[0] or "text/plain"
        self.send_body(404, b"not found", content_type)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Gemma 4 MTP derby dashboard")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=7777)
    parser.add_argument("--results-dir", default="results/gemma4-mtp-sglang")
    parser.add_argument("--sglang-url", default="http://127.0.0.1:30000")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    handler = DashboardHandler
    handler.results_dir = Path(args.results_dir).resolve()
    handler.sglang_url = args.sglang_url
    os.makedirs(handler.results_dir, exist_ok=True)

    server = ThreadingHTTPServer((args.host, args.port), handler)
    print(f"Gemma 4 MTP Derby dashboard: http://{args.host}:{args.port}")
    print(f"results_dir={handler.results_dir}")
    print(f"sglang_url={handler.sglang_url}")
    server.serve_forever()


if __name__ == "__main__":
    main()