Coverage for src/lilbee/cli/model.py: 100%
174 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-28 17:20 +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
10import sys
11from pathlib import Path
12from typing import TYPE_CHECKING
14import typer
15from rich.progress import BarColumn, Progress, TextColumn, TimeRemainingColumn
16from rich.table import Table
17from rich.text import Text
19from lilbee.app.models import (
20 ListModelsResult,
21 PullEvent,
22 PullProgressEvent,
23 PullResult,
24 PullStatus,
25 ShowModelResult,
26 list_models_data,
27 pull_model_data,
28 remove_model_data,
29 show_model_data,
30)
31from lilbee.cli import theme
32from lilbee.cli.app import (
33 apply_overrides,
34 console,
35 data_dir_option,
36 global_option,
37)
38from lilbee.cli.helpers import json_output, print_prefixed
39from lilbee.core.config import cfg
40from lilbee.runtime.console import PlainConsole, styled
41from lilbee.runtime.progress.columns import literal_text_column
43if TYPE_CHECKING:
44 from collections.abc import Callable
46 from lilbee.catalog import DownloadProgress
47 from lilbee.catalog.types import ModelSource, ModelTask
50def _render_list(data: ListModelsResult) -> Table:
51 table = Table(title="Installed models")
52 table.add_column("Name", style=theme.ACCENT)
53 table.add_column("Source", style=theme.MUTED)
54 table.add_column("Task")
55 table.add_column("Size", justify="right")
56 for entry in data.models:
57 size = f"{entry.size_gb:.2f} GB" if entry.size_gb is not None else ""
58 table.add_row(Text(entry.name), entry.source, entry.task or "", size)
59 return table
62def _render_show(data: ShowModelResult) -> Text:
63 lines: list[str] = []
64 if data.catalog is not None:
65 lines.extend(
66 [
67 f" display_name: {data.catalog.display_name}",
68 f" task: {data.catalog.task}",
69 f" size_gb: {data.catalog.size_gb}",
70 f" min_ram_gb: {data.catalog.min_ram_gb}",
71 f" hf_repo: {data.catalog.hf_repo}",
72 f" description: {data.catalog.description}",
73 ]
74 )
75 lines.append(f" installed: {data.installed}")
76 if data.source:
77 lines.append(f" source: {data.source}")
78 if data.path:
79 lines.append(f" path: {data.path}")
80 if data.manifest is not None:
81 lines.append(f" downloaded: {data.manifest.downloaded_at}")
82 return styled((data.model, theme.ACCENT), *(f"\n{line}" for line in lines))
85model_app = typer.Typer(
86 name="model",
87 help="Manage installed and available models (pull / list / show / rm / browse).",
88 no_args_is_help=True,
89)
91_source_option = typer.Option(
92 None,
93 "--source",
94 "-s",
95 help="Filter by source: native, remote, ollama, lm_studio, or frontier (default: all).",
96)
97_task_option = typer.Option(
98 None,
99 "--task",
100 "-t",
101 help="Filter by task: 'chat', 'embedding', 'vision', or 'rerank'.",
102)
103_yes_option = typer.Option(
104 False,
105 "--yes",
106 "-y",
107 help="Skip confirmation prompt.",
108)
111def _parse_source_or_bad_param(value: str | None) -> ModelSource | None:
112 """Parse a CLI --source value, raising typer.BadParameter on bad input."""
113 from lilbee.catalog.types import ModelSource
115 try:
116 return ModelSource.parse(value)
117 except ValueError as exc:
118 if cfg.json_mode:
119 json_output({"error": str(exc)})
120 raise SystemExit(1) from None
121 raise typer.BadParameter(str(exc)) from exc
124def _parse_task_or_bad_param(value: str | None) -> ModelTask | None:
125 """Parse a CLI --task value, mirroring --source's JSON-envelope + friendly error.
127 Without this, an invalid --task leaked Python's "'x' is not a valid ModelTask"
128 and broke --json by emitting Typer usage text instead of the error envelope.
129 """
130 from lilbee.catalog.types import ModelTask
132 if not value:
133 return None
134 try:
135 return ModelTask(value)
136 except ValueError:
137 allowed = ", ".join(t.value for t in ModelTask)
138 msg = f"invalid task {value!r}; expected one of: {allowed}"
139 if cfg.json_mode:
140 json_output({"error": msg})
141 raise SystemExit(1) from None
142 raise typer.BadParameter(msg) from None
145def _apply_catalog_overrides(*, data_dir: Path | None, use_global: bool) -> None:
146 """Apply CLI overrides for a catalog command, with the fleet left cold.
148 Catalog commands list, read, download and delete model files; none of them
149 runs inference, so none should pay for a fleet warm (or its alarming
150 warm-up traceback on a slow host). Setting it here rather than at each
151 command means a catalog command added later cannot silently eager-start.
152 """
153 apply_overrides(data_dir=data_dir, use_global=use_global)
154 cfg.worker_pool_eager_start = False
157@model_app.command("list")
158def list_cmd(
159 source: str | None = _source_option,
160 task: str | None = _task_option,
161 data_dir: Path | None = data_dir_option,
162 use_global: bool = global_option,
163) -> None:
164 """List installed models across all sources."""
165 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global)
166 parsed_task = _parse_task_or_bad_param(task)
167 data = list_models_data(source=_parse_source_or_bad_param(source), task=parsed_task)
168 if cfg.json_mode:
169 json_output(data.model_dump())
170 return
171 if not data.models:
172 console.print("No models installed.")
173 return
174 console.print(_render_list(data))
177@model_app.command("show")
178def show_cmd(
179 ref: str = typer.Argument(..., help="Model ref (e.g. 'Qwen/Qwen3-0.6B-GGUF')."),
180 data_dir: Path | None = data_dir_option,
181 use_global: bool = global_option,
182) -> None:
183 """Show catalog and installed metadata for a model."""
184 from lilbee.modelhub.model_manager import ModelNotFoundError
186 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global)
187 try:
188 data = show_model_data(ref)
189 except ModelNotFoundError as exc:
190 if cfg.json_mode:
191 json_output({"error": str(exc)})
192 else:
193 console.print(Text(str(exc), style=theme.ERROR), soft_wrap=True)
194 raise typer.Exit(1) from None
195 if cfg.json_mode:
196 json_output(data.model_dump())
197 return
198 console.print(_render_show(data))
201def _run_pull(
202 ref: str,
203 src: ModelSource,
204 on_update: Callable[[DownloadProgress], None],
205 *,
206 allow_unsupported: bool = False,
207) -> PullResult:
208 """Invoke ``pull_model_data`` and translate known errors to typer.Exit."""
209 from lilbee.catalog.compat import UnsupportedArchError
211 try:
212 return pull_model_data(ref, src, on_update=on_update, allow_unsupported=allow_unsupported)
213 except UnsupportedArchError as exc:
214 msg = (
215 f"Architecture {exc.architecture!r} is not supported by this lilbee build.\n"
216 "Pass --allow-unsupported to try anyway."
217 )
218 if cfg.json_mode:
219 json_output(
220 {
221 "error": "unsupported_arch",
222 "arch": exc.architecture,
223 "ref": exc.ref,
224 }
225 )
226 else:
227 print_prefixed(console, "Error: ", msg, style=theme.ERROR)
228 raise typer.Exit(1) from None
229 except (RuntimeError, PermissionError) as exc:
230 if cfg.json_mode:
231 json_output({"error": str(exc)})
232 else:
233 print_prefixed(console, "Error: ", exc, style=theme.ERROR)
234 raise typer.Exit(1) from None
237def _pull_json_stream(ref: str, src: ModelSource, *, allow_unsupported: bool) -> None:
238 """Emit newline-delimited JSON progress events, then the final result."""
240 def on_update(p: DownloadProgress) -> None:
241 event = PullProgressEvent(
242 model=ref, percent=p.percent, detail=p.detail, cache_hit=p.is_cache_hit
243 )
244 json_output(event.model_dump())
246 final = _run_pull(ref, src, on_update, allow_unsupported=allow_unsupported)
247 json_output({**final.model_dump(), "event": PullEvent.DONE.value})
250def _pull_interactive_progress(ref: str, src: ModelSource, *, allow_unsupported: bool) -> None:
251 """Drive Rich's Live progress bar during a native HuggingFace download."""
252 err_console = PlainConsole(stderr=True, force_terminal=True)
253 with Progress(
254 literal_text_column("{task.description}", style="progress.description"),
255 BarColumn(),
256 TextColumn("{task.percentage:>3.0f}%"),
257 literal_text_column("{task.fields[detail]}"),
258 TimeRemainingColumn(),
259 console=err_console,
260 transient=False,
261 ) as progress:
262 task_id = progress.add_task(f"Downloading {ref}", total=100, detail="")
264 def on_update(p: DownloadProgress) -> None:
265 progress.update(task_id, completed=p.percent, detail=p.detail)
267 final = _run_pull(ref, src, on_update, allow_unsupported=allow_unsupported)
269 if final.status == PullStatus.ALREADY_INSTALLED:
270 console.print(Text(f"{ref} is already installed."), soft_wrap=True)
271 else:
272 console.print(Text.assemble("Pulled ", (ref, theme.ACCENT), "."), soft_wrap=True)
275@model_app.command("pull")
276def pull_cmd(
277 ref: str = typer.Argument(..., help="Model ref to download (e.g. 'Qwen/Qwen3-0.6B-GGUF')."),
278 source: str = typer.Option(
279 "native",
280 "--source",
281 "-s",
282 help="Pull from 'native' (HuggingFace GGUF) or 'remote' (SDK-managed).",
283 ),
284 allow_unsupported: bool = typer.Option(
285 False,
286 "--allow-unsupported",
287 help="Pull even if the architecture isn't in the supported set (load may still fail).",
288 ),
289 data_dir: Path | None = data_dir_option,
290 use_global: bool = global_option,
291) -> None:
292 """Download a model."""
293 from lilbee.catalog.types import ModelSource
295 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global)
296 src = _parse_source_or_bad_param(source) or ModelSource.NATIVE
297 if cfg.json_mode:
298 _pull_json_stream(ref, src, allow_unsupported=allow_unsupported)
299 else:
300 _pull_interactive_progress(ref, src, allow_unsupported=allow_unsupported)
303def _confirm_remove_or_exit(ref: str, yes: bool) -> None:
304 if yes or cfg.json_mode:
305 return
306 if not typer.confirm(f"Remove {ref}?", default=False):
307 console.print("Aborted.")
308 raise typer.Exit(0)
311@model_app.command("rm")
312def rm_cmd(
313 ref: str = typer.Argument(..., help="Model ref to remove."),
314 source: str | None = _source_option,
315 yes: bool = _yes_option,
316 data_dir: Path | None = data_dir_option,
317 use_global: bool = global_option,
318) -> None:
319 """Remove an installed model."""
320 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global)
321 src = _parse_source_or_bad_param(source)
322 _confirm_remove_or_exit(ref, yes)
323 try:
324 data = remove_model_data(ref, source=src)
325 except ValueError as exc:
326 if cfg.json_mode:
327 json_output({"error": str(exc)})
328 else:
329 console.print(Text(str(exc), style=theme.ERROR), soft_wrap=True)
330 raise typer.Exit(1) from None
331 if cfg.json_mode:
332 json_output(data.model_dump())
333 if not data.deleted:
334 raise typer.Exit(1)
335 return
336 if not data.deleted:
337 console.print(Text(f"Not found: {ref}", style=theme.WARNING), soft_wrap=True)
338 raise typer.Exit(1)
339 suffix = f" ({data.freed_gb:.2f} GB freed)" if data.freed_gb else ""
340 console.print(Text.assemble("Removed ", (ref, theme.ACCENT), suffix, "."), soft_wrap=True)
343def _is_interactive_terminal() -> bool:
344 """Return True when both stdin and stdout are connected to a TTY.
346 Extracted as a module-level helper so tests can patch it deterministically;
347 CliRunner replaces ``sys.stdin`` during invoke which makes direct
348 monkey-patching of ``sys.stdin.isatty`` unreliable.
349 """
350 return sys.stdin.isatty() and sys.stdout.isatty()
353@model_app.command("browse")
354def browse_cmd(
355 data_dir: Path | None = data_dir_option,
356 use_global: bool = global_option,
357) -> None:
358 """Open the Textual TUI directly on the model catalog screen.
360 Exit codes follow the project convention: 2 for invalid flag
361 combinations (``--json`` with an interactive-only command), 1 for
362 runtime environment failures (no TTY).
363 """
364 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global)
365 if cfg.json_mode:
366 json_output({"error": "model browse is interactive, not available in --json mode"})
367 raise typer.Exit(2)
368 if not _is_interactive_terminal():
369 console.print(styled(("Error:", theme.ERROR), " model browse requires a terminal."))
370 raise typer.Exit(1)
372 from lilbee.cli.tui import run_tui
373 from lilbee.cli.tui.messages import CATALOG_VIEW
375 run_tui(initial_view=CATALOG_VIEW)