Coverage for src / lilbee / cli / model.py: 100%
159 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-06-28 01:01 +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, remote, ollama, lm_studio, or frontier (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 *,
175 allow_unsupported: bool = False,
176) -> PullResult:
177 """Invoke ``pull_model_data`` and translate known errors to typer.Exit."""
178 from lilbee.catalog.compat import UnsupportedArchError
180 try:
181 return pull_model_data(ref, src, on_update=on_update, allow_unsupported=allow_unsupported)
182 except UnsupportedArchError as exc:
183 msg = (
184 f"Architecture {exc.architecture!r} is not supported by this lilbee build.\n"
185 "Pass --allow-unsupported to try anyway."
186 )
187 if cfg.json_mode:
188 json_output(
189 {
190 "error": "unsupported_arch",
191 "arch": exc.architecture,
192 "ref": exc.ref,
193 }
194 )
195 else:
196 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] {msg}")
197 raise typer.Exit(1) from None
198 except (RuntimeError, PermissionError) as exc:
199 if cfg.json_mode:
200 json_output({"error": str(exc)})
201 else:
202 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] {exc}")
203 raise typer.Exit(1) from None
206def _pull_json_stream(ref: str, src: ModelSource, *, allow_unsupported: bool) -> None:
207 """Emit newline-delimited JSON progress events, then the final result."""
209 def on_update(p: DownloadProgress) -> None:
210 event = PullProgressEvent(
211 model=ref, percent=p.percent, detail=p.detail, cache_hit=p.is_cache_hit
212 )
213 json_output(event.model_dump())
215 final = _run_pull(ref, src, on_update, allow_unsupported=allow_unsupported)
216 json_output({**final.model_dump(), "event": PullEvent.DONE.value})
219def _pull_interactive_progress(ref: str, src: ModelSource, *, allow_unsupported: bool) -> None:
220 """Drive Rich's Live progress bar during a native HuggingFace download."""
221 err_console = Console(stderr=True, force_terminal=True)
222 with Progress(
223 TextColumn("[progress.description]{task.description}"),
224 BarColumn(),
225 TextColumn("{task.percentage:>3.0f}%"),
226 TextColumn("{task.fields[detail]}"),
227 TimeRemainingColumn(),
228 console=err_console,
229 transient=False,
230 ) as progress:
231 task_id = progress.add_task(f"Downloading {ref}", total=100, detail="")
233 def on_update(p: DownloadProgress) -> None:
234 progress.update(task_id, completed=p.percent, detail=p.detail)
236 final = _run_pull(ref, src, on_update, allow_unsupported=allow_unsupported)
238 if final.status == PullStatus.ALREADY_INSTALLED:
239 console.print(f"{ref} is already installed.")
240 else:
241 console.print(f"Pulled [{theme.ACCENT}]{ref}[/{theme.ACCENT}].")
244@model_app.command("pull")
245def pull_cmd(
246 ref: str = typer.Argument(..., help="Model ref to download (e.g. 'Qwen/Qwen3-0.6B-GGUF')."),
247 source: str = typer.Option(
248 "native",
249 "--source",
250 "-s",
251 help="Pull from 'native' (HuggingFace GGUF) or 'remote' (SDK-managed).",
252 ),
253 allow_unsupported: bool = typer.Option(
254 False,
255 "--allow-unsupported",
256 help="Pull even if the architecture isn't in the supported set (load may still fail).",
257 ),
258 data_dir: Path | None = data_dir_option,
259 use_global: bool = global_option,
260) -> None:
261 """Download a model."""
262 from lilbee.catalog.types import ModelSource
264 apply_overrides(data_dir=data_dir, use_global=use_global)
265 src = _parse_source_or_bad_param(source) or ModelSource.NATIVE
266 if cfg.json_mode:
267 _pull_json_stream(ref, src, allow_unsupported=allow_unsupported)
268 else:
269 _pull_interactive_progress(ref, src, allow_unsupported=allow_unsupported)
272def _confirm_remove_or_exit(ref: str, yes: bool) -> None:
273 if yes or cfg.json_mode:
274 return
275 if not typer.confirm(f"Remove {ref}?", default=False):
276 console.print("Aborted.")
277 raise typer.Exit(0)
280@model_app.command("rm")
281def rm_cmd(
282 ref: str = typer.Argument(..., help="Model ref to remove."),
283 source: str | None = _source_option,
284 yes: bool = _yes_option,
285 data_dir: Path | None = data_dir_option,
286 use_global: bool = global_option,
287) -> None:
288 """Remove an installed model."""
289 apply_overrides(data_dir=data_dir, use_global=use_global)
290 src = _parse_source_or_bad_param(source)
291 _confirm_remove_or_exit(ref, yes)
292 try:
293 data = remove_model_data(ref, source=src)
294 except ValueError as exc:
295 if cfg.json_mode:
296 json_output({"error": str(exc)})
297 else:
298 console.print(f"[{theme.ERROR}]{exc}[/{theme.ERROR}]")
299 raise typer.Exit(1) from None
300 if cfg.json_mode:
301 json_output(data.model_dump())
302 if not data.deleted:
303 raise typer.Exit(1)
304 return
305 if not data.deleted:
306 console.print(f"[{theme.WARNING}]Not found: {ref}[/{theme.WARNING}]")
307 raise typer.Exit(1)
308 suffix = f" ({data.freed_gb:.2f} GB freed)" if data.freed_gb else ""
309 console.print(f"Removed [{theme.ACCENT}]{ref}[/{theme.ACCENT}]{suffix}.")
312def _is_interactive_terminal() -> bool:
313 """Return True when both stdin and stdout are connected to a TTY.
315 Extracted as a module-level helper so tests can patch it deterministically;
316 CliRunner replaces ``sys.stdin`` during invoke which makes direct
317 monkey-patching of ``sys.stdin.isatty`` unreliable.
318 """
319 import sys
321 return sys.stdin.isatty() and sys.stdout.isatty()
324@model_app.command("browse")
325def browse_cmd(
326 data_dir: Path | None = data_dir_option,
327 use_global: bool = global_option,
328) -> None:
329 """Open the Textual TUI directly on the model catalog screen.
331 Exit codes follow the project convention: 2 for invalid flag
332 combinations (``--json`` with an interactive-only command), 1 for
333 runtime environment failures (no TTY).
334 """
335 apply_overrides(data_dir=data_dir, use_global=use_global)
336 if cfg.json_mode:
337 json_output({"error": "model browse is interactive, not available in --json mode"})
338 raise typer.Exit(2)
339 if not _is_interactive_terminal():
340 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] model browse requires a terminal.")
341 raise typer.Exit(1)
343 from lilbee.cli.tui import run_tui
345 run_tui(initial_view="Catalog")