app.py

"""
RunLocalAI catalog + leaderboard — HuggingFace Space.

Two live, read-only surfaces over the RunLocalAI corpus:

  1. 🏆 Benchmark leaderboard — reproducible quality scores
     (HumanEval+, MBPP+, TurkishMMLU) ranked per benchmark, each row
     carrying provenance: run log, reproduction command, first-party /
     community status. Source: GET /api/v1/quality-benchmarks.

  2. 🛠️ Model catalog — every open-weight model worth running locally,
     with license tone, params, context, and vendor. Source:
     GET /api/v1/models.

Both endpoints are public, keyless, CC-BY-4.0. The catalog at
runlocalai.co is the source of truth — this Space is a discovery surface
for the HuggingFace community. Click any model name to read the full
operator-grade page.
"""

import gradio as gr
import pandas as pd
import requests

SITE_URL = "https://www.runlocalai.co"
MODELS_URL = f"{SITE_URL}/api/v1/models"
QB_URL = f"{SITE_URL}/api/v1/quality-benchmarks"
API_LIMIT = 200  # /api/v1/models caps at 200; rows come pre-sorted by popularity

# ─────────────────────────────────────────────────────────────────────────
# Leaderboard (tab 1) — GET /api/v1/quality-benchmarks
# ─────────────────────────────────────────────────────────────────────────

STATUS_DISPLAY = {
    "first-party": "⭐ First-party",
    "community": "👥 Community",
    "verified": "✅ Verified",
    "pending": "⏳ Pending",
}

LB_COLS = ["Rank", "Benchmark", "Model", "Params (B)", "Quant", "Runtime", "Score", "Status", "Proof", "Tested"]
LB_DATATYPES = ["str", "str", "markdown", "number", "str", "str", "number", "str", "markdown", "str"]
ALL_BENCHMARKS = "All benchmarks"


def fetch_leaderboard():
    """Fetch quality-benchmark runs. Returns (runs_df, benchmarks_meta)."""
    try:
        r = requests.get(QB_URL, timeout=30)
        r.raise_for_status()
        payload = r.json()
    except Exception as exc:  # noqa: BLE001
        return pd.DataFrame({"Error": [f"Could not fetch leaderboard: {exc}"]}), {}

    benchmarks = {b.get("slug"): b for b in payload.get("benchmarks", []) if isinstance(b, dict)}
    runs = payload.get("runs", []) or []
    if not runs:
        return pd.DataFrame({"Status": ["No benchmark runs yet"]}), benchmarks

    records = []
    for run in runs:
        if not isinstance(run, dict):
            continue
        bslug = run.get("benchmark") or ""
        bdef = benchmarks.get(bslug, {})
        bname = bdef.get("name") or bslug or "—"

        mslug = run.get("model_slug") or ""
        mname = run.get("model_name") or mslug or "—"
        model_link = f"[{mname}]({SITE_URL}/models/{mslug})" if mslug else mname

        score = run.get("score")
        score_val = round(float(score), 1) if isinstance(score, (int, float)) else None

        log_url = run.get("test_run_log_url") or ""
        proof = f"[run log]({log_url})" if log_url else "—"

        status = run.get("submission_status") or ""
        tested = (run.get("tested_at") or "")[:10]

        records.append({
            "Benchmark": bname,
            "Model": model_link,
            "Params (B)": run.get("model_params_b"),
            "Quant": run.get("quant") or "—",
            "Runtime": run.get("runtime") or "—",
            "Score": score_val,
            "Status": STATUS_DISPLAY.get(status, status or "—"),
            "Proof": proof,
            "Tested": tested or "—",
            "_benchmark": bname,
            "_score_raw": float(score) if isinstance(score, (int, float)) else -1.0,
        })

    return pd.DataFrame(records), benchmarks


def leaderboard_view(df: pd.DataFrame, benchmark_label: str) -> pd.DataFrame:
    """Filter to a benchmark (or all), rank by score within each, add medals."""
    if "_score_raw" not in df.columns:  # error / empty passthrough
        return df
    out = df.copy()
    if benchmark_label and benchmark_label != ALL_BENCHMARKS:
        out = out[out["_benchmark"] == benchmark_label]
    if out.empty:
        return pd.DataFrame({"Status": ["No runs for this benchmark yet"]})

    out = out.sort_values(["_benchmark", "_score_raw"], ascending=[True, False])
    ranks = out.groupby("_benchmark")["_score_raw"].rank(ascending=False, method="min").astype(int)
    medal = {1: "🥇", 2: "🥈", 3: "🥉"}
    out["Rank"] = [medal.get(int(rk), str(int(rk))) for rk in ranks]
    return out[LB_COLS].reset_index(drop=True)


def benchmark_blurb(benchmarks: dict) -> str:
    """Render the 'what these benchmarks measure' note from API metadata."""
    if not benchmarks:
        return ""
    lines = ["### What these scores mean\n"]
    for b in benchmarks.values():
        name = b.get("name", "")
        metric = b.get("metric", {}) or {}
        unit = f"{metric.get('label', '')} {metric.get('unit', '')}".strip()
        src = b.get("source", {}) or {}
        authors = src.get("authors", "")
        url = src.get("url", "")
        cats = ", ".join(b.get("categories", []) or [])
        src_link = f"[dataset]({url})" if url else ""
        lines.append(f"- **{name}** — {unit} · _{cats}_ · {authors} {src_link}".rstrip())
    lines.append(
        "\nEvery run is **measured first-party** on real consumer hardware and carries a public "
        "run log + a one-line reproduction command. "
        f"Methodology: [runlocalai.co/benchmarks/methodology]({SITE_URL}/benchmarks/methodology)."
    )
    return "\n".join(lines)


# ─────────────────────────────────────────────────────────────────────────
# Catalog (tab 2) — GET /api/v1/models
# ─────────────────────────────────────────────────────────────────────────

MODALITY_DISPLAY = {
    "text": "💬 Text",
    "vision": "👁️ Vision",
    "audio": "🎙️ Audio",
    "video": "🎥 Video",
    "embedding": "📐 Embedding",
    "rerank": "🔍 Rerank",
    "image-gen": "🎨 Image-gen",
}

CAT_HIDDEN = ["_modality_raw", "_params_raw", "_commercial_raw", "_family"]
CAT_DATATYPES = ["markdown", "str", "str", "str", "str", "str", "str", "markdown", "number", "number"]


def fetch_catalog() -> pd.DataFrame:
    """Fetch the latest model catalog. Falls back gracefully on error."""
    try:
        r = requests.get(MODELS_URL, params={"limit": API_LIMIT}, timeout=30)
        if r.status_code == 401:
            return pd.DataFrame(
                {"Error": ["Catalog API requires a key right now. "
                           "Browse the full catalog at runlocalai.co/models"]}
            )
        r.raise_for_status()
        payload = r.json()
        if isinstance(payload, dict):
            rows = payload.get("data") or payload.get("models") or []
        elif isinstance(payload, list):
            rows = payload
        else:
            rows = []
    except Exception as exc:  # noqa: BLE001
        return pd.DataFrame({"Error": [f"Could not fetch catalog: {exc}"]})

    if not rows:
        return pd.DataFrame({"Status": ["Catalog is empty"]})

    records = []
    for m in rows:
        if not isinstance(m, dict):
            continue
        modalities = m.get("modalities") or ["text"]
        modality = modalities[0] if isinstance(modalities, list) and modalities else "text"

        params_b = m.get("parameter_count_b") or 0
        if params_b and params_b < 1:
            params_label = f"{int(round(params_b * 1000))}M"
        elif params_b:
            params_label = f"{params_b}B"
        else:
            params_label = "—"

        commercial = "✅ Yes" if m.get("license_commercial_ok") else "⚠️ Restricted"
        license_short = (m.get("license") or "—")[:24]

        ctx = m.get("context_length") or 0
        ctx_label = f"{int(ctx / 1024)}K" if ctx >= 1024 else (f"{ctx}" if ctx > 0 else "—")

        slug = m.get("slug", "") or ""
        name = m.get("name") or slug or "—"
        name_link = f"[{name}]({SITE_URL}/models/{slug})" if slug else name

        hf_repo = m.get("hf_repo") or ""
        hf_link = f"[hf.co/{hf_repo}](https://huggingface.co/{hf_repo})" if hf_repo else "—"

        rating = m.get("our_rating_score")
        rating_val = round(float(rating), 1) if isinstance(rating, (int, float)) else 0.0

        records.append({
            "Model": name_link,
            "Modality": MODALITY_DISPLAY.get(modality, modality),
            "Params": params_label,
            "Context": ctx_label,
            "License": license_short,
            "Commercial": commercial,
            "Vendor": m.get("vendor") or "—",
            "HuggingFace": hf_link,
            "Rating": rating_val,
            "Popularity": m.get("popularity_score") or 0,
            "_modality_raw": modality,
            "_params_raw": float(params_b) if params_b else 0.0,
            "_commercial_raw": bool(m.get("license_commercial_ok")),
            "_family": m.get("family") or "other",
        })

    if not records:
        return pd.DataFrame({"Status": ["Catalog is empty"]})

    df = pd.DataFrame(records)
    return df.sort_values("Popularity", ascending=False).reset_index(drop=True)


def apply_filters(df, modality, commercial_only, max_params, family, search):
    if not all(c in df.columns for c in CAT_HIDDEN):  # error / status passthrough
        return df
    out = df.copy()
    if modality and modality != "All":
        out = out[out["_modality_raw"] == modality]
    if commercial_only:
        out = out[out["_commercial_raw"]]
    if max_params and max_params < 200:  # 200 = no cap
        out = out[out["_params_raw"] <= max_params]
    if family and family != "All":
        out = out[out["_family"] == family]
    if search:
        s = search.lower().strip()
        mask = (
            out["Model"].str.lower().str.contains(s, na=False)
            | out["Vendor"].str.lower().str.contains(s, na=False)
            | out["HuggingFace"].str.lower().str.contains(s, na=False)
        )
        out = out[mask]
    return out.drop(columns=CAT_HIDDEN)


# ─────────────────────────────────────────────────────────────────────────
# Initial data load
# ─────────────────────────────────────────────────────────────────────────

LB_DATA, BENCHMARKS = fetch_leaderboard()
CATALOG = fetch_catalog()

benchmark_options = [ALL_BENCHMARKS] + sorted(
    {b.get("name") for b in BENCHMARKS.values() if b.get("name")}
)
modality_options = ["All"] + sorted({m for m in CATALOG.get("_modality_raw", []) if isinstance(m, str)})
family_options = ["All"] + sorted({f for f in CATALOG.get("_family", []) if isinstance(f, str)})

# ─────────────────────────────────────────────────────────────────────────
# UI
# ─────────────────────────────────────────────────────────────────────────

with gr.Blocks(
    title="RunLocalAI — local AI leaderboard & catalog",
    theme=gr.themes.Soft(primary_hue="amber", neutral_hue="slate"),
) as demo:
    gr.Markdown(
        f"""
        # 🛠️ RunLocalAI — local AI leaderboard & catalog

        Reproducible benchmark scores and the full open-weight model catalog for running AI on
        **your own hardware**. Every benchmark is measured first-party with a public run log and a
        one-line reproduction command — no vibes, no leaderboard laundering.

        Source of truth: **[runlocalai.co]({SITE_URL})** · Data license: **CC-BY-4.0** ·
        Click any model name for the full operator-grade page.
        """
    )

    with gr.Tabs():
        # ── Tab 1: Leaderboard ────────────────────────────────────────────
        with gr.Tab("🏆 Benchmark leaderboard"):
            gr.Markdown(
                "Ranked, reproducible quality scores on real consumer GPUs. "
                "Pick a benchmark to see the head-to-head ranking."
            )
            benchmark_dd = gr.Dropdown(
                benchmark_options, value=ALL_BENCHMARKS, label="Benchmark", interactive=True
            )
            lb_table = gr.Dataframe(
                value=leaderboard_view(LB_DATA, ALL_BENCHMARKS),
                interactive=False,
                wrap=True,
                datatype=LB_DATATYPES,
            )
            lb_refresh = gr.Button("🔁 Refresh leaderboard", variant="secondary")
            gr.Markdown(benchmark_blurb(BENCHMARKS))

            def _lb_filter(label):
                return leaderboard_view(LB_DATA, label)

            def _lb_refresh(label):
                global LB_DATA, BENCHMARKS
                LB_DATA, BENCHMARKS = fetch_leaderboard()
                return leaderboard_view(LB_DATA, label)

            benchmark_dd.change(_lb_filter, inputs=benchmark_dd, outputs=lb_table)
            lb_refresh.click(_lb_refresh, inputs=benchmark_dd, outputs=lb_table)

        # ── Tab 2: Catalog ────────────────────────────────────────────────
        with gr.Tab("🛠️ Model catalog"):
            gr.Markdown(
                "Every open-weight model worth running locally — LLMs, embeddings, rerankers, "
                "ASR, TTS, diffusion, vision encoders — with license tone and VRAM math."
            )
            with gr.Row():
                modality_dd = gr.Dropdown(modality_options, value="All", label="Modality", interactive=True)
                family_dd = gr.Dropdown(family_options, value="All", label="Family", interactive=True)
                max_params_slider = gr.Slider(
                    minimum=0.1, maximum=200, value=200, step=0.5,
                    label="Max params (B). 200 = no cap.",
                )
            with gr.Row():
                search_box = gr.Textbox(
                    label="Search (model / vendor / hf repo)",
                    placeholder="qwen, kokoro, gemma, deepseek …",
                )
                commercial_only = gr.Checkbox(label="Commercial-license only", value=False)

            cat_table = gr.Dataframe(
                value=apply_filters(CATALOG, "All", False, 200, "All", ""),
                interactive=False,
                wrap=True,
                datatype=CAT_DATATYPES,
            )
            cat_refresh = gr.Button("🔁 Refresh catalog", variant="secondary")

            cat_inputs = [modality_dd, commercial_only, max_params_slider, family_dd, search_box]

            def _cat_filter(mod, com, mp, fam, search):
                return apply_filters(CATALOG, mod, com, mp, fam, search)

            def _cat_refresh(mod, com, mp, fam, search):
                global CATALOG
                CATALOG = fetch_catalog()
                return apply_filters(CATALOG, mod, com, mp, fam, search)

            for ctrl in cat_inputs:
                ctrl.change(_cat_filter, inputs=cat_inputs, outputs=cat_table)
            cat_refresh.click(_cat_refresh, inputs=cat_inputs, outputs=cat_table)

    gr.Markdown(
        f"""
        ---
        **Catalog hubs:**
        [Small LMs]({SITE_URL}/small-language-models) ·
        [Embeddings]({SITE_URL}/embeddings) ·
        [Audio]({SITE_URL}/audio) ·
        [Image]({SITE_URL}/image-models) ·
        [Coding]({SITE_URL}/coding-models) ·
        [Turkish]({SITE_URL}/turkish-models) ·
        [Benchmarks]({SITE_URL}/benchmarks)

        **Machine-readable:**
        [models]({MODELS_URL}) ·
        [quality-benchmarks]({QB_URL}) ·
        [OpenAPI]({SITE_URL}/api/v2/openapi)

        Data licensed **CC-BY-4.0** — attribute to runlocalai.co with a link.
        """
    )

if __name__ == "__main__":
    demo.launch()