Update update_data.py

update_data.py

@@ -1,4 +1,4 @@
-# /// script
+# ///  script
 # requires-python = ">=3.11"
 # dependencies = [
 #     "httpx",
@@ -6,17 +6,19 @@
 # ]
 # ///
 """
-Scheduled job: regenerate data.json and upload to the benchmark-race Space.
-
-Run locally:
-    uv run update_data.py
-
-Schedule on HF Jobs (twice daily):
-    hf jobs scheduled uv run "0 8,20 * * *" \
-        --secrets HF_TOKEN \
-        https://huggingface.co/spaces/davanstrien/benchmark-race/resolve/main/update_data.py
+Regenerate data.json and upload to the elevow/benchmarks Space.
+Source template: duplicated from davanstrien/benchmark-race
+https://huggingface.co/spaces/elevow/benchmarks
+Run locally (from repo root or this folder):
+    export HF_TOKEN=hf_...
+    uv run scripts/elevow-benchmarks/update_data.py
+Or copy this file to your Space repo root on Hugging Face and run there.
+Schedule on HF Jobs (example — point to YOUR raw file):
+    hf jobs scheduled uv run "0 8,20 * * *" \\
+        --secrets HF_TOKEN \\
+        https://huggingface.co/spaces/elevow/benchmarks/resolve/main/update_data.py
 """
-
+from __future__ import annotations
 import json
 import os
 import re
@@ -24,12 +26,24 @@ import tempfile
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
 from pathlib import Path
-
+from typing import Any
 import httpx
 from huggingface_hub import HfApi
-
-SPACE_REPO = "davanstrien/benchmark-race"
-
+# Upload target: your fork (was davanstrien/benchmark-race in upstream).
+SPACE_REPO = os.environ.get("BENCHMARK_SPACE_REPO", "elevow/benchmarks")
+ALIGNED_LOGO_URL = (
+    "https://www.google.com/s2/favicons?sz=128&domain_url="
+    "https%3A%2F%2Ftryaligned.ai"
+)
+ALIGNED_LOGOS_KEY = "AlignedAI"
+ALIGNED_COLOR = "#059669"
+# Full HF model_id strings from leaderboard APIs — add any row that should show Aligned branding.
+MODEL_IDS_USE_ALIGNED_LOGO: frozenset[str] = frozenset(
+    {
+        # Populate from live leaderboard responses, e.g.:
+        # "Qwen/Qwen2.5-Coder-32B-Instruct",
+    }
+)
 BENCHMARK_CONFIGS = [
     {"dataset": "SWE-bench/SWE-bench_Verified", "key": "sweVerified", "name": "SWE-bench Verified", "gated": False},
     {"dataset": "ScaleAI/SWE-bench_Pro", "key": "swePro", "name": "SWE-bench Pro", "gated": False},
@@ -42,15 +56,25 @@ BENCHMARK_CONFIGS = [
     {"dataset": "harborframework/terminal-bench-2.0", "key": "terminalBench", "name": "Terminal-Bench 2.0", "gated": False},
     {"dataset": "FutureMa/EvasionBench", "key": "evasionBench", "name": "EvasionBench", "gated": False},
 ]
-
 PALETTE = [
     "#6366f1", "#0d9488", "#d97706", "#e11d48", "#7c3aed",
     "#16a34a", "#2563eb", "#ea580c", "#8b5cf6", "#0891b2",
     "#c026d3", "#65a30d", "#dc2626", "#0284c7", "#a21caf",
     "#059669", "#9333ea", "#ca8a04", "#be185d", "#0369a1",
 ]
-
-
+def inject_aligned_race_branding(
+    benchmarks: dict[str, Any],
+    logos: dict[str, str],
+    color_map: dict[str, str],
+) -> None:
+    """Add Aligned logo URL, optional per-model race_logo_key, and bar color."""
+    logos[ALIGNED_LOGOS_KEY] = ALIGNED_LOGO_URL
+    color_map[ALIGNED_LOGOS_KEY] = ALIGNED_COLOR
+    for _key, bm in benchmarks.items():
+        for m in bm.get("models") or []:
+            mid = m.get("model_id") or ""
+            if mid in MODEL_IDS_USE_ALIGNED_LOGO:
+                m["race_logo_key"] = ALIGNED_LOGOS_KEY
 def fetch_leaderboard(config: dict, hf_token: str | None) -> list[dict]:
     url = f"https://huggingface.co/api/datasets/{config['dataset']}/leaderboard"
     headers = {}
@@ -59,7 +83,6 @@ def fetch_leaderboard(config: dict, hf_token: str | None) -> list[dict]:
     elif config["gated"]:
         print(f"  {config['name']}: skipped (gated, no token)")
         return []
-
     print(f"  {config['name']}: fetching scores...")
     try:
         resp = httpx.get(url, headers=headers, timeout=30)
@@ -72,8 +95,7 @@ def fetch_leaderboard(config: dict, hf_token: str | None) -> list[dict]:
     except Exception as e:
         print(f"    error: {e}")
         return []
-
-    seen = {}
+    seen: dict[str, float] = {}
     for entry in data:
         model_id = entry.get("modelId")
         score = entry.get("value")
@@ -81,16 +103,12 @@ def fetch_leaderboard(config: dict, hf_token: str | None) -> list[dict]:
             score = float(score)
             if model_id not in seen or score > seen[model_id]:
                 seen[model_id] = score
-
     print(f"    {len(seen)} models")
     return [{"model_id": mid, "score": s} for mid, s in seen.items()]
-
-
 def fetch_model_dates(model_ids: list[str], hf_token: str | None) -> dict[str, dict]:
     api = HfApi()
-    results = {}
-
-    def _get_info(mid):
+    results: dict[str, dict] = {}
+    def _get_info(mid: str):
         try:
             info = api.model_info(mid, token=hf_token)
             params_b = None
@@ -103,17 +121,13 @@ def fetch_model_dates(model_ids: list[str], hf_token: str | None) -> dict[str, d
             return mid, info.created_at.strftime("%Y-%m-%d"), params_b
         except Exception:
             return mid, None, None
-
     with ThreadPoolExecutor(max_workers=8) as pool:
         futures = {pool.submit(_get_info, mid): mid for mid in model_ids}
         for f in as_completed(futures):
             mid, date, params = f.result()
             if date:
                 results[mid] = {"date": date, "parameters_b": params}
-
     return results
-
-
 def fetch_logo(provider: str) -> str | None:
     try:
         resp = httpx.get(
@@ -125,10 +139,8 @@ def fetch_logo(provider: str) -> str | None:
     except Exception:
         pass
     return None
-
-
 def fetch_all_logos(providers: set[str]) -> dict[str, str]:
-    logos = {}
+    logos: dict[str, str] = {}
     with ThreadPoolExecutor(max_workers=8) as pool:
         futures = {pool.submit(fetch_logo, p): p for p in providers}
         for f in as_completed(futures):
@@ -137,31 +149,24 @@ def fetch_all_logos(providers: set[str]) -> dict[str, str]:
             if url:
                 logos[p] = url
     return logos
-
-
-def main():
+def main() -> None:
     hf_token = os.environ.get("HF_TOKEN")
-    print("Generating data.json for bar chart race\n")
-
-    all_scores: dict[str, list[dict]] = {}
+    print(f"Generating data.json → upload to {SPACE_REPO}\n")
+    all_scores: dict[str, dict] = {}
     all_model_ids: set[str] = set()
-
     for config in BENCHMARK_CONFIGS:
         rows = fetch_leaderboard(config, hf_token)
         if rows:
             all_scores[config["key"]] = {"name": config["name"], "rows": rows}
             all_model_ids.update(r["model_id"] for r in rows)
-
     print(f"\n{len(all_model_ids)} unique models across {len(all_scores)} benchmarks")
     print("Fetching model dates...")
     model_dates = fetch_model_dates(list(all_model_ids), hf_token)
     print(f"  got dates for {len(model_dates)}/{len(all_model_ids)} models")
-
     all_providers: set[str] = set()
-    benchmarks = {}
-
+    benchmarks: dict[str, Any] = {}
     for key, info in all_scores.items():
-        models = []
+        models: list[dict] = []
         for row in info["rows"]:
             mid = row["model_id"]
             if mid not in model_dates:
@@ -178,34 +183,29 @@ def main():
             })
         if models:
             benchmarks[key] = {"name": info["name"], "models": models}
-
     print(f"\nFetching logos for {len(all_providers)} providers...")
     logos = fetch_all_logos(all_providers)
     print(f"  got {len(logos)} logos")
-
-    color_map = {}
+    color_map: dict[str, str] = {}
     for i, provider in enumerate(sorted(all_providers)):
         color_map[provider] = PALETTE[i % len(PALETTE)]
-
+    inject_aligned_race_branding(benchmarks, logos, color_map)
+    print(f"  injected {ALIGNED_LOGOS_KEY} logo + color; race_logo_key on {len(MODEL_IDS_USE_ALIGNED_LOGO)} id(s) configured")
     output = {
         "benchmarks": benchmarks,
         "logos": logos,
         "colors": color_map,
         "generated_at": datetime.now(timezone.utc).isoformat(),
     }
-
     data_json = json.dumps(output, indent=2)
     print(f"\nGenerated {len(data_json) / 1024:.1f} KB")
     for key, bm in benchmarks.items():
         print(f"  {bm['name']}: {len(bm['models'])} models")
-
-    # Upload to Space
     print(f"\nUploading data.json to {SPACE_REPO}...")
     api = HfApi()
-    with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False, encoding="utf-8") as f:
         f.write(data_json)
         tmp_path = f.name
-
     try:
         api.upload_file(
             path_or_fileobj=tmp_path,
@@ -217,7 +217,5 @@ def main():
         print("Done!")
     finally:
         Path(tmp_path).unlink(missing_ok=True)
-
-
 if __name__ == "__main__":
-    main()
+    main()