llm/scripts/generate_model_chat_status.py

from __future__ import annotations

import argparse
import datetime as dt
import json
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_HEALTH_URL = "http://192.168.220.91:8765/api/health"
DEFAULT_OUTPUT = ROOT / "reports" / "model-chat" / "status.md"


def fetch_json(url: str, timeout: int) -> dict[str, Any]:
    request = urllib.request.Request(url, headers={"Accept": "application/json"})
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            payload = response.read().decode("utf-8")
    except urllib.error.URLError as exc:
        raise RuntimeError(f"failed to fetch {url}: {exc}") from exc
    data = json.loads(payload)
    if not isinstance(data, dict):
        raise RuntimeError(f"health endpoint returned {type(data).__name__}, expected object")
    return data


def table(headers: list[str], rows: list[list[Any]]) -> list[str]:
    lines = [
        "| " + " | ".join(headers) + " |",
        "| " + " | ".join("---" for _ in headers) + " |",
    ]
    for row in rows:
        lines.append("| " + " | ".join(str(value) if value is not None else "" for value in row) + " |")
    return lines


def summarize_routes(health: dict[str, Any]) -> list[list[Any]]:
    rows = []
    for route in health.get("routes") or []:
        model = route.get("model") or {}
        readiness = route.get("readiness") or {}
        rows.append(
            [
                route.get("plugin"),
                model.get("id"),
                route.get("served_model_name"),
                readiness.get("status") or ("online" if route.get("available") else "offline"),
                route.get("base_url"),
            ]
        )
    return rows


def summarize_endpoints(health: dict[str, Any]) -> list[list[Any]]:
    rows = []
    for endpoint in health.get("endpoints") or []:
        models = ", ".join(endpoint.get("models") or [])
        rows.append(
            [
                endpoint.get("id"),
                endpoint.get("status"),
                endpoint.get("base_url"),
                models,
                endpoint.get("latency_ms"),
            ]
        )
    return rows


def summarize_profiles(health: dict[str, Any]) -> list[list[Any]]:
    rows = []
    for profile_id, profile in (health.get("gpu_profile_status") or {}).items():
        rows.append(
            [
                profile_id,
                "yes" if profile.get("ready") else "no",
                ", ".join(profile.get("missing") or []),
                ", ".join(profile.get("conflicts") or []),
            ]
        )
    return rows


def summarize_model_services(health: dict[str, Any]) -> list[list[Any]]:
    rows = []
    for service in health.get("model_services") or []:
        storage = service.get("storage") or {}
        endpoint_models = ", ".join(service.get("endpoint_models") or [])
        rows.append(
            [
                service.get("id"),
                ", ".join(service.get("plugins") or []),
                service.get("runtime"),
                service.get("served_model_name"),
                service.get("status"),
                "yes" if service.get("online") else "no",
                storage.get("status") or ("ok" if storage.get("exists") else "missing"),
                endpoint_models,
            ]
        )
    return rows


def build_markdown(health: dict[str, Any], source_url: str) -> str:
    generated_at = dt.datetime.now(dt.UTC).replace(microsecond=0).isoformat()
    gpu = health.get("gpu") or {}
    lines = [
        "# Model Chat Status",
        "",
        f"- Generated at: `{generated_at}`",
        f"- Source: `{source_url}`",
        f"- Health status: `{health.get('status')}`",
        f"- Host: `{health.get('host')}`",
        f"- Registry models: `{health.get('registry_models')}`",
        f"- GPU: `{gpu.get('summary') or 'unknown'}`",
        "",
        "## Plugin Routes",
        "",
        *table(["plugin", "model", "served model", "status", "base URL"], summarize_routes(health)),
        "",
        "## GPU Profiles",
        "",
        *table(["profile", "ready", "missing", "conflicts"], summarize_profiles(health)),
        "",
        "## Endpoints",
        "",
        *table(["endpoint", "status", "base URL", "models", "latency ms"], summarize_endpoints(health)),
        "",
        "## Model Services",
        "",
        *table(
            ["model", "plugins", "runtime", "served model", "status", "online", "storage", "endpoint models"],
            summarize_model_services(health),
        ),
        "",
    ]
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description="Generate a Markdown status report from Model Chat /api/health.")
    parser.add_argument("--url", default=DEFAULT_HEALTH_URL)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument("--timeout", type=int, default=60)
    parser.add_argument("--print", action="store_true")
    args = parser.parse_args()

    health = fetch_json(args.url, args.timeout)
    markdown = build_markdown(health, args.url)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(markdown + "\n", encoding="utf-8")
    if args.print:
        sys.stdout.write(markdown + "\n")
    else:
        print(f"Wrote {args.output}")
    return 0


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