Coverage for src / lilbee / cli / model.py: 100%
146 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-05-15 20:55 +0000
1"""`lilbee model` sub-app: list/show/pull/rm/browse for installed models.
3Thin Typer wrapper around the surface-agnostic use-cases in
4:mod:`lilbee.app.models`. Bare result models live in ``app.models``;
5the Rich renderers below adapt them for human-readable terminal output.
6"""
8from __future__ import annotations
10from pathlib import Path
11from typing import TYPE_CHECKING
13import typer
14from rich.console import Console
15from rich.progress import BarColumn, Progress, TextColumn, TimeRemainingColumn
16from rich.table import Table
18from lilbee.app.models import (
19 ListModelsResult,
20 PullEvent,
21 PullProgressEvent,
22 PullResult,
23 PullStatus,
24 ShowModelResult,
25 list_models_data,
26 pull_model_data,
27 remove_model_data,
28 show_model_data,
29)
30from lilbee.cli import theme
31from lilbee.cli.app import (
32 apply_overrides,
33 console,
34 data_dir_option,
35 global_option,
36)
37from lilbee.cli.helpers import json_output
38from lilbee.core.config import cfg
40if TYPE_CHECKING:
41 from collections.abc import Callable
43 from lilbee.catalog import DownloadProgress
44 from lilbee.catalog.types import ModelSource
47def _render_list(data: ListModelsResult) -> Table:
48 table = Table(title="Installed models")
49 table.add_column("Name", style=theme.ACCENT)
50 table.add_column("Source", style=theme.MUTED)
51 table.add_column("Task")
52 table.add_column("Size", justify="right")
53 for entry in data.models:
54 size = f"{entry.size_gb:.2f} GB" if entry.size_gb is not None else ""
55 table.add_row(entry.name, entry.source, entry.task or "", size)
56 return table
59def _render_show(data: ShowModelResult) -> str:
60 lines = [f"[{theme.ACCENT}]{data.model}[/{theme.ACCENT}]"]
61 if data.catalog is not None:
62 lines.extend(
63 [
64 f" display_name: {data.catalog.display_name}",
65 f" task: {data.catalog.task}",
66 f" size_gb: {data.catalog.size_gb}",
67 f" min_ram_gb: {data.catalog.min_ram_gb}",
68 f" hf_repo: {data.catalog.hf_repo}",
69 f" description: {data.catalog.description}",
70 ]
71 )
72 lines.append(f" installed: {data.installed}")
73 if data.source:
74 lines.append(f" source: {data.source}")
75 if data.path:
76 lines.append(f" path: {data.path}")
77 if data.manifest is not None:
78 lines.append(f" downloaded: {data.manifest.downloaded_at}")
79 return "\n".join(lines)
82model_app = typer.Typer(
83 name="model",
84 help="Manage installed and available models (pull / list / show / rm / browse).",
85 no_args_is_help=True,
86)
88_source_option = typer.Option(
89 None,
90 "--source",
91 "-s",
92 help="Filter by source: 'native' or 'remote' (default: all).",
93)
94_task_option = typer.Option(
95 None,
96 "--task",
97 "-t",
98 help="Filter by task: 'chat', 'embedding', 'vision', or 'rerank'.",
99)
100_yes_option = typer.Option(
101 False,
102 "--yes",
103 "-y",
104 help="Skip confirmation prompt.",
105)
108def _parse_source_or_bad_param(value: str | None) -> ModelSource | None:
109 """Parse a CLI --source value, raising typer.BadParameter on bad input."""
110 from lilbee.catalog.types import ModelSource
112 try:
113 return ModelSource.parse(value)
114 except ValueError as exc:
115 if cfg.json_mode:
116 json_output({"error": str(exc)})
117 raise SystemExit(1) from None
118 raise typer.BadParameter(str(exc)) from exc
121@model_app.command("list")
122def list_cmd(
123 source: str | None = _source_option,
124 task: str | None = _task_option,
125 data_dir: Path | None = data_dir_option,
126 use_global: bool = global_option,
127) -> None:
128 """List installed models across all sources."""
129 from lilbee.catalog.types import ModelTask
131 apply_overrides(data_dir=data_dir, use_global=use_global)
132 try:
133 parsed_task = ModelTask(task) if task else None
134 except ValueError as exc:
135 raise typer.BadParameter(str(exc)) from exc
136 data = list_models_data(source=_parse_source_or_bad_param(source), task=parsed_task)
137 if cfg.json_mode:
138 json_output(data.model_dump())
139 return
140 if not data.models:
141 console.print("No models installed.")
142 return
143 console.print(_render_list(data))
146@model_app.command("show")
147def show_cmd(
148 ref: str = typer.Argument(..., help="Model ref (e.g. 'Qwen/Qwen3-0.6B-GGUF')."),
149 data_dir: Path | None = data_dir_option,
150 use_global: bool = global_option,
151) -> None:
152 """Show catalog and installed metadata for a model."""
153 from lilbee.modelhub.model_manager import ModelNotFoundError
155 apply_overrides(data_dir=data_dir, use_global=use_global)
156 try:
157 data = show_model_data(ref)
158 except ModelNotFoundError as exc:
159 if cfg.json_mode:
160 json_output({"error": str(exc)})
161 else:
162 console.print(f"[{theme.ERROR}]{exc}[/{theme.ERROR}]")
163 raise typer.Exit(1) from None
164 if cfg.json_mode:
165 json_output(data.model_dump())
166 return
167 console.print(_render_show(data))
170def _run_pull(
171 ref: str,
172 src: ModelSource,
173 on_update: Callable[[DownloadProgress], None],
174) -> PullResult:
175 """Invoke ``pull_model_data`` and translate known errors to typer.Exit."""
176 try:
177 return pull_model_data(ref, src, on_update=on_update)
178 except (RuntimeError, PermissionError) as exc:
179 if cfg.json_mode:
180 json_output({"error": str(exc)})
181 else:
182 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] {exc}")
183 raise typer.Exit(1) from None
186def _pull_json_stream(ref: str, src: ModelSource) -> None:
187 """Emit newline-delimited JSON progress events, then the final result."""
189 def on_update(p: DownloadProgress) -> None:
190 event = PullProgressEvent(
191 model=ref, percent=p.percent, detail=p.detail, cache_hit=p.is_cache_hit
192 )
193 json_output(event.model_dump())
195 final = _run_pull(ref, src, on_update)
196 json_output({**final.model_dump(), "event": PullEvent.DONE.value})
199def _pull_interactive_progress(ref: str, src: ModelSource) -> None:
200 """Drive Rich's Live progress bar during a native HuggingFace download."""
201 err_console = Console(stderr=True, force_terminal=True)
202 with Progress(
203 TextColumn("[progress.description]{task.description}"),
204 BarColumn(),
205 TextColumn("{task.percentage:>3.0f}%"),
206 TextColumn("{task.fields[detail]}"),
207 TimeRemainingColumn(),
208 console=err_console,
209 transient=False,
210 ) as progress:
211 task_id = progress.add_task(f"Downloading {ref}", total=100, detail="")
213 def on_update(p: DownloadProgress) -> None:
214 progress.update(task_id, completed=p.percent, detail=p.detail)
216 final = _run_pull(ref, src, on_update)
218 if final.status == PullStatus.ALREADY_INSTALLED:
219 console.print(f"{ref} is already installed.")
220 else:
221 console.print(f"Pulled [{theme.ACCENT}]{ref}[/{theme.ACCENT}].")
224@model_app.command("pull")
225def pull_cmd(
226 ref: str = typer.Argument(..., help="Model ref to download (e.g. 'Qwen/Qwen3-0.6B-GGUF')."),
227 source: str = typer.Option(
228 "native",
229 "--source",
230 "-s",
231 help="Pull from 'native' (HuggingFace GGUF) or 'remote' (SDK-managed).",
232 ),
233 data_dir: Path | None = data_dir_option,
234 use_global: bool = global_option,
235) -> None:
236 """Download a model."""
237 from lilbee.catalog.types import ModelSource
239 apply_overrides(data_dir=data_dir, use_global=use_global)
240 src = _parse_source_or_bad_param(source) or ModelSource.NATIVE
241 if cfg.json_mode:
242 _pull_json_stream(ref, src)
243 else:
244 _pull_interactive_progress(ref, src)
247def _confirm_remove_or_exit(ref: str, yes: bool) -> None:
248 if yes or cfg.json_mode:
249 return
250 if not typer.confirm(f"Remove {ref}?", default=False):
251 console.print("Aborted.")
252 raise typer.Exit(0)
255@model_app.command("rm")
256def rm_cmd(
257 ref: str = typer.Argument(..., help="Model ref to remove."),
258 source: str | None = _source_option,
259 yes: bool = _yes_option,
260 data_dir: Path | None = data_dir_option,
261 use_global: bool = global_option,
262) -> None:
263 """Remove an installed model."""
264 apply_overrides(data_dir=data_dir, use_global=use_global)
265 src = _parse_source_or_bad_param(source)
266 _confirm_remove_or_exit(ref, yes)
267 data = remove_model_data(ref, source=src)
268 if cfg.json_mode:
269 json_output(data.model_dump())
270 if not data.deleted:
271 raise typer.Exit(1)
272 return
273 if not data.deleted:
274 console.print(f"[{theme.WARNING}]Not found: {ref}[/{theme.WARNING}]")
275 raise typer.Exit(1)
276 suffix = f" ({data.freed_gb:.2f} GB freed)" if data.freed_gb else ""
277 console.print(f"Removed [{theme.ACCENT}]{ref}[/{theme.ACCENT}]{suffix}.")
280def _is_interactive_terminal() -> bool:
281 """Return True when both stdin and stdout are connected to a TTY.
283 Extracted as a module-level helper so tests can patch it deterministically;
284 CliRunner replaces ``sys.stdin`` during invoke which makes direct
285 monkey-patching of ``sys.stdin.isatty`` unreliable.
286 """
287 import sys
289 return sys.stdin.isatty() and sys.stdout.isatty()
292@model_app.command("browse")
293def browse_cmd(
294 data_dir: Path | None = data_dir_option,
295 use_global: bool = global_option,
296) -> None:
297 """Open the Textual TUI directly on the model catalog screen.
299 Exit codes follow the project convention: 2 for invalid flag
300 combinations (``--json`` with an interactive-only command), 1 for
301 runtime environment failures (no TTY).
302 """
303 apply_overrides(data_dir=data_dir, use_global=use_global)
304 if cfg.json_mode:
305 json_output({"error": "model browse is interactive, not available in --json mode"})
306 raise typer.Exit(2)
307 if not _is_interactive_terminal():
308 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] model browse requires a terminal.")
309 raise typer.Exit(1)
311 from lilbee.cli.tui import run_tui
313 run_tui(initial_view="Catalog")