server.py

#!/usr/bin/env python3
"""Serve the dashboard and a small JSON API for local demonstration."""

from __future__ import annotations

import argparse
from functools import partial
import json
import math
import threading
from datetime import datetime, timezone
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse


ROOT = Path(__file__).resolve().parent


def clamp(value: float, minimum: float, maximum: float) -> float:
    return max(minimum, min(maximum, value))


class DashboardStore:
    def __init__(self, seed_path: Path) -> None:
        self.seed = json.loads(seed_path.read_text(encoding="utf-8"))
        self.tick = 0
        self.lock = threading.Lock()

    def snapshot(self) -> dict:
        with self.lock:
            self.tick += 1
            servers = []
            for index, source in enumerate(self.seed["servers"]):
                phase = self.tick / 3 + index
                cpu = clamp(source["cpu"] + math.sin(phase) * 4, 0, 99)
                gpu = clamp(source["gpu"] + math.cos(phase * 1.2) * 5, 0, 99) if source["gpu"] else 0
                temperature = clamp(source["temperature"] + math.sin(phase / 2) * 0.8, 18, 40)
                power = clamp(source["power"] + math.cos(phase) * 0.04, 0.2, 1.6)
                status = "warning" if cpu >= 80 or temperature >= 33 else "healthy"
                servers.append({**source, "cpu": round(cpu, 1), "gpu": round(gpu, 1), "temperature": round(temperature, 1), "power": round(power, 2), "status": status})

            alerts = _build_alerts(servers)
            return {
                "generated_at": datetime.now(timezone.utc).isoformat(),
                "servers": servers,
                "health_items": self.seed["health_items"],
                "alerts": alerts,
            }


def _build_alerts(servers: list[dict]) -> list[dict[str, str]]:
    """Build demo alerts without assuming a fixed seed-data array length."""
    by_id = {str(server["id"]): server for server in servers}
    alerts = []
    if edge_gpu := by_id.get("edge-gpu-04"):
        alerts.append({"severity": "critical", "title": "edge-gpu-04 GPU 温度接近阈值", "detail": f"当前 {edge_gpu['temperature']}°C · 阈值 35°C", "time": "刚刚"})
    if compute := by_id.get("compute-12"):
        alerts.append({"severity": "warning", "title": "Rack-B01 CPU 负载需要关注", "detail": f"compute-12 当前 CPU {compute['cpu']}%", "time": "刚刚"})
    alerts.append({"severity": "warning", "title": "storage-02 存储巡检提示", "detail": "演示告警，用于验证告警列表渲染", "time": "刚刚"})
    return alerts


STORE = DashboardStore(ROOT / "data" / "dashboard-seed.json")


class DashboardHandler(SimpleHTTPRequestHandler):
    store = STORE

    def do_GET(self) -> None:  # noqa: N802
        if urlparse(self.path).path == "/api/dashboard":
            body = json.dumps(self.store.snapshot(), ensure_ascii=False).encode("utf-8")
            self.send_response(HTTPStatus.OK)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(body)
            return
        super().do_GET()


def create_server(host: str = "127.0.0.1", port: int = 8080) -> ThreadingHTTPServer:
    """Create the local server without starting its request loop.

    Keeping creation separate from ``serve_forever`` lets API contract tests use
    port 0, then cleanly shut down the same standard-library server.
    """
    handler = partial(DashboardHandler, directory=str(ROOT))
    return ThreadingHTTPServer((host, port), handler)


def run_server(host: str = "127.0.0.1", port: int = 8080) -> None:
    """Run the dashboard with the original host and port defaults."""
    server = create_server(host, port)
    print(f"Dashboard: http://{host}:{port}")
    print(f"API:       http://{host}:{port}/api/dashboard")
    try:
        server.serve_forever()
    finally:
        server.server_close()


def main() -> None:
    parser = argparse.ArgumentParser(description="Run the local dashboard API server.")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8080)
    args = parser.parse_args()
    run_server(args.host, args.port)


if __name__ == "__main__":
    main()