#!/usr/bin/env python3 """ Download LEAP bundle outputs by request ID and inspect the files. Per Liquid AI docs: leap-bundle create produces .gguf (default, GGUF) or .bundle (ExecuTorch, with --executorch). This script finds and inspects both. Uses leap-bundle list (--json per request) and leap-bundle download with --output-path. Reports all bundle artifacts (.gguf, .bundle) and optionally runs short inference on .gguf (llama-cpp). Requires: pip install leap-bundle LEAP auth: leap-bundle login """ import argparse import json import os import re import subprocess import sys from pathlib import Path def _leap_env() -> dict[str, str]: env = os.environ.copy() env["PYTHONUTF8"] = "1" return env def run(cmd: list[str], capture: bool = True, cwd: Path | None = None) -> subprocess.CompletedProcess: kwargs = { "cwd": str(cwd) if cwd else None, "text": True, "encoding": "utf-8", "errors": "replace", "env": _leap_env(), } if capture: kwargs["capture_output"] = True return subprocess.run(cmd, **kwargs) def list_request(request_id: str) -> dict | None: """Get details for one request; returns parsed JSON or None.""" r = run(["leap-bundle", "list", str(request_id), "--json"], capture=True) if r.returncode != 0: return None out = (r.stdout or "").strip() try: return json.loads(out) except json.JSONDecodeError: return None def list_all_request_ids() -> list[str]: """Run leap-bundle list (no id) and parse table for request IDs. Returns list of ID strings.""" r = run(["leap-bundle", "list"], capture=True) out = (r.stdout or r.stderr or "") ids: list[str] = [] # Table rows: first column is often the ID (integer) for line in out.splitlines(): parts = line.split() if parts and parts[0].isdigit(): ids.append(parts[0]) # Fallback: any line with a pipe or spaces and a leading number (rich table) if not ids: for line in out.splitlines(): m = re.search(r"[\|\s](\d{1,6})[\|\s]", line) if m: ids.append(m.group(1)) # Fallback: JSON-like "request_id": N or "id": N if not ids: for m in re.finditer(r'"(?:request_id|id)"\s*:\s*(\d+)', out): ids.append(m.group(1)) return list(dict.fromkeys(ids)) def get_status(data: dict) -> str: """Extract status from list request JSON.""" s = (data.get("status") or data.get("Status") or "").lower() return s def download_bundle(request_id: str, output_path: Path) -> tuple[bool, str]: """Run leap-bundle download --output-path . Returns (success, stderr_or_empty).""" output_path.mkdir(parents=True, exist_ok=True) r = run( ["leap-bundle", "download", str(request_id), "--output-path", str(output_path)], capture=True, ) err = (r.stderr or r.stdout or "").strip() return r.returncode == 0, err # Per Liquid AI docs: create produces .gguf (default) or .bundle (--executorch) BUNDLE_EXTENSIONS = (".gguf", ".bundle") def find_bundle_files(root: Path) -> list[Path]: """Return all LEAP bundle artifact files (.gguf, .bundle) under root.""" out: list[Path] = [] for ext in BUNDLE_EXTENSIONS: out.extend(root.rglob(f"*{ext}")) return sorted(out) def inspect_file(path: Path, run_inference: bool = False, root: Path | None = None) -> None: """Print path, size, type; run short inference only for .gguf (llama-cpp).""" size_mb = path.stat().st_size / (1024**2) try: disp = path.relative_to(root) if root else path except ValueError: disp = path kind = "GGUF" if path.suffix == ".gguf" else "ExecuTorch (.bundle)" print(f" {disp} {size_mb:.1f} MB [{kind}]") if run_inference and path.suffix == ".gguf": try: from llama_cpp import Llama print(" Running short inference (llama-cpp)...") llm = Llama(model_path=str(path), n_ctx=256, verbose=False) out = llm("Bonjour, une phrase en français.\n", max_tokens=24, temperature=0.3) text = (out["choices"][0]["text"] or "").strip() print(f" -> {text[:150]}") except ImportError: print(" (Install llama-cpp-python to run inference)") except Exception as e: print(f" Inference error: {e}") elif run_inference and path.suffix == ".bundle": print(" (ExecuTorch .bundle; inference via LEAP SDK, not llama-cpp)") def main() -> int: p = argparse.ArgumentParser( description="Download LEAP bundle outputs by request ID and inspect files (.gguf or .bundle per Liquid AI docs).", epilog="Requires: leap-bundle (pip install leap-bundle). Auth: leap-bundle login ", ) p.add_argument( "--output-dir", type=Path, default=Path("./luth_bundle_downloads"), help="Directory to download each bundle into (default: ./luth_bundle_downloads)", ) p.add_argument( "--request-ids", type=str, nargs="*", metavar="ID", help="Bundle request IDs to download (e.g. 1 2 3)", ) p.add_argument( "--from-file", type=Path, metavar="FILE", help="Text file with one request ID per line", ) p.add_argument( "--list", action="store_true", help="Run leap-bundle list and download all completed requests", ) p.add_argument( "--infer", action="store_true", help="Run a short inference on each downloaded GGUF (requires llama-cpp-python)", ) p.add_argument( "--inspect-only", action="store_true", help="Only inspect existing bundle files (.gguf, .bundle) under --output-dir; do not download", ) args = p.parse_args() args.output_dir = args.output_dir.resolve() request_ids: list[str] = [] if args.inspect_only: args.output_dir.mkdir(parents=True, exist_ok=True) bundles = find_bundle_files(args.output_dir) print(f"Inspecting {len(bundles)} bundle file(s) (.gguf / .bundle) under {args.output_dir}\n") for f in bundles: inspect_file(f, run_inference=args.infer, root=args.output_dir) return 0 if args.list: print("Fetching bundle request list...") request_ids = list_all_request_ids() if not request_ids: print("No request IDs found from list.", file=sys.stderr) print("If you have existing bundle requests (e.g. from bundle_luth.py --all), run:", file=sys.stderr) print(" python download_bundles.py --request-ids 1 2 3 4 5", file=sys.stderr) return 1 print(f"Found {len(request_ids)} request(s): {request_ids}") else: if args.request_ids: request_ids.extend(args.request_ids) if args.from_file: if not args.from_file.exists(): print(f"File not found: {args.from_file}", file=sys.stderr) return 1 for line in args.from_file.read_text(encoding="utf-8", errors="replace").splitlines(): rid = line.strip() if rid and rid.isdigit(): request_ids.append(rid) if not request_ids: print("Provide --request-ids, --from-file, or --list.", file=sys.stderr) return 1 args.output_dir.mkdir(parents=True, exist_ok=True) downloaded: list[Path] = [] for rid in request_ids: print(f"\n--- Request ID {rid} ---") info = list_request(rid) status = get_status(info) if info else "" if status: print(f" Status: {status}") if "completed" not in status and "complete" not in status: print(" Skipping (not completed).") continue else: print(" (Status unknown; attempting download.)") dest = args.output_dir / f"request_{rid}" print(f" Downloading to {dest} ...") ok, err = download_bundle(rid, dest) if ok: for f in find_bundle_files(dest): downloaded.append(f) kind = "GGUF" if f.suffix == ".gguf" else ".bundle" print(f" Downloaded: {f.name} ({f.stat().st_size / (1024**2):.1f} MB) [{kind}]") else: print(" Download failed.", file=sys.stderr) if "signed_url" in err: print(" (LEAP CLI/API 'signed_url' error – try again later or check LEAP status.)", file=sys.stderr) print("\n" + "=" * 60) print("Inspection summary (bundle artifacts: .gguf / .bundle)") print("=" * 60) all_bundles = find_bundle_files(args.output_dir) for f in all_bundles: inspect_file(f, run_inference=args.infer, root=args.output_dir) if not all_bundles: print(" No bundle files (.gguf or .bundle) found.") return 0 if __name__ == "__main__": sys.exit(main())