8ef8c401d6
A terminal dashboard that polls the router's GET /metrics and renders quota burn vs plan, per-model cost/energy/carbon, verdict mix, recent routing decisions, and health warnings. Auto-refresh with last-good-data error resilience and keyboard controls (q/Q/Ctrl+C quit, r refresh, 1-5 focus panels). Pins textual (deliberate UI-only dependency, imported only by tui.py, never by the service).
366 lines
12 KiB
Python
366 lines
12 KiB
Python
"""Textual terminal dashboard over the router's GET /metrics endpoint.
|
|
|
|
This is a SEPARATE entrypoint from the FastAPI service: ``python tui.py`` runs
|
|
the dashboard in the foreground. Importing this module never starts the app —
|
|
``DashboardApp.run()``/``run_test()`` are what run it.
|
|
|
|
Data layer vs rendering is split so the TUI is testable without a terminal:
|
|
|
|
* ``fetch_metrics(base_url)`` — pure HTTP GET of ``<base_url>/metrics`` using
|
|
``requests``; returns the parsed JSON as a plain dict. Raises on HTTP or
|
|
network error so the caller can surface it.
|
|
* ``build_model(data)`` — pure: turns the raw /metrics JSON into a plain dict
|
|
of rendered panel payloads (lists of display rows). No textual involved.
|
|
|
|
The ``DashboardApp`` class is a Textual ``App``; tests drive it via
|
|
``App.run_test()`` with a stubbed fetcher and assert on the *data model*
|
|
(``app._last_model``, the panel payloads) rather than pixels.
|
|
|
|
Only this module imports ``textual`` anywhere in the repository; the service
|
|
dispatch path never touches it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Callable, Optional
|
|
|
|
import requests
|
|
from textual.app import App, ComposeResult
|
|
from textual.containers import VerticalScroll
|
|
from textual.widgets import DataTable, Footer, Header, Static
|
|
|
|
__all__ = ["fetch_metrics", "build_model", "DashboardApp"]
|
|
|
|
DEFAULT_BASE_URL = "http://127.0.0.1:8080"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Data layer — importable and testable without a running TUI.
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def fetch_metrics(base_url: str) -> dict:
|
|
"""GET ``<base_url>/metrics`` and return the parsed JSON dict.
|
|
|
|
Raises on any HTTP or network error (via ``raise_for_status`` and the
|
|
underlying ``requests`` exception) so the caller can catch it and render
|
|
a "cannot reach router" state.
|
|
"""
|
|
url = base_url.rstrip("/") + "/metrics"
|
|
resp = requests.get(url, timeout=10)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def build_model(data: dict) -> dict:
|
|
"""Turn raw /metrics JSON into a plain dict of rendered panel payloads.
|
|
|
|
Returns keys: ``quota`` (list of {label, value} rows), ``per_model``
|
|
(list of rows), ``verdict_mix`` (list of {verdict, count}),
|
|
``recent_decisions`` (list of rows), ``warnings`` (list of strings).
|
|
"""
|
|
quota = data.get("quota")
|
|
if quota:
|
|
quota_rows = [
|
|
{"label": "plan_kwh", "value": quota.get("plan_kwh")},
|
|
{"label": "metered_kwh_30d", "value": quota.get("metered_kwh_30d")},
|
|
{
|
|
"label": "fraction",
|
|
"value": quota.get("metered_fraction_of_plan"),
|
|
},
|
|
{"label": "calls", "value": quota.get("metered_calls_30d")},
|
|
]
|
|
else:
|
|
quota_rows = [{"label": "quota", "value": "N/A (plan not set)"}]
|
|
|
|
per_model = [
|
|
{
|
|
"model": r.get("model_id"),
|
|
"calls": r.get("calls"),
|
|
"cost_usd": r.get("sum_cost_usd"),
|
|
"energy_kwh": r.get("sum_energy_kwh"),
|
|
"carbon_g_co2eq": r.get("sum_carbon_g_co2eq"),
|
|
}
|
|
for r in (data.get("per_model") or [])
|
|
]
|
|
|
|
mix = data.get("verdict_mix") or {}
|
|
verdict_mix = [
|
|
{"verdict": k, "count": v} for k, v in sorted(mix.items())
|
|
]
|
|
|
|
recent = []
|
|
for r in data.get("recent_decisions") or []:
|
|
recent.append(
|
|
{
|
|
"id": r.get("id"),
|
|
"kind": r.get("kind"),
|
|
"category": r.get("task_category"),
|
|
"tier": r.get("task_tier"),
|
|
"selected": r.get("selected_model") or "none",
|
|
"est_cost_usd": r.get("est_cost_usd"),
|
|
}
|
|
)
|
|
|
|
coverage = data.get("coverage") or {}
|
|
warnings = list(coverage.get("warnings") or [])
|
|
|
|
return {
|
|
"quota": quota_rows,
|
|
"per_model": per_model,
|
|
"verdict_mix": verdict_mix,
|
|
"recent_decisions": recent,
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Textual App — the terminal dashboard.
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def _fmt_usd(v: Optional[float]) -> str:
|
|
if v is None:
|
|
return "n/a"
|
|
return f"{v:.6f}"
|
|
|
|
|
|
class DashboardApp(App):
|
|
"""A terminal dashboard that reads GET /metrics and renders panels."""
|
|
|
|
TITLE = "LLM Router Monitor"
|
|
# How often to re-fetch /metrics when running. A class attribute so tests
|
|
# can override it per-instance (via the ``refresh_seconds`` constructor
|
|
# arg) and drive ticks without real wall-clock sleeps.
|
|
REFRESH_SECONDS = 5.0
|
|
# Textual's binding table is a framework-read list; not mutated by us.
|
|
BINDINGS = [ # noqa: RUF012
|
|
("q", "quit", "Quit"),
|
|
("Q", "quit", "Quit"),
|
|
("ctrl+c", "quit", "Quit"),
|
|
("r", "refresh", "Refresh"),
|
|
("1", "focus_panel(0)", "Model table"),
|
|
("2", "focus_panel(1)", "Verdict table"),
|
|
("3", "focus_panel(2)", "Decision table"),
|
|
("4", "focus_panel(3)", "Quota panel"),
|
|
("5", "focus_panel(4)", "Warnings panel"),
|
|
]
|
|
|
|
CSS = """
|
|
#error-panel {
|
|
background: $error;
|
|
color: $text;
|
|
padding: 1;
|
|
margin: 1 0;
|
|
display: none;
|
|
}
|
|
#error-panel.visible {
|
|
display: block;
|
|
}
|
|
#loading-panel {
|
|
color: $text-muted;
|
|
text-style: italic;
|
|
display: none;
|
|
}
|
|
#loading-panel.visible {
|
|
display: block;
|
|
}
|
|
.panel-title {
|
|
text-style: bold;
|
|
color: $accent;
|
|
margin: 1 0 0 0;
|
|
}
|
|
DataTable {
|
|
height: auto;
|
|
max-height: 12;
|
|
border: round $primary;
|
|
}
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
base_url: Optional[str] = None,
|
|
fetcher: Optional[Callable[[str], dict]] = None,
|
|
refresh_seconds: Optional[float] = None,
|
|
) -> None:
|
|
self.base_url = base_url or os.environ.get(
|
|
"ROUTER_METRICS_URL", DEFAULT_BASE_URL
|
|
)
|
|
self._fetcher: Callable[[str], dict] = fetcher or fetch_metrics
|
|
self.refresh_seconds = (
|
|
refresh_seconds if refresh_seconds is not None else self.REFRESH_SECONDS
|
|
)
|
|
self._last_model: Optional[dict] = None
|
|
self._last_error: Optional[Exception] = None
|
|
self._refreshing = False
|
|
# Ordered list of focusable panels, indexed by the 1-5 number keys.
|
|
self._panels = [
|
|
"model-table",
|
|
"verdict-table",
|
|
"decision-table",
|
|
"quota-panel",
|
|
"warnings-panel",
|
|
]
|
|
super().__init__()
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(show_clock=True)
|
|
with VerticalScroll():
|
|
yield Static("", id="error-panel")
|
|
yield Static("", id="loading-panel")
|
|
yield Static("Quota burn", classes="panel-title")
|
|
yield Static("—", id="quota-panel")
|
|
yield Static("Per-model", classes="panel-title")
|
|
yield DataTable(id="model-table")
|
|
yield Static("Verdict mix", classes="panel-title")
|
|
yield DataTable(id="verdict-table")
|
|
yield Static("Recent decisions", classes="panel-title")
|
|
yield DataTable(id="decision-table")
|
|
yield Static("Health / warnings", classes="panel-title")
|
|
yield Static("—", id="warnings-panel")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
self._set_up_tables()
|
|
self._refresh()
|
|
# Poll on the configured interval. set_interval returns a Timer; we
|
|
# keep the reference so a test can stop it, but we only ever schedule
|
|
# from on_mount so the timer is started exactly once.
|
|
if self.refresh_seconds > 0:
|
|
self._interval_timer = self.set_interval(
|
|
self.refresh_seconds, self._on_interval
|
|
)
|
|
|
|
def _on_interval(self) -> None:
|
|
self._refresh()
|
|
|
|
def action_refresh(self) -> None:
|
|
"""Force an immediate re-fetch (``r``)."""
|
|
self._refresh()
|
|
|
|
def _set_up_tables(self) -> None:
|
|
model_table = self.query_one("#model-table", DataTable)
|
|
model_table.add_columns("model", "calls", "cost $", "kWh", "gCO2eq")
|
|
verdict_table = self.query_one("#verdict-table", DataTable)
|
|
verdict_table.add_columns("verdict", "count")
|
|
decision_table = self.query_one("#decision-table", DataTable)
|
|
decision_table.add_columns(
|
|
"time", "kind", "category", "tier", "selected", "est $"
|
|
)
|
|
# The two Static panels are also number-key targets (4 and 5); make
|
|
# them focusable so 1-5 focus is uniform.
|
|
self.query_one("#quota-panel", Static).can_focus = True
|
|
self.query_one("#warnings-panel", Static).can_focus = True
|
|
|
|
def _refresh(self) -> None:
|
|
self._refreshing = True
|
|
self._set_loading(True)
|
|
try:
|
|
data = self._fetcher(self.base_url)
|
|
except Exception as exc: # noqa: BLE001 — any fetch failure must degrade
|
|
self._last_error = exc
|
|
self._render_error(exc)
|
|
# _last_model and the rendered panels are deliberately untouched:
|
|
# a transient failure keeps displaying the last good data and the
|
|
# interval timer retries on the next tick.
|
|
return
|
|
finally:
|
|
self._refreshing = False
|
|
self._set_loading(False)
|
|
self._last_error = None
|
|
model = build_model(data)
|
|
self._last_model = model
|
|
self._render(model)
|
|
|
|
def _set_loading(self, visible: bool) -> None:
|
|
try:
|
|
panel = self.query_one("#loading-panel", Static)
|
|
except Exception: # noqa: BLE001 — panel not composed yet (pre-mount)
|
|
return
|
|
panel.update("refreshing metrics…" if visible else "")
|
|
if visible:
|
|
panel.add_class("visible")
|
|
else:
|
|
panel.remove_class("visible")
|
|
|
|
def _render_error(self, exc: Exception) -> None:
|
|
panel = self.query_one("#error-panel", Static)
|
|
panel.update(
|
|
f"cannot reach router at {self.base_url}: {exc.__class__.__name__}: {exc}"
|
|
)
|
|
panel.add_class("visible")
|
|
|
|
def action_focus_panel(self, index: int) -> None:
|
|
"""Move focus to one of the numbered panels (1-5)."""
|
|
if index < 0 or index >= len(self._panels):
|
|
return
|
|
widget = self.query_one(f"#{self._panels[index]}")
|
|
try:
|
|
widget.focus()
|
|
except Exception: # noqa: BLE001 — a Static still takes focus harmlessly
|
|
pass
|
|
|
|
def _render(self, model: dict) -> None:
|
|
# Quota panel
|
|
quota_widget = self.query_one("#quota-panel", Static)
|
|
quota_widget.update(
|
|
"\n".join(
|
|
f"{r['label']}: {r['value']}" for r in model["quota"]
|
|
)
|
|
)
|
|
|
|
# Per-model table
|
|
mt = self.query_one("#model-table", DataTable)
|
|
mt.clear()
|
|
for r in model["per_model"]:
|
|
mt.add_row(
|
|
str(r["model"]),
|
|
str(r["calls"]),
|
|
_fmt_usd(r["cost_usd"]),
|
|
f"{r['energy_kwh']:.6g}",
|
|
f"{r['carbon_g_co2eq']:.4g}",
|
|
)
|
|
|
|
# Verdict mix
|
|
vt = self.query_one("#verdict-table", DataTable)
|
|
vt.clear()
|
|
for r in model["verdict_mix"]:
|
|
vt.add_row(r["verdict"], str(r["count"]))
|
|
if not model["verdict_mix"]:
|
|
vt.add_row("(no data)", "")
|
|
|
|
# Recent decisions
|
|
dt = self.query_one("#decision-table", DataTable)
|
|
dt.clear()
|
|
for r in model["recent_decisions"]:
|
|
time_s = str(r["id"]) # id is the stable ordering handle
|
|
dt.add_row(
|
|
str(time_s),
|
|
str(r["kind"]),
|
|
str(r["category"]),
|
|
str(r["tier"]),
|
|
str(r["selected"]),
|
|
_fmt_usd(r["est_cost_usd"]),
|
|
)
|
|
if not model["recent_decisions"]:
|
|
dt.add_row("(no decisions)", "", "", "", "", "")
|
|
|
|
# Warnings panel
|
|
warn_widget = self.query_one("#warnings-panel", Static)
|
|
if model["warnings"]:
|
|
warn_widget.update(
|
|
"\n".join(f"• {w}" for w in model["warnings"])
|
|
)
|
|
else:
|
|
warn_widget.update("No warnings.")
|
|
|
|
|
|
def main() -> None:
|
|
DashboardApp().run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|