"""Optional Textual explorer for ``ooai-llm`` model workflows.
Purpose:
Provide an interactive terminal surface over catalog, comparison, suite,
profile, and benchmark workflows.
Design:
Textual imports remain lazy so the base package stays lightweight. Data
loading, filtering, and refresh throttling live in ``ooai_llm.tui.data`` so
behavior can be tested without launching a terminal app.
"""
from __future__ import annotations
from collections.abc import Callable
from time import monotonic
from typing import Any, Literal
from .commands import default_tui_command_specs
from .data import (
TUIBenchmarkRow,
TUICatalogRow,
TUIComparisonRow,
TUIConfig,
TUIRefreshGate,
TUIRow,
TUISnapshot,
TUISuiteRow,
build_tui_snapshot,
filter_tui_rows,
rows_for_view,
snapshot_metrics,
snapshot_providers,
)
from .themes import tui_theme_css
__all__ = ["TUIConfig", "TUISnapshot", "build_tui_snapshot", "create_tui_app", "run_tui"]
AppView = Literal["home", "cheapest", "coding", "catalog", "suites", "profiles", "benchmarks"]
[docs]
def create_tui_app(config: TUIConfig | None = None, *, clock: Callable[[], float] = monotonic) -> Any:
"""Create the optional Textual app.
Args:
config: Initial source, provider, token-shape, and budget options.
clock: Monotonic clock injection used by tests for refresh cooldowns.
Raises:
RuntimeError: If Textual is not installed.
"""
config = config or TUIConfig()
try:
from textual.app import App, ComposeResult, SystemCommand
from textual.binding import Binding
from textual.containers import Container, Horizontal, Vertical
from textual.css.query import NoMatches
from textual.widgets import (
Button,
ContentSwitcher,
DataTable,
Footer,
Header,
Input,
Markdown,
Select,
Static,
)
except ImportError as exc: # pragma: no cover - covered by CLI missing-extra smoke tests.
raise RuntimeError('Textual is required for the TUI. Install with: pip install "ooai-llm[tui]"') from exc
view_order: tuple[AppView, ...] = (
("home", "cheapest", "coding", "catalog", "suites", "profiles", "benchmarks")
if config.show_benchmarks
else ("home", "cheapest", "coding", "catalog", "suites", "profiles")
)
class OOAILLMApp(App[None]):
"""Textual application for interactive model exploration."""
CSS = tui_theme_css(config.theme)
BINDINGS = [
Binding("tab", "next_view", "Next view", priority=True),
Binding("shift+tab", "previous_view", "Prev view", priority=True),
("n", "next_view", "Next view"),
("p", "previous_view", "Prev view"),
("]", "next_view", "Next view"),
("[", "previous_view", "Prev view"),
("1", "show_home", "Home"),
("2", "show_cheapest", "Cheapest"),
("3", "show_coding", "Coding"),
("4", "show_catalog", "Catalog"),
("5", "show_suites", "Suites"),
("6", "show_profiles", "Profiles"),
("7", "show_benchmarks", "Benchmarks"),
("/", "focus_search", "Search"),
("escape", "clear_search", "Clear search"),
("r", "refresh", "Refresh"),
("q", "quit", "Quit"),
]
def __init__(self) -> None:
super().__init__()
self.config = config
self.snapshot = TUISnapshot(config=config)
self.refresh_gate = TUIRefreshGate(cooldown_seconds=config.refresh_cooldown_seconds)
self.active_view: AppView = self._initial_active_view()
self.active_provider = "all"
self.query = config.query
self.visible_rows: dict[str, list[TUIRow]] = {}
self._clock = clock
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
with Vertical(id="layout"):
with Horizontal(id="dashboard"):
yield Static("", id="summary-card", classes="card")
yield Static("", id="status-card", classes="card")
yield Static("", id="notes-card", classes="card")
with Horizontal(id="nav"):
for view in view_order:
yield Button(
self._nav_button_label(view),
id=f"nav-{view}",
classes="nav-button",
compact=True,
)
with Horizontal(id="controls"):
yield Input(
value=self.query,
placeholder="Filter models, capabilities, providers, endpoints...",
id="search",
)
yield Select([("All providers", "all")], value="all", allow_blank=False, id="provider-filter")
yield Select(self._view_options(), value=self.active_view, allow_blank=False, id="view-select")
yield Button("Refresh", id="refresh-button", variant="warning")
with ContentSwitcher(initial=self.active_view, id="content"):
with Container(id="home", classes="pane"):
yield Markdown(self._home_markdown(), id="home-markdown")
with Horizontal(id="cheapest", classes="pane table-workspace"):
yield from self._table_pane("cheapest")
with Horizontal(id="coding", classes="pane table-workspace"):
yield from self._table_pane("coding")
with Horizontal(id="catalog", classes="pane table-workspace"):
yield from self._table_pane("catalog")
with Horizontal(id="suites", classes="pane table-workspace"):
yield from self._table_pane("suites")
with Container(id="profiles", classes="pane"):
yield Markdown(self._profile_markdown(), id="profile-markdown")
if self.config.show_benchmarks:
with Horizontal(id="benchmarks", classes="pane table-workspace"):
yield from self._table_pane("benchmarks")
yield Footer()
def on_mount(self) -> None:
self.title = "ooai-llm"
self.sub_title = "model catalog, cost, suites, profiles, and benchmark surfaces"
self._refresh_tables(force=True)
self._set_view(self.active_view)
def get_system_commands(self, screen: Any) -> Any:
if screen is not None:
yield from super().get_system_commands(screen)
for spec in default_tui_command_specs(show_benchmarks=self.config.show_benchmarks):
yield SystemCommand(
spec.title,
spec.help,
getattr(self, spec.action),
discover=spec.discover,
)
def on_button_pressed(self, event: Any) -> None:
button_id = getattr(event.button, "id", None)
if button_id == "refresh-button":
self.action_refresh()
elif isinstance(button_id, str) and button_id.startswith("nav-"):
view = button_id.removeprefix("nav-")
if view in view_order:
self._set_view(view) # type: ignore[arg-type]
def on_input_changed(self, event: Any) -> None:
if getattr(event.input, "id", None) != "search":
return
self.query = event.value
self._populate_tables()
self._update_status("Filter updated.")
def on_select_changed(self, event: Any) -> None:
select_id = getattr(event.select, "id", None)
value = str(event.value)
if select_id == "provider-filter":
self.active_provider = value
self._populate_tables()
self._update_status(f"Provider filter: {value}.")
elif select_id == "view-select" and value in view_order:
self._set_view(value) # type: ignore[arg-type]
def on_data_table_row_highlighted(self, event: Any) -> None:
table_id = getattr(event.data_table, "id", "")
view = table_id.removesuffix("-table")
if view == self.active_view:
self._update_detail(view) # type: ignore[arg-type]
def action_refresh(self) -> None:
self._refresh_tables(force=False)
def action_next_view(self) -> None:
index = view_order.index(self.active_view)
self._set_view(view_order[(index + 1) % len(view_order)])
def action_previous_view(self) -> None:
index = view_order.index(self.active_view)
self._set_view(view_order[(index - 1) % len(view_order)])
def action_show_home(self) -> None:
self._set_view("home")
def action_show_cheapest(self) -> None:
self._set_view("cheapest")
def action_show_coding(self) -> None:
self._set_view("coding")
def action_show_catalog(self) -> None:
self._set_view("catalog")
def action_show_suites(self) -> None:
self._set_view("suites")
def action_show_profiles(self) -> None:
self._set_view("profiles")
def action_show_benchmarks(self) -> None:
if self.config.show_benchmarks:
self._set_view("benchmarks")
def action_focus_search(self) -> None:
self.query_one("#search", Input).focus()
def action_clear_search(self) -> None:
self.query = ""
self.query_one("#search", Input).value = ""
self._populate_tables()
self._update_status("Search cleared.")
def _table_pane(self, view: str) -> ComposeResult:
yield DataTable(id=f"{view}-table", zebra_stripes=True, cursor_type="row")
yield Markdown("Select a row to see details.", id=f"{view}-detail", classes="detail")
@staticmethod
def _view_options() -> list[tuple[str, str]]:
labels = {
"home": "Home",
"cheapest": "Cheapest",
"coding": "Coding",
"catalog": "Catalog",
"suites": "Suites",
"profiles": "Profiles",
"benchmarks": "Benchmarks",
}
return [(labels[view], view) for view in view_order]
@staticmethod
def _nav_button_label(view: AppView) -> str:
labels = {
"home": "1 Home",
"cheapest": "2 Cheapest",
"coding": "3 Coding",
"catalog": "4 Catalog",
"suites": "5 Suites",
"profiles": "6 Profiles",
"benchmarks": "7 Benchmarks",
}
return labels[view]
@staticmethod
def _initial_active_view() -> AppView:
for view in ("cheapest", "coding", "catalog", "suites", "benchmarks"):
if view in view_order and config.should_load_view(view):
return view
return "home"
def _refresh_tables(self, *, force: bool) -> None:
decision = self.refresh_gate.request(now=self._clock(), force=force)
if not decision.allowed:
self._update_status(decision.message)
self.notify(decision.message, title="Refresh skipped", severity="warning")
return
self._update_status("Refreshing catalog snapshot...")
self.snapshot = build_tui_snapshot(self.config.model_copy(update={"query": self.query}))
self._update_provider_options()
self._update_dashboard()
self._populate_tables()
self._update_status("Snapshot loaded. Tab/Shift+Tab changes views; / focuses search.")
self.notify("Snapshot loaded.", title="ooai-llm", severity="information", timeout=2)
def _populate_tables(self) -> None:
for view in ("cheapest", "coding", "catalog", "suites", "benchmarks"):
if view == "benchmarks" and not self.config.show_benchmarks:
continue
self._populate_table(view) # type: ignore[arg-type]
self._update_detail(self.active_view)
def _populate_table(self, view: AppView) -> None:
if view not in {"cheapest", "coding", "catalog", "suites", "benchmarks"}:
return
table = self.query_one(f"#{view}-table", DataTable)
table.clear(columns=True)
columns = self._columns_for_view(view)
for column in columns:
table.add_column(column)
if not self.snapshot.config.should_load_view(view): # type: ignore[arg-type]
self.visible_rows[view] = []
table.add_row("Skipped by --views", *([""] * (len(columns) - 1)))
return
rows = filter_tui_rows(
rows_for_view(self.snapshot, view), # type: ignore[arg-type]
query=self.query,
provider=self.active_provider,
)
self.visible_rows[view] = rows
if not rows:
table.add_row("No matching rows", *([""] * (len(columns) - 1)))
return
for row in rows:
table.add_row(*self._cells_for_row(row))
def _set_view(self, view: AppView) -> None:
if view not in view_order:
return
self.active_view = view
self.query_one("#content", ContentSwitcher).current = view
view_select = self.query_one("#view-select", Select)
if str(view_select.value) != view:
view_select.value = view
self._update_nav_state()
table = self.query_one(f"#{view}-table", DataTable) if view not in {"home", "profiles"} else None
if table is not None:
table.focus()
self._update_detail(view)
self._update_status(f"View: {view}.")
def _update_provider_options(self) -> None:
providers = snapshot_providers(self.snapshot)
options = [("All providers", "all"), *[(provider.title(), provider) for provider in providers]]
provider_select = self.query_one("#provider-filter", Select)
provider_select.set_options(options)
if self.active_provider != "all" and self.active_provider not in providers:
self.active_provider = "all"
provider_select.value = self.active_provider
def _update_nav_state(self) -> None:
for view in view_order:
button = self.query_one(f"#nav-{view}", Button)
button.set_class(view == self.active_view, "active")
def _update_dashboard(self) -> None:
metrics = snapshot_metrics(self.snapshot)
self.query_one("#summary-card", Static).update(
"[b]Explorer[/b]\n"
f"source: {self.config.source}\n"
f"catalog: {metrics['catalog_scope']}\n"
f"providers: {metrics['providers']}\n"
f"loaded: {metrics['loaded_views']}\n"
f"catalog rows: {metrics['catalog_rows']}\n"
f"coding rows: {metrics['coding_rows']}"
)
self.query_one("#notes-card", Static).update(
"[b]Snapshot notes[/b]\n"
f"notes: {metrics['notes']}\n"
f"suites: {metrics['suite_rows']}\n"
f"bench endpoints: {metrics['benchmark_rows']}\n"
f"limit: {self.config.limit or 'all'}"
)
def _update_status(self, message: str) -> None:
self.query_one("#status-card", Static).update(
"[b]Controls[/b]\n"
f"{message}\n"
f"theme: {self.config.theme}\n"
f"loaded: {', '.join(self.config.resolved_load_views()) or 'none'}\n"
f"shape: {self.config.input_tokens:,} in / {self.config.output_tokens:,} out\n"
f"budget: ${self.config.budget_usd}"
)
def _update_detail(self, view: AppView) -> None:
if view in {"home", "profiles"}:
return
if view == "benchmarks" and not self.config.show_benchmarks:
return
rows = self.visible_rows.get(view, [])
try:
detail = self.query_one(f"#{view}-detail", Markdown)
table = self.query_one(f"#{view}-table", DataTable)
except NoMatches:
return
if not self.snapshot.config.should_load_view(view): # type: ignore[arg-type]
detail.update(
"### Not loaded\n\n"
"This view was skipped by the TUI load scope. Restart with a broader "
"`--views` value, for example `--views cheapest,coding,catalog`."
)
return
if not rows:
detail.update("### No rows\n\nAdjust the search or provider filter, then refresh if needed.")
return
row_index = min(max(table.cursor_row, 0), len(rows) - 1)
detail.update(self._detail_for_row(rows[row_index]))
def _home_markdown(self) -> str:
providers = ", ".join(self.config.providers or ["all configured providers"])
load_views = ", ".join(self.config.resolved_load_views()) or "none"
return (
"# ooai-llm explorer\n\n"
"Use this screen to choose model candidates, compare call costs, inspect capabilities, "
"and jump from discovery into profile/suite workflows.\n\n"
f"**Source:** `{self.config.source}`\n\n"
f"**Catalog scope:** `{self.config.catalog_scope}`\n\n"
f"**Providers:** `{providers}`\n\n"
f"**Theme:** `{self.config.theme}`\n\n"
f"**Loaded views:** `{load_views}`\n\n"
f"**Call shape:** `{self.config.input_tokens:,}` input / "
f"`{self.config.output_tokens:,}` output tokens\n\n"
f"**Budget:** `${self.config.budget_usd}`\n\n"
"**Navigation:** click the top view buttons, use `1`-`7`, or use `n`/`p`, `]`/`[`, "
"`Tab`/`Shift+Tab` for next/previous view.\n\n"
"**Other keys:** `/` search, `Esc` clear search, `r` refresh, `q` quit.\n"
)
@staticmethod
def _profile_markdown() -> str:
return (
"# Profiles and runtimes\n\n"
"`ChatModelProfile` turns a selected model into serializable runtime configuration. "
"`LLM` wraps the runnable and records observed usage/cost when LangChain/provider "
"metadata is present.\n\n"
"```python\n"
"profile = ChatModelProfile(id='coding', model='mistral:MODEL', temperature=0)\n"
"runtime = profile.create_runtime(id='coding-runtime')\n"
"response = runtime.invoke('Review this function for bugs.')\n"
"print(runtime.total_cost_usd)\n"
"```\n\n"
"Useful CLI checks:\n\n"
"```bash\n"
"ooai-llm profiles validate --input profile.json\n"
"ooai-llm profiles resolve --input profile.json --format json\n"
"ooai-llm models suite --from-catalog --coding-only --sort cost --limit 5\n"
"```\n"
)
@staticmethod
def _columns_for_view(view: AppView) -> tuple[str, ...]:
if view in {"cheapest", "coding"}:
return ("Provider", "Model", "Call", "Calls/$", "Price/1M", "Tokens", "Capabilities")
if view == "catalog":
return ("Provider", "Model", "Release", "Price/1M", "Tokens", "Capabilities")
if view == "suites":
return ("Key", "Role", "Provider", "Model", "Capabilities")
return ("Name", "Method", "Path", "Query", "Stability", "Description")
@staticmethod
def _cells_for_row(row: TUIRow) -> tuple[str, ...]:
if isinstance(row, TUIComparisonRow):
return (
row.provider,
row.model,
row.call_cost,
row.calls_per_budget,
row.price_per_1m,
row.tokens,
row.capabilities,
)
if isinstance(row, TUICatalogRow):
return (row.provider, row.model, row.release, row.price_per_1m, row.tokens, row.capabilities)
if isinstance(row, TUISuiteRow):
return (row.key, row.role, row.provider, row.model, row.capabilities)
if isinstance(row, TUIBenchmarkRow):
return (row.name, row.method, row.path, row.query, row.stability, row.description)
return ("unknown",)
@staticmethod
def _detail_for_row(row: TUIRow) -> str:
if isinstance(row, TUIComparisonRow):
return (
f"### {row.model}\n\n"
f"**Provider:** `{row.provider}`\n\n"
f"**Estimated call cost:** `{row.call_cost}`\n\n"
f"**Calls per budget:** `{row.calls_per_budget}`\n\n"
f"**Price / 1M input-output:** `{row.price_per_1m}`\n\n"
f"**Context / output tokens:** `{row.tokens}`\n\n"
f"**Release:** `{row.release}`\n\n"
f"**Source:** `{row.source}`\n\n"
f"**Capabilities:** {row.capabilities or 'n/a'}\n"
)
if isinstance(row, TUICatalogRow):
return (
f"### {row.model}\n\n"
f"**Provider:** `{row.provider}`\n\n"
f"**Release:** `{row.release}`\n\n"
f"**Price / 1M input-output:** `{row.price_per_1m}`\n\n"
f"**Context / output tokens:** `{row.tokens}`\n\n"
f"**Source:** `{row.source}`\n\n"
f"**Capabilities:** {row.capabilities or 'n/a'}\n"
)
if isinstance(row, TUISuiteRow):
return (
f"### {row.key}\n\n"
f"**Role:** `{row.role}`\n\n"
f"**Provider:** `{row.provider}`\n\n"
f"**Model:** `{row.model}`\n\n"
f"**Capabilities:** {row.capabilities or 'n/a'}\n\n"
"Use this as a candidate for `suite.to_profiles()` when wiring LangGraph/LangChain experiments.\n"
)
if isinstance(row, TUIBenchmarkRow):
return (
f"### {row.name}\n\n"
f"**Method:** `{row.method}`\n\n"
f"**Path:** `{row.path}`\n\n"
f"**Query:** `{row.query}`\n\n"
f"**Stability:** `{row.stability}`\n\n"
f"{row.description}\n\n"
"The TUI lists endpoint surfaces only; CLI benchmark commands perform the live HTTP fetches.\n"
)
return "### Unknown row"
return OOAILLMApp()
[docs]
def run_tui(config: TUIConfig | None = None) -> None:
"""Run the optional Textual application.
Args:
config: Initial source, provider, token-shape, and budget options.
Raises:
RuntimeError: If Textual is not installed.
"""
create_tui_app(config).run()