mission-control-v2/backend/services/catalog.py

"""
Kuratierter Modell-Katalog ("Cookbook", inspiriert von Odysseus): EINE Quelle der
Wahrheit für KORREKTE Metadaten (total/active params, moe, generation) statt
Namens-Raterei. Macht Empfehlung + Upgrade-Erkennung präzise und MoE-bewusst
für die bandbreiten-limitierte Strix-Halo-Box.

Daten: backend/models_catalog.json. Fällt sanft aus (leerer Katalog), wenn die
Datei fehlt → discover nutzt dann nur die HF-Dynamik.
"""

import json
import math
import os
import re

from services.fit import estimate_memory_gb, estimate_speed

_CATALOG_PATH = os.path.join(os.path.dirname(__file__), "..", "models_catalog.json")
_cache: dict = {"mtime": 0.0, "models": []}


def _load() -> list[dict]:
    try:
        mt = os.path.getmtime(_CATALOG_PATH)
        if mt != _cache["mtime"]:
            with open(_CATALOG_PATH, encoding="utf-8") as f:
                data = json.load(f) or {}
            _cache.update(mtime=mt, models=[m for m in data.get("models", []) if m.get("name")])
    except (OSError, ValueError):
        _cache.update(mtime=0.0, models=[])
    return _cache["models"]


def _norm(name: str) -> str:
    """Vergleichs-Stamm: kleingeschrieben, Org-Prefix/Quant/GGUF/Split entfernt."""
    s = (name or "").lower().split("/")[-1]
    s = re.sub(r"\.gguf$", "", s)
    s = re.sub(r"-\d+-of-\d+$", "", s)
    s = re.sub(r"[-_](ud-)?(i?q\d[\w]*|f16|bf16|fp16|f32|mxfp4)$", "", s)
    return s.strip("-_ ")


def entries() -> list[dict]:
    return list(_load())


def entries_for_role(role: str) -> list[dict]:
    return [e for e in _load() if e.get("role") == role]


def meta_for_name(name: str) -> dict | None:
    """Katalog-Metadaten zu einem Modell(namen) — matcht lokalen Namen ODER HF-Repo."""
    n = _norm(name)
    if not n:
        return None
    for e in _load():
        cand = {_norm(e.get("name", "")), _norm(e.get("repo", ""))}
        if n in cand or any(c and (c in n or n in c) for c in cand):
            return e
    return None


def fit_of(e: dict, ram_gb: float) -> dict:
    """Hardware-Fit eines Katalog-Eintrags (MoE-bewusst über active_params_b)."""
    total = float(e.get("total_params_b") or 7)
    active = float(e.get("active_params_b") or total)
    quant = e.get("quant") or "Q4_K_M"
    ctx = int(e.get("ctx") or 32768)
    req_gb = estimate_memory_gb(total, quant, ctx)
    tps = estimate_speed(req_gb, ram_gb, (active / total) if total else 1.0)
    usable = max(ram_gb - 4.0, 0)
    if req_gb > usable:
        level, text = "too_tight", "Zu groß (OOM)"
    elif req_gb > usable * 0.8:
        level, text = "marginal", "Könnte knapp werden"
    else:
        level, text = "perfect", "Passt perfekt"
    return {"level": level, "text": text, "req_gb": round(req_gb, 1), "tps": round(tps, 0)}


def stack_score(e: dict, ram_gb: float) -> float:
    """Score für DIESE Hardware: muss passen, dann Wissen (total params) + Tempo
    (tps — belohnt MoE durch niedrige aktive Params automatisch). Bandbreiten-Box
    → MoE gewinnt bei vergleichbarem Wissen gegen dense."""
    fit = fit_of(e, ram_gb)
    if fit["level"] == "too_tight":
        return -100.0
    total = float(e.get("total_params_b") or 7)
    fit_bonus = 3.0 if fit["level"] == "perfect" else 1.0
    knowledge = math.log2(total + 1) / 8.0          # ~0..1 (bis ~256B)
    speed = min((fit["tps"] or 0) / 80.0, 1.0)       # normalisiert; MoE = hohe tps
    return fit_bonus + 1.2 * knowledge + 1.0 * speed


def to_model_dict(e: dict, ram_gb: float) -> dict:
    """Katalog-Eintrag → discover-kompatibles Modell-Dict (echte Metadaten)."""
    total = float(e.get("total_params_b") or 7)
    active = e.get("active_params_b")
    repo = e.get("repo") or e.get("name")
    role = e.get("role")
    caps = {
        "moe": bool(e.get("moe")), "active_b": active,
        "tools": "yes" if e.get("tools") else "no",
        "vision": bool(e.get("vision")), "coder": role == "coder",
        "reasoning": role == "heavy", "embedding": False,
        "ctx": e.get("ctx"), "params_b": total, "arch": e.get("family"),
    }
    return {
        "name": e.get("name"), "author": repo.split("/")[0] if "/" in repo else "catalog",
        "repo": repo, "role": role, "params_b": total, "active_b": active,
        "moe": bool(e.get("moe")), "generation": e.get("generation"),
        "family": e.get("family"), "quant": e.get("quant") or "Q4_K_M",
        "tags": ["catalog"], "downloads": 0, "fit": fit_of(e, ram_gb),
        "optimal_ctx": int(e.get("ctx") or 32768), "caps": caps, "curated": True,
    }