"""Native Hermes dashboard API for managing a local Ollama instance.""" from __future__ import annotations import json import re import threading import time import uuid from datetime import datetime, timedelta, timezone from html import unescape from html.parser import HTMLParser from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import unquote, urlencode from urllib.request import Request, urlopen from zoneinfo import ZoneInfo from fastapi import APIRouter, HTTPException from pydantic import BaseModel from hermes_constants import get_hermes_home router = APIRouter() LOCAL_OLLAMA = "http://127.0.0.1:11434" REMOTE_OLLAMA = "https://ollama.com" CATALOG_FILE = "catalog.json" MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,190}$") MELBOURNE = ZoneInfo("Australia/Melbourne") POPULAR_RAM_LIMIT_GIB = 30.0 _jobs: dict[str, dict[str, Any]] = {} _jobs_lock = threading.Lock() _catalog_lock = threading.Lock() CAPABILITY_INFO = { "completion": "Text generation and chat completion.", "tools": "Tool/function calling for agent workflows.", "thinking": "Explicit reasoning/thinking output support.", "vision": "Image input and visual understanding.", "audio": "Audio input or audio-aware inference.", "video": "Video input or video-aware inference.", } FAMILY_STRENGTHS = { "qwen": ["general reasoning", "coding", "multilingual work", "tool use"], "qwen35": ["general reasoning", "coding", "long-context work", "tool use"], "gemma": ["general assistance", "reasoning", "tool use", "efficient local inference"], "nemotron": ["reasoning", "agent workflows", "long-context work", "technical tasks"], "deepseek": ["coding", "mathematical reasoning", "technical analysis"], "gpt-oss": ["general reasoning", "coding", "agent workflows"], "mistral": ["general assistance", "multilingual work", "coding"], "kimi": ["long-context work", "reasoning", "coding"], "minimax": ["agent workflows", "reasoning", "long-context work"], "glm": ["reasoning", "coding", "multilingual work"], "lfm2": ["fast local assistants", "low-resource inference", "general chat"], } def _home() -> Path: path = get_hermes_home() / "ollama-manager" path.mkdir(parents=True, exist_ok=True) return path def _json_request(url: str, method: str = "GET", payload: Any = None, timeout: int = 30) -> dict[str, Any]: data = None if payload is None else json.dumps(payload).encode("utf-8") headers = {"Accept": "application/json"} if data is not None: headers["Content-Type"] = "application/json" request = Request(url, data=data, headers=headers, method=method) with urlopen(request, timeout=timeout) as response: raw = response.read() value = json.loads(raw.decode("utf-8")) if raw else {} return value if isinstance(value, dict) else {} def _valid_name(name: str) -> str: name = str(name or "").strip() if not MODEL_RE.fullmatch(name): raise HTTPException(400, "Invalid Ollama model name") return name def _local_tags() -> list[dict[str, Any]]: try: payload = _json_request(LOCAL_OLLAMA + "/api/tags", timeout=15) models = payload.get("models", []) return [item for item in models if isinstance(item, dict)] except (HTTPError, URLError, OSError, ValueError): return [] def _local_ps() -> list[dict[str, Any]]: try: payload = _json_request(LOCAL_OLLAMA + "/api/ps", timeout=10) models = payload.get("models", []) return [item for item in models if isinstance(item, dict)] except (HTTPError, URLError, OSError, ValueError): return [] def _is_mlx(raw: dict[str, Any] | str) -> bool: text = str(raw if isinstance(raw, str) else { "name": raw.get("name") or raw.get("model"), "format": (raw.get("details") or {}).get("format"), "capabilities": raw.get("capabilities"), }).lower() return "mlx" in text or bool(re.search(r"(?:^|[-:])mlx(?:$|[-:])", text)) def _family_key(name: str) -> str: return str(name or "").split(":", 1)[0].strip() def _known_ram_fit(model: dict[str, Any]) -> bool: """Return True only when both size and RAM are known and fit this host.""" size_gb = model.get("size_gb") ram_gb = model.get("expected_ram_gb") return ( isinstance(size_gb, (int, float)) and isinstance(ram_gb, (int, float)) and float(ram_gb) <= POPULAR_RAM_LIMIT_GIB ) def _popular_fit_models( raw_rows: list[dict[str, Any]], family_rows: dict[str, list[dict[str, Any]]], catalog_view, installed_names: set[str], loaded: dict[str, Any], ) -> list[dict[str, Any]]: """Select popular models that are usable within this host's RAM budget. Oversized popular entries may be represented by the largest known smaller family variant that fits. Unknown entries are never shown. """ result: list[dict[str, Any]] = [] seen: set[str] = set() seen_footprints: set[tuple[str, float, float]] = set() seen_families: set[str] = set() for raw in raw_rows: if _is_mlx(raw): continue original = catalog_view(raw, "popular") if _known_ram_fit(original): candidate = original else: original_size = original.get("size_gb") if not isinstance(original_size, (int, float)): continue family = _family_key(original["name"]) variants: list[dict[str, Any]] = [] for variant_raw in family_rows.get(family, []): if _is_mlx(variant_raw): continue variant = _model_view(variant_raw, loaded.get(str(variant_raw.get("name") or variant_raw.get("model"))), source="popular") variant["installed"] = variant["name"] in installed_names variant["loaded"] = variant["name"] in loaded if _known_ram_fit(variant) and float(variant.get("size_gb", 0)) < float(original_size): variants.append(variant) if not variants: continue candidate = max(variants, key=lambda item: (float(item.get("size_gb", 0)), item["name"])) candidate["popular_origin"] = original["name"] candidate["popular_note"] = f"Smaller fit variant for {original['name']}" if candidate["name"] in seen: continue candidate_family = _family_key(candidate["name"]) if candidate.get("popular_origin") and candidate_family in seen_families: continue footprint = ( candidate_family, round(float(candidate.get("size_gb", 0)), 2), round(float(candidate.get("expected_ram_gb", 0)), 1), ) if footprint in seen_footprints: continue seen.add(candidate["name"]) seen_families.add(candidate_family) seen_footprints.add(footprint) result.append(candidate) for rank, row in enumerate(result, 1): row["popularity_rank"] = rank return result def _parse_number(text: Any) -> float | None: match = re.search(r"([0-9]+(?:\.[0-9]+)?)", str(text or "")) return float(match.group(1)) if match else None def _ram_estimate(size_bytes: Any, parameter_size: Any, quantization: Any, context_length: Any) -> tuple[float | None, str]: size = float(size_bytes or 0) if size > 0: base = size / (1024 ** 3) # Ollama's runtime needs allocator/graph overhead beyond the GGUF blob. estimate = base * 1.15 + 0.5 basis = "disk size × 1.15 + 0.5 GiB runtime overhead" else: params = _parse_number(parameter_size) if params is None: return None, "Unavailable: source did not publish a model size" bits = 4.5 if "Q4" in str(quantization).upper() else 8.0 estimate = params * (bits / 8.0) + 0.8 basis = "parameter/quantization estimate; source size unavailable" context = int(context_length or 0) if context >= 524288: estimate += 2.0 basis += "; includes large-context allowance" elif context >= 131072: estimate += 1.0 basis += "; includes long-context allowance" return round(estimate, 1), basis def _infer_capabilities(name: str, family: str, details: dict[str, Any], advertised: Any) -> list[str]: caps = [str(item).lower() for item in advertised or [] if item] text = f"{name} {family}".lower() if not caps or "completion" not in caps: caps.append("completion") if any(token in text for token in ("vision", "vl", "gemma4", "muse-glimmer", "qwen3.8")) and "vision" not in caps: caps.append("vision") if any(token in text for token in ("audio", "omni")) and "audio" not in caps: caps.append("audio") if "video" in text and "video" not in caps: caps.append("video") if any(token in text for token in ("thinking", "reasoning", "nemotron", "deepseek", "qwen", "kimi")) and "thinking" not in caps: caps.append("thinking") if any(token in text for token in ("tool", "agent", "qwen", "gemma", "nemotron", "deepseek", "gpt-oss")) and "tools" not in caps: caps.append("tools") order = ["completion", "tools", "thinking", "vision", "audio", "video"] return [cap for cap in order if cap in set(caps)] def _architecture(name: str, family: str, details: dict[str, Any]) -> tuple[str, bool]: text = f"{name} {family} {details.get('parent_model', '')}".lower() moe = any(token in text for token in ("moe", "mixture", "_h_", "a3b")) label = "Mixture of Experts (MoE)" if moe else "Dense / single-expert" if family: label += f" · {family}" return label, moe def _strengths(name: str, family: str, capabilities: list[str], context_length: Any) -> list[str]: text = f"{name} {family}".lower() result: list[str] = [] for key, values in FAMILY_STRENGTHS.items(): if key in text: result.extend(values) break if "tools" in capabilities: result.append("tool-enabled automation") if "vision" in capabilities: result.append("image-aware tasks") if int(context_length or 0) >= 131072: result.append("long documents and large codebases") if not result: result = ["general local inference"] return list(dict.fromkeys(result)) def _model_view(raw: dict[str, Any], loaded: dict[str, Any] | None = None, source: str = "local") -> dict[str, Any]: details = raw.get("details") if isinstance(raw.get("details"), dict) else {} name = str(raw.get("name") or raw.get("model") or "") family = str(details.get("family") or (details.get("families") or [""])[0] or "") capabilities = _infer_capabilities(name, family, details, raw.get("capabilities")) context_length = details.get("context_length") or raw.get("context_length") architecture, is_moe = _architecture(name, family, details) size_bytes = raw.get("size") or 0 ram_gb, ram_basis = _ram_estimate(size_bytes, details.get("parameter_size"), details.get("quantization_level"), context_length) loaded = loaded or {} return { "name": name, "source": source, "downloadable": source == "catalog", "installed": source == "local", "loaded": bool(loaded), "size_bytes": int(size_bytes or 0), "size_gb": round(float(size_bytes or 0) / (1024 ** 3), 2) if size_bytes else None, "size_label": raw.get("size_label") or (f"{round(float(size_bytes or 0) / (1024 ** 3), 2)} GiB" if size_bytes else "Unknown"), "loaded_bytes": int(loaded.get("size") or 0), "loaded_vram_bytes": int(loaded.get("size_vram") or 0), "digest": raw.get("digest", ""), "modified_at": raw.get("modified_at"), "family": family or "unknown", "architecture": architecture, "is_moe": is_moe, "parameter_size": details.get("parameter_size") or "unknown", "quantization": details.get("quantization_level") or "unknown", "format": details.get("format") or "unknown", "context_length": context_length, "input_modalities": raw.get("input_modalities") or (["Text", "Image"] if "vision" in capabilities else ["Text"]), "embedding_length": details.get("embedding_length"), "capabilities": capabilities, "capability_breakdown": {cap: CAPABILITY_INFO[cap] for cap in capabilities if cap in CAPABILITY_INFO}, "strengths": _strengths(name, family, capabilities, context_length), "expected_ram_gb": ram_gb, "expected_ram_label": f"{ram_gb:.1f} GiB baseline" if ram_gb is not None else "Unknown", "expected_ram_basis": ram_basis, } class _VariantPageParser(HTMLParser): """Extract the public Ollama tag rows without depending on third-party HTML packages.""" def __init__(self) -> None: super().__init__() self._depth = 0 self._parts: list[str] = [] self._href = "" self.rows: list[dict[str, Any]] = [] def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: attrs_map = dict(attrs) classes = attrs_map.get("class") or "" if tag == "div" and "group" in classes.split() and "px-4" in classes.split(): self._depth = 1 self._parts = [] self._href = "" return if self._depth: if tag == "div": self._depth += 1 if tag == "a" and (attrs_map.get("href") or "").startswith("/library/") and not self._href: self._href = attrs_map["href"] or "" def handle_data(self, data: str) -> None: if self._depth: self._parts.append(data) def handle_endtag(self, tag: str) -> None: if not self._depth or tag != "div": return self._depth -= 1 if self._depth == 0 and self._href: self.rows.append({"href": self._href, "text": " ".join("".join(self._parts).split())}) self._parts = [] self._href = "" def _text_request(url: str, timeout: int = 30) -> str: request = Request(url, headers={"Accept": "text/html"}) with urlopen(request, timeout=timeout) as response: return response.read().decode("utf-8", errors="replace") def _size_bytes(label: str) -> int: match = re.search(r"([0-9]+(?:\.[0-9]+)?)\s*(KB|MB|GB|TB)", label.upper()) if not match: return 0 multipliers = {"KB": 10**3, "MB": 10**6, "GB": 10**9, "TB": 10**12} return int(float(match.group(1)) * multipliers[match.group(2)]) def _parse_variant_page(html: str, family: str) -> list[dict[str, Any]]: parser = _VariantPageParser() parser.feed(html) rows: list[dict[str, Any]] = [] seen: set[str] = set() for item in parser.rows: href = unescape(item["href"]) name = unquote(href.rsplit("/", 1)[-1]) if not name or name in seen or not name.startswith(family + ":"): continue seen.add(name) text = item["text"] size_match = re.search(r"•\s*([0-9]+(?:\.[0-9]+)?(?:KB|MB|GB|TB))\s*•", text, re.I) context_match = re.search(r"([0-9]+)K\s+context window", text, re.I) input_match = re.search(r"([^•]+?)\s+input\s+•", text, re.I) size_label = size_match.group(1).upper() if size_match else ("cloud" if ("-cloud" in name or ":cloud" in name) else "Unknown") modalities = [part.strip() for part in (input_match.group(1).split(",") if input_match else []) if part.strip()] caps = ["vision"] if any(part.lower() == "image" for part in modalities) else [] rows.append({ "name": name, "model": name, "size": _size_bytes(size_label), "size_label": size_label, "modified_at": None, "digest": "", "details": {"family": family, "context_length": int(context_match.group(1)) * 1024 if context_match else None, "format": "gguf"}, "capabilities": caps, "input_modalities": modalities or ["Text"], "is_mlx": _is_mlx(name) or bool(re.search(r"\bMLX\b", text, re.I)), }) return [row for row in rows if not row["is_mlx"]] def _fetch_family_variants(family: str) -> list[dict[str, Any]]: try: html = _text_request(f"{REMOTE_OLLAMA}/library/{family}/tags", timeout=30) return _parse_variant_page(html, family) except (HTTPError, URLError, OSError, ValueError): return [] def _catalog_path() -> Path: return _home() / CATALOG_FILE def _read_catalog() -> dict[str, Any]: try: value = json.loads(_catalog_path().read_text(encoding="utf-8")) return value if isinstance(value, dict) else {} except (OSError, ValueError): return {} def _write_catalog(value: dict[str, Any]) -> None: path = _catalog_path() temp = path.with_suffix(".tmp") temp.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") temp.replace(path) def _refresh_family_variants(rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: families = sorted({_family_key(str(row.get("name") or row.get("model"))) for row in rows if not _is_mlx(row)}) return {family: _fetch_family_variants(family) for family in families if family} def refresh_catalog(force: bool = True) -> dict[str, Any]: with _catalog_lock: try: query = urlencode({"limit": 100, "sort": "popular"}) payload = _json_request(f"{REMOTE_OLLAMA}/api/tags?{query}", timeout=30) rows = [item for item in payload.get("models", []) if isinstance(item, dict) and not _is_mlx(item)] catalog = { "fetched_at": datetime.now(timezone.utc).isoformat(), "source": f"{REMOTE_OLLAMA}/api/tags", "models": rows, "families": _refresh_family_variants(_local_tags()), } _write_catalog(catalog) return catalog except (HTTPError, URLError, OSError, ValueError) as exc: cached = _read_catalog() if cached: cached["last_error"] = str(exc) return cached return {"fetched_at": None, "source": f"{REMOTE_OLLAMA}/api/tags", "models": [], "last_error": str(exc)} def _catalog_stale(catalog: dict[str, Any]) -> bool: raw = catalog.get("fetched_at") if not raw: return True try: fetched = datetime.fromisoformat(str(raw).replace("Z", "+00:00")) return datetime.now(timezone.utc) - fetched > timedelta(hours=20) except ValueError: return True def _ensure_catalog() -> dict[str, Any]: catalog = _read_catalog() if not catalog.get("models"): return refresh_catalog() if _catalog_stale(catalog) and not _catalog_lock.locked(): threading.Thread(target=refresh_catalog, kwargs={"force": True}, daemon=True, name="ollama-catalog-refresh").start() return catalog def _next_refresh() -> str: now = datetime.now(MELBOURNE) target = now.replace(hour=1, minute=0, second=0, microsecond=0) if now >= target: target += timedelta(days=1) return target.isoformat() def _job_snapshot() -> list[dict[str, Any]]: with _jobs_lock: return [dict(item) for item in _jobs.values()] def _set_job(job_id: str, **values: Any) -> None: with _jobs_lock: if job_id in _jobs: _jobs[job_id].update(values, updated_at=time.time()) def _run_pull(job_id: str, name: str, action: str) -> None: try: payload = json.dumps({"name": name, "stream": True}).encode("utf-8") request = Request(LOCAL_OLLAMA + "/api/pull", data=payload, headers={"Content-Type": "application/json"}, method="POST") with urlopen(request, timeout=3600) as response: for raw_line in response: try: event = json.loads(raw_line.decode("utf-8")) except ValueError: continue status = str(event.get("status") or "working") completed = int(event.get("completed") or 0) total = int(event.get("total") or 0) percent = round(completed * 100 / total, 1) if total else None _set_job(job_id, status=status, completed=completed, total=total, percent=percent, digest=event.get("digest")) if event.get("error"): raise RuntimeError(str(event["error"])) _set_job(job_id, state="completed", status="success", percent=100) except Exception as exc: _set_job(job_id, state="failed", status="error", error=str(exc)) def _run_delete(job_id: str, name: str) -> None: try: _json_request(LOCAL_OLLAMA + "/api/delete", method="DELETE", payload={"name": name}, timeout=120) _set_job(job_id, state="completed", status="deleted", percent=100) except Exception as exc: _set_job(job_id, state="failed", status="error", error=str(exc)) def _new_job(name: str, action: str) -> str: job_id = uuid.uuid4().hex with _jobs_lock: _jobs[job_id] = {"id": job_id, "name": name, "action": action, "state": "running", "status": "starting", "percent": 0, "created_at": time.time(), "updated_at": time.time()} target = _run_delete if action == "delete" else _run_pull args = (job_id, name) if action == "delete" else (job_id, name, action) threading.Thread(target=target, args=args, daemon=True, name=f"ollama-{action}-{job_id[:8]}").start() return job_id class ModelRequest(BaseModel): name: str @router.get("/status") def status() -> dict[str, Any]: tags = _local_tags() ps_rows = _local_ps() loaded = {str(row.get("name") or row.get("model")): row for row in ps_rows} local = [ _model_view(row, loaded.get(str(row.get("name") or row.get("model")))) for row in tags if not _is_mlx(row) ] catalog = _ensure_catalog() installed_names = {row["name"] for row in local} def catalog_view(raw: dict[str, Any], source: str) -> dict[str, Any]: name = str(raw.get("name") or raw.get("model")) view = _model_view(raw, loaded.get(name), source=source) view["installed"] = name in installed_names view["loaded"] = name in loaded return view downloadable = [ catalog_view(row, "catalog") for row in catalog.get("models", []) if not _is_mlx(row) and str(row.get("name") or row.get("model")) not in installed_names ] family_rows = catalog.get("families") if isinstance(catalog.get("families"), dict) else {} popular = _popular_fit_models( [row for row in catalog.get("models", []) if not _is_mlx(row)], family_rows, catalog_view, installed_names, loaded, ) for row in local: family = _family_key(row["name"]) variants = [] for raw in family_rows.get(family, []): if _is_mlx(raw): continue variant = _model_view(raw, loaded.get(raw["name"]), source="variant") variant["installed"] = variant["name"] in installed_names variant["current"] = variant["name"] == row["name"] variants.append(variant) row["variants"] = sorted(variants, key=lambda item: (item.get("size_bytes") or 0, item["name"])) return { "ollama": {"available": bool(tags or ps_rows), "version": _ollama_version(), "endpoint": LOCAL_OLLAMA}, "models": local, "popular": popular, "popular_filter": { "max_expected_ram_gib": POPULAR_RAM_LIMIT_GIB, "requires_known_size": True, "requires_known_ram": True, "smaller_fit_variants_substituted": True, }, "catalog": downloadable, "catalog_updated_at": catalog.get("fetched_at"), "catalog_source": catalog.get("source"), "catalog_error": catalog.get("last_error"), "next_catalog_refresh": _next_refresh(), "jobs": _job_snapshot(), "generated_at": time.time(), } def _ollama_version() -> str | None: try: return str(_json_request(LOCAL_OLLAMA + "/api/version", timeout=5).get("version") or "unknown") except Exception: return None @router.post("/catalog/refresh") def catalog_refresh() -> dict[str, Any]: catalog = refresh_catalog() return {"ok": bool(catalog.get("models")), "updated_at": catalog.get("fetched_at"), "count": len(catalog.get("models", [])), "error": catalog.get("last_error")} @router.post("/pull") def pull_model(body: ModelRequest) -> dict[str, Any]: name = _valid_name(body.name) return {"ok": True, "job_id": _new_job(name, "download"), "message": f"Downloading or updating {name}"} @router.post("/redownload") def redownload_model(body: ModelRequest) -> dict[str, Any]: name = _valid_name(body.name) return {"ok": True, "job_id": _new_job(name, "redownload"), "message": f"Re-downloading or updating {name}"} @router.delete("/model") def delete_model(body: ModelRequest) -> dict[str, Any]: name = _valid_name(body.name) return {"ok": True, "job_id": _new_job(name, "delete"), "message": f"Removing {name}"} def create_ollama_routes(app) -> None: app.include_router(router, prefix="/api/plugins/ollama-manager")