app/recommend.py

"""Ranking: for each model, the best quant for your hardware, then the models themselves.

Every quant of every repo holding the model is fitted to the hardware; the best is the
one that runs most comfortably (all on GPU, then experts in RAM, then more layers on
GPU, then CPU), and among equally comfortable ones the one that matches the priority.
Models are then ranked on that fit, popularity, recency and community reports.

Nothing here looks at who uploaded a quant beyond its download count.
"""

from __future__ import annotations

import math
from dataclasses import dataclass, field
from datetime import datetime, timezone

from .catalogue import Group
from .detail import ModelDetail
from .fit import Fit, fit, llama_cpp_command, ollama_command
from .geometry import estimate_without_header
from .hardware import Hardware
from .parsing import QUANT_BPW, QuantFile, min_model_bytes, quant_bits

PRIORITIES = ["balanced", "quality", "small"]
SORTS = ["best", "popular", "recent", "community"]
WEIGHTS = {"fit": 0.35, "capacity": 0.20, "popularity": 0.25, "recency": 0.10, "community": 0.10}


@dataclass
class Query:
    hardware: Hardware | None
    context: int = 8192
    kv_type: str = "f16"
    priority: str = "balanced"
    min_bits: int = 0  # skip quants below this bit level (Q4 and up: 4)
    vram_only: bool = False  # discrete GPUs: nothing may spill into system RAM
    mmproj: str = "gpu"      # vision projector: "gpu" (llama-server default), "ram" or "off"


@dataclass
class Option:
    repo: str
    repo_downloads: int
    quant: QuantFile
    fit: Fit | None
    estimated: bool
    projector_bytes: int | None = None

    def to_dict(self, q: Query) -> dict:
        d = {"repo": self.repo, "repo_downloads": self.repo_downloads, **self.quant.to_dict(),
             "estimated": self.estimated, "fit": self.fit.to_dict() if self.fit else None,
             "projector_size": self.projector_bytes,
             "url": f"https://huggingface.co/{self.repo}"}
        if self.fit:
            proj = q.mmproj if self.projector_bytes and q.mmproj != "gpu" else None
            d["commands"] = {"llama.cpp": llama_cpp_command(self.repo, self.quant.quant, self.fit, q.context, q.kv_type, proj),
                             "ollama": ollama_command(self.repo, self.quant.quant)}
        else:
            d["commands"] = {"llama.cpp": f"llama-server -hf {self.repo}:{self.quant.quant} -c {q.context}",
                             "ollama": ollama_command(self.repo, self.quant.quant)}
        return d


def preference(opt: Option, priority: str) -> float:
    b = opt.quant.bpw or 4.85
    if priority == "quality":
        return min(b, 8.5)
    if priority == "small":
        return -(opt.quant.size or 0) / 1e9
    return -abs(b - 5.0) if b >= 3.5 else -3.0 - (3.5 - b)  # Q4_K_M .. Q5_K_M


def below_floor(qf: QuantFile, params: int | None, min_bits: int) -> bool:
    if not min_bits:
        return False
    bits = qf.bits
    if bits is None and params and qf.size:
        bits = quant_bits("", qf.size * 8 / params)
    return bits is not None and bits < min_bits


PLACEHOLDER = {0: "Q4_K_M", 1: "Q4_K_M", 2: "Q4_K_M", 3: "Q4_K_M", 4: "Q4_K_M", 5: "Q5_K_M", 6: "Q6_K", 8: "Q8_0"}


def options_for(group: Group, detail: ModelDetail | None, q: Query) -> list[Option]:
    opts: list[Option] = []
    geo = detail.geometry if detail else None
    n_vocab = detail.n_vocab if detail else None
    downloads = {r.id: r.downloads for r in group.repos}
    if detail:
        for repo, rf in detail.repos.items():
            for qf in rf.quants:
                if not qf.size or qf.size < min_model_bytes(group.params):
                    continue  # a draft model, head or encoder that survived the name checks
                if below_floor(qf, group.params, q.min_bits):
                    continue
                if geo is not None:
                    g = geo.scaled(qf.size, group.params)
                    est = False
                elif group.params:
                    g = estimate_without_header(group.params, qf.bpw or 4.85, qf.size, group.context_length, group.moe)
                    est = True
                else:
                    continue
                f = fit(g, q.hardware, q.context, q.kv_type, n_vocab, rf.mmproj_bytes or 0, q.vram_only, q.mmproj) if q.hardware else None
                opts.append(Option(repo, downloads.get(repo, 0), qf, f, est, rf.mmproj_bytes))
    # A size guessed from the parameter count, only while the files are unknown. Once they are
    # listed and none is a usable quant (a drafter, a projector), there is nothing to guess about.
    files_known = bool(detail and detail.repos and all(rf.error is None for rf in detail.repos.values()))
    if not opts and group.params and group.repos and not files_known:
        lead = group.repos[0]
        name = PLACEHOLDER.get(q.min_bits, "Q8_0")
        qf = QuantFile(name, [], int(group.params * QUANT_BPW[name] / 8), QUANT_BPW[name])
        g = (geo.scaled(qf.size, group.params) if geo else
             estimate_without_header(group.params, qf.bpw, qf.size, group.context_length, group.moe))
        f = fit(g, q.hardware, q.context, q.kv_type, n_vocab, vram_only=q.vram_only) if q.hardware else None
        opts.append(Option(lead.id, lead.downloads, qf, f, geo is None))
    return opts


def best_option(opts: list[Option], priority: str) -> Option | None:
    if not opts:
        return None
    return max(opts, key=lambda o: ((o.fit.score if o.fit else 0.0), preference(o, priority), o.repo_downloads))


@dataclass
class Scored:
    group: Group
    best: Option | None
    options: list[Option]
    score: float
    components: dict
    reasons: list[str] = field(default_factory=list)
    detail_loaded: bool = False
    detail_repos: dict | None = None

    def to_dict(self, q: Query, with_options: bool = False) -> dict:
        d = {**self.group.summary(), "score": round(self.score, 1), "components": self.components,
             "reasons": self.reasons, "detail_loaded": self.detail_loaded,
             "best": self.best.to_dict(q) if self.best else None}
        if with_options:
            d["options"] = [o.to_dict(q) for o in self.options]
        return d


def _age_days(ts: str | None) -> float | None:
    if not ts:
        return None
    try:
        t = datetime.fromisoformat(ts.replace("Z", "+00:00"))
        return max((datetime.now(timezone.utc) - t).total_seconds() / 86400, 0)
    except ValueError:
        return None


class Engine:
    version = "fit-v1"

    @staticmethod
    def _capacity(g: Group, best: Option | None, q: Query) -> float:
        """Bigger models that still run well rank higher: a 27B that fits beats a 1B that fits.
        Scaled to what the hardware could hold at ~4.5 bits; neutral without hardware."""
        if q.hardware is None or not g.params or best is None or best.fit is None or best.fit.tier in ("no", "no_vram"):
            return 0.5 if q.hardware is None else 0.0
        mem = q.hardware.usable_vram_gb() if q.hardware.kind != "cpu" else q.hardware.ram_gb * 0.8
        if q.hardware.kind == "gpu":
            mem += q.hardware.ram_gb * 0.25  # experts or layers can spill into RAM
        reference = max(mem * 1.6e9, 1e9)   # params that fit in `mem` at ~5 bits
        return round(min(math.log1p(g.params / 1e9) / math.log1p(reference / 1e9), 1.0), 3)

    def score(self, groups: list[Group], q: Query, details: dict[str, ModelDetail], community: dict) -> list[Scored]:
        max_dl = max((g.downloads for g in groups), default=1) or 1
        out = []
        for g in groups:
            det = details.get(g.key)
            opts = options_for(g, det, q)
            best = best_option(opts, q.priority)
            comp, reasons = {}, []
            if q.hardware is None or best is None or best.fit is None:
                comp["fit"] = 0.5
                if q.hardware is None:
                    reasons.append("Pick your hardware to see what fits.")
            else:
                comp["fit"] = round(best.fit.score / 5, 3)
                where = f"{best.quant.quant} from {best.repo.split('/')[0]}"
                reasons.append(f"{best.fit.label}: {where} ({(best.quant.size or 0) / 1e9:.1f} GB)"
                               + (" — estimated from the parameter count" if best.estimated else ""))
            comp["capacity"] = self._capacity(g, best, q)
            comp["popularity"] = round(math.log1p(g.downloads) / math.log1p(max_dl), 3)
            age = _age_days(g.created) or _age_days(g.modified)
            comp["recency"] = round(math.exp(-age / 120), 3) if age is not None else 0.3
            c = community.get(g.key)
            if c and (c["feedback"] or c["benchmarks"]):
                share = c["positive_share"] if c["positive_share"] is not None else 0.6
                comp["community"] = round(min(1.0, (c["feedback"] + c["benchmarks"]) / 5) * share, 3)
                reasons.append(f"{c['feedback']} report(s), {c['benchmarks']} benchmark(s) from the community")
            else:
                comp["community"] = 0.0
            total = 100 * sum(WEIGHTS[k] * v for k, v in comp.items())
            out.append(Scored(g, best, opts, total, comp, reasons, bool(det and det.repos), det.repos if det else None))
        return out


def sort_results(scored: list[Scored], sort: str, community: dict | None = None) -> list[Scored]:
    if sort == "popular":
        return sorted(scored, key=lambda s: -s.group.downloads)
    if sort == "recent":
        return sorted(scored, key=lambda s: s.group.created or s.group.modified or "", reverse=True)
    if sort == "community":
        c = community or {}
        return sorted(scored, key=lambda s: (-(c.get(s.group.key, {}).get("feedback", 0)
                                                + c.get(s.group.key, {}).get("benchmarks", 0)), -s.score))
    return sorted(scored, key=lambda s: (-s.score, -s.group.downloads))