"""Pretty end-of-launch access card: all local URLs and API endpoints.""" from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass from rich.console import Console, Group from rich.panel import Panel from rich.table import Table from rich.text import Text from gpu_rent.config import Config from gpu_rent.llm_runtime import normalize_runtime from gpu_rent.state import load_state Log = Callable[[str], None] @dataclass(frozen=True) class AccessLink: label: str url: str note: str = "" def resolve_llm_runtime(cfg: Config) -> str: """Same source as tunnel_forwards: live cfg only (not stale state notes).""" return normalize_runtime(getattr(cfg, "llm_runtime", "none")) def collect_access_links(cfg: Config, *, tunneled: bool) -> list[AccessLink]: """Build the list of user-facing endpoints (unit-tested).""" links: list[AccessLink] = [] swarm = bool(getattr(cfg, "enable_swarmui", True)) if tunneled: if swarm: port = cfg.swarmui_local_port base = f"http://127.0.0.1:{port}" links.extend( [ AccessLink("SwarmUI UI", base, "браузер"), AccessLink("SwarmUI API", f"{base}/API/", "HTTP JSON"), AccessLink("SwarmUI MCP", f"{base}/mcp", "Cursor mcp.json"), ] ) runtime = resolve_llm_runtime(cfg) if runtime == "ollama": o = cfg.ollama_local_port links.extend( [ AccessLink( "Ollama API", f"http://127.0.0.1:{o}", f"OLLAMA_HOST=http://127.0.0.1:{o}", ), AccessLink( "Ollama tags", f"http://127.0.0.1:{o}/api/tags", "список моделей", ), AccessLink( "Ollama chat", f"http://127.0.0.1:{o}/api/chat", "POST generate", ), ] ) if not links: links.append( AccessLink( "Туннель", "gpu-rent tunnel", "нет сервисов — ENABLE_SWARMUI / LLM_RUNTIME", ) ) else: links.append( AccessLink( "Туннель", "gpu-rent tunnel --open", "локальные URL появятся после tunnel", ) ) return links def mcp_snippet_lines(cfg: Config) -> list[str]: if not bool(getattr(cfg, "enable_swarmui", True)): return ["# SwarmUI MCP skip (llm-only)"] port = cfg.swarmui_local_port return [ "{", ' "mcpServers": {', ' "swarmui": {', f' "url": "http://127.0.0.1:{port}/mcp"', " }", " }", "}", ] def render_access_panel( cfg: Config, *, tunneled: bool = True, host: str | None = None, title: str = "gpu-rent · доступы", ) -> Panel: links = collect_access_links(cfg, tunneled=tunneled) table = Table(show_header=True, header_style="bold", box=None, pad_edge=False) table.add_column("сервис", style="cyan", no_wrap=True) table.add_column("URL / команда", style="bold white") table.add_column("заметка", style="dim") for link in links: table.add_row(link.label, link.url, link.note) cmds = Table(show_header=False, box=None, pad_edge=False) cmds.add_column(style="dim", no_wrap=True) cmds.add_column() cmds.add_row("открыть UI", "gpu-rent open") if resolve_llm_runtime(cfg) == "ollama": cmds.add_row("открыть LLM", "gpu-rent open --llm") cmds.add_row("hold killer", "gpu-rent hold") cmds.add_row("стоп GPU", "gpu-rent stop") if tunneled: cmds.add_row("Ctrl+C / Ctrl+D", "stop GPU, диски остаются") if not tunneled: cmds.add_row("туннель", "gpu-rent tunnel --open") mcp = Text("\n".join(mcp_snippet_lines(cfg)), style="green") subtitle = Text() if tunneled: subtitle.append("туннель слушает localhost", style="green") if host: subtitle.append(f" · VM {host}", style="dim") else: subtitle.append("облако готово, туннеля нет", style="yellow") if host: subtitle.append(f" · FIP {host}", style="dim") warn_bits: list[str] = [] try: notes = load_state().notes or {} if notes.get("idle_killer") == "failed": warn_bits.append( "idle-killer НЕ вооружён — GPU может крутиться без авто-stop → gpu-rent stop" ) if notes.get("stack_vm_error"): warn_bits.append(f"стек VM: {str(notes['stack_vm_error'])[:140]}") if notes.get("gpu_env_error"): warn_bits.append(f"GPU-стек: {str(notes['gpu_env_error'])[:140]}") if notes.get("llm_error"): warn_bits.append(f"LLM ошибка: {str(notes['llm_error'])[:120]}") except Exception: pass parts: list = [subtitle, Text("")] if warn_bits: parts.append(Text("⚠ ВНИМАНИЕ — биллинг / готовность", style="bold white on red")) for w in warn_bits: parts.append(Text(f" • {w}", style="bold red")) parts.append(Text("")) parts.extend( [ table, Text(""), Text("команды", style="bold"), cmds, Text(""), Text("Cursor MCP (вставь в mcp.json — файл сам не трогаем)", style="bold"), mcp, ] ) body = Group(*parts) border = "red" if warn_bits else "bright_blue" return Panel( body, title=f"[bold]{title}[/bold]", border_style=border, padding=(1, 2), ) def print_access_card( cfg: Config, *, tunneled: bool = True, host: str | None = None, console: Console | None = None, log: Log | None = None, ) -> None: """Print Rich panel; fall back to plain lines if needed.""" panel = render_access_panel(cfg, tunneled=tunneled, host=host) if console is not None: console.print() console.print(panel) console.print() return if log is not None: # Plain fallback for non-Rich loggers log("") log("══ gpu-rent · доступы ══") try: notes = load_state().notes or {} if notes.get("idle_killer") == "failed": log("⚠ idle-killer НЕ вооружён — GPU без авто-stop → gpu-rent stop") if notes.get("stack_vm_error"): log(f"⚠ стек VM: {notes['stack_vm_error']}") if notes.get("gpu_env_error"): log(f"⚠ GPU-стек: {notes['gpu_env_error']}") except Exception: pass for link in collect_access_links(cfg, tunneled=tunneled): extra = f" ({link.note})" if link.note else "" log(f" {link.label:14} {link.url}{extra}") log("— Cursor MCP —") for line in mcp_snippet_lines(cfg): log(line) log("hold: gpu-rent hold | stop: gpu-rent stop") log("Ctrl+C / Ctrl+D — stop GPU (диски остаются)") log("") return from gpu_rent.term import console as default_console default_console.print() default_console.print(panel) default_console.print() def print_mcp_snippet(cfg: Config, log: Log) -> None: """Back-compat: plain MCP + links via log.""" print_access_card(cfg, tunneled=True, log=log)