417 lines
15 KiB
Python
417 lines
15 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
|
|
|
|
from textual.app import App, ComposeResult
|
|
from textual.containers import VerticalScroll
|
|
from textual.widgets import DataTable, Footer, Header, Static
|
|
|
|
from tui_model import (
|
|
DEFAULT_BASE_URL,
|
|
build_category_breakdown,
|
|
build_model,
|
|
decision_row,
|
|
fetch_metrics,
|
|
)
|
|
from tui_screens import DecisionDetailScreen
|
|
from tui_sse import DecisionStream
|
|
|
|
__all__ = ["DashboardApp"]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 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"),
|
|
("e", "show_decision_detail", "Detail"),
|
|
("1", "focus_panel(0)", "Model table"),
|
|
("2", "focus_panel(1)", "Verdict table"),
|
|
("3", "focus_panel(2)", "Decision table"),
|
|
("4", "focus_panel(3)", "Breakdown table"),
|
|
("5", "focus_panel(4)", "Quota panel"),
|
|
("6", "focus_panel(5)", "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,
|
|
live_events: bool = False,
|
|
) -> 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
|
|
)
|
|
# Live events default off so tests don't start a network thread.
|
|
# The ``python tui.py`` entrypoint enables it.
|
|
self.live_events = live_events
|
|
self._event_stream: Optional[DecisionStream] = None
|
|
self._last_model: Optional[dict] = None
|
|
self._last_error: Optional[Exception] = None
|
|
# Track decision ids already rendered so the SSE replay (and every
|
|
# reconnect replay) does not duplicate rows that came from the initial
|
|
# /metrics poll or an earlier replay cycle.
|
|
self._seen_ids: set[int | None] = set()
|
|
self._refreshing = False
|
|
# Ordered list of focusable panels, indexed by the 1-6 number keys.
|
|
self._panels = [
|
|
"model-table",
|
|
"verdict-table",
|
|
"decision-table",
|
|
"category-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 (enter = details)", classes="panel-title")
|
|
yield DataTable(id="decision-table")
|
|
yield Static("Category → model breakdown", classes="panel-title")
|
|
yield DataTable(id="category-table")
|
|
yield Static("Health / warnings", classes="panel-title")
|
|
yield Static("—", id="warnings-panel")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
self._unmounted = False
|
|
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
|
|
)
|
|
if self.live_events:
|
|
self._event_stream = DecisionStream(
|
|
self.base_url, self._on_live_decision
|
|
)
|
|
self._event_stream.start()
|
|
|
|
def on_unmount(self) -> None:
|
|
self._unmounted = True
|
|
if hasattr(self, "_interval_timer"):
|
|
self._interval_timer.stop()
|
|
if self._event_stream is not None:
|
|
self._event_stream.stop()
|
|
|
|
def _on_interval(self) -> None:
|
|
if getattr(self, "_unmounted", False):
|
|
return
|
|
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(
|
|
"id", "kind", "category", "tier", "ctx", "selected", "est $"
|
|
)
|
|
decision_table.cursor_type = "row"
|
|
decision_table.zebra_stripes = True
|
|
category_table = self.query_one("#category-table", DataTable)
|
|
category_table.add_columns("category", "tier", "count", "majority", "share")
|
|
# Static panels are also number-key targets; make them focusable so
|
|
# 1-6 focus cycling is uniform.
|
|
self.query_one("#quota-panel", Static).can_focus = True
|
|
self.query_one("#warnings-panel", Static).can_focus = True
|
|
|
|
def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
|
|
"""Open the detail popup for the highlighted recent decision."""
|
|
if event.data_table.id != "decision-table" or not self._last_model:
|
|
return
|
|
row_index = event.cursor_row
|
|
decisions = self._last_model.get("recent_decisions", [])
|
|
if row_index is None or row_index < 0 or row_index >= len(decisions):
|
|
return
|
|
self.push_screen(DecisionDetailScreen(decisions[row_index]))
|
|
|
|
def action_show_decision_detail(self) -> None:
|
|
"""Open details for the currently selected decision (``e`` key)."""
|
|
decision_table = self.query_one("#decision-table", DataTable)
|
|
if decision_table.cursor_row is None or not self._last_model:
|
|
return
|
|
row_index = decision_table.cursor_row
|
|
decisions = self._last_model.get("recent_decisions", [])
|
|
if row_index < 0 or row_index >= len(decisions):
|
|
return
|
|
self.push_screen(DecisionDetailScreen(decisions[row_index]))
|
|
|
|
def _on_live_decision(self, decision: dict) -> None:
|
|
"""Thread callback: marshal the live decision onto the UI thread."""
|
|
self.call_from_thread(self._handle_live_decision, decision)
|
|
|
|
def _handle_live_decision(self, decision: dict) -> None:
|
|
"""Add a decision from the SSE stream to the front of the model."""
|
|
if self._last_model is None:
|
|
return
|
|
new_row = decision_row(decision)
|
|
if new_row.get("id") in self._seen_ids:
|
|
# The SSE endpoint replays its recent ring buffer on every connect
|
|
# (and this stream reconnects automatically after transient
|
|
# errors), so the same id can arrive more than once. Skip it to
|
|
# avoid duplicating a row already present from /metrics or an
|
|
# earlier replay.
|
|
return
|
|
if new_row.get("id") is not None:
|
|
self._seen_ids.add(new_row["id"])
|
|
self._last_model["recent_decisions"].insert(0, new_row)
|
|
# Keep the same cap the backend uses for /metrics consistency.
|
|
self._last_model["recent_decisions"] = self._last_model["recent_decisions"][:50]
|
|
self._update_category_breakdown(new_row)
|
|
self._render_decisions_table()
|
|
self._render_category_table()
|
|
|
|
def _update_category_breakdown(self, new_row: dict) -> None:
|
|
"""Fold one newly inserted decision into the category breakdown.
|
|
|
|
``build_category_breakdown`` rebuilds the whole (up to 50-row) list with
|
|
a sort + Counter per event, which is wasteful on the UI thread when only
|
|
the new row's (category, tier) bucket changed. Only a brand-new bucket
|
|
needs a full rebuild to add its row; an existing bucket keeps its counts
|
|
accurate (count rises on the next /metrics poll, at most one event stale).
|
|
"""
|
|
key = (new_row.get("category"), new_row.get("tier"))
|
|
existing = next(
|
|
(
|
|
r
|
|
for r in self._last_model["category_breakdown"]
|
|
if (r["category"], r["tier"]) == key
|
|
),
|
|
None,
|
|
)
|
|
if existing is None:
|
|
self._last_model["category_breakdown"] = build_category_breakdown(
|
|
self._last_model["recent_decisions"]
|
|
)
|
|
|
|
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._seen_ids = {r.get("id") for r in model["recent_decisions"]}
|
|
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-6)."""
|
|
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)", "")
|
|
|
|
self._render_decisions_table()
|
|
self._render_category_table()
|
|
|
|
# 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 _render_decisions_table(self) -> None:
|
|
if self._last_model is None:
|
|
return
|
|
dt = self.query_one("#decision-table", DataTable)
|
|
dt.clear()
|
|
for r in self._last_model["recent_decisions"]:
|
|
dt.add_row(
|
|
str(r.get("id")),
|
|
str(r.get("kind")),
|
|
str(r.get("category")),
|
|
str(r.get("tier")),
|
|
str(r.get("required_context_tokens")),
|
|
str(r.get("selected")),
|
|
_fmt_usd(r.get("est_cost_usd")),
|
|
)
|
|
if not self._last_model["recent_decisions"]:
|
|
dt.add_row("(no decisions)", "", "", "", "", "", "")
|
|
|
|
def _render_category_table(self) -> None:
|
|
if self._last_model is None:
|
|
return
|
|
ct = self.query_one("#category-table", DataTable)
|
|
ct.clear()
|
|
for r in self._last_model["category_breakdown"]:
|
|
ct.add_row(
|
|
str(r["category"]),
|
|
str(r["tier"]),
|
|
str(r["count"]),
|
|
str(r["majority"]),
|
|
f"{r['share']:.0%}",
|
|
)
|
|
if not self._last_model["category_breakdown"]:
|
|
ct.add_row("(no decisions)", "", "", "", "")
|
|
|
|
|
|
def main() -> None:
|
|
DashboardApp(live_events=True).run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|