"""Gradio front-end for the PRIMO public benchmark. A left rail navigates six pages: Boards (a grid of every board), Board (one leaderboard at a time), Tasks ("what is actually being tested?"), Submit ("how do I enter?"), Contribute ("what is missing, and how do I add it?"), Method ("can I trust this?"). Every page is addressable -- ``?board=`` opens a board and ``?tab=contribute`` opens a tab -- which is what the rail links and the open cards use. The tab strip is hidden in CSS; the rail is the navigation. Upload one embedding file spanning every dataset (rows keyed by ``dataset_id`` + ``sample_id``); a fixed linear probe scores each task (a dataset may carry several hidden targets). Results roll up into boards -- the whole modality, one therapeutic area, one task family -- and each board ranks the models that covered all of its tasks, one column per category in its native metric (AUROC or Pearson), plus a ``Mean`` of those columns that orders the rows and is labelled as the cross-metric average it is. Disclosure policy -- what the public pages may show: per task disease, tissue, area, what is predicted, class names, n, metric aggregated the public archives the cohorts sit in, and their licences never study accessions, dataset_id -> cohort, hub keys, per-sample labels Naming the accession behind ``d002`` would put every label one GEO download away, so ``sources`` is collapsed to its archive (NCBI GEO / EMBL-EBI ArrayExpress) and ``citation`` is never rendered at all. The submitter's Hugging Face username is a name-ownership lock, kept private and never rendered. Failures are surfaced by who owns them: a bad file tells the submitter exactly what to fix; an evaluator-side failure says "our side, please retry" and logs the traceback rather than blaming the submission. A page load is ONE fetch: ``_init`` pulls the registry, the boards and the persisted results once and threads that state into every page, rather than each page fetching for itself. """ import os import traceback from datetime import datetime, timezone from pathlib import Path import gradio as gr import pandas as pd from boards import Board, build_boards, by_slug from evaluator import ( EvaluatorError, SubmissionError, _norm_id, fetch_manifest, fetch_tasks_registry, manifest_ids, score_all, scoreable_tasks, ) from leaderboard import RESERVED_COLUMNS, source_repositories from render import rail_html, render_board, render_boards, render_tasks from results import ( BASELINE_TAG, IS_BASELINE, OWNER, RESULT_COLUMNS, append_results, append_submission, owner_of, read_results, ) HERE = Path(__file__).parent PAGES_DIR = HERE / "pages" CSS_PATH = str(HERE / "primo.css") ALLOWED_PATHS = [str(HERE / "assets"), str(HERE / "fonts")] TOKEN = os.environ.get("HF_TOKEN") TAB_IDS = ("boards", "board", "tasks", "submit", "contribute", "method") FOOT_TABS = frozenset({"tasks", "submit", "contribute", "method"}) THEME = gr.themes.Base( font=["Funnel Sans", "sans-serif"], font_mono=["DM Mono", "monospace"] ) SUBMIT_HEAD = ( '

Submit a model

' "

One embedding file. Partial coverage is fine. You are ranked on every " "board you cover in full.

" ) TABLE_SORT_JS = """ () => { if (window.pmTableSortBound) return; window.pmTableSortBound = true; const value = (cell) => { const text = cell.textContent.trim(); if (!text || text.toLowerCase() === "n/a") return { missing: true, text }; const number = Number(text.replaceAll(",", "")); return Number.isFinite(number) ? { missing: false, number, text } : { missing: false, text }; }; const sort = (header) => { const table = header.closest("table.pm-table"); const body = table?.tBodies[0]; if (!body) return; const index = Number(header.dataset.sortIndex); const direction = header.dataset.sortDirection === "asc" ? -1 : 1; const rows = Array.from(body.rows); rows.sort((left, right) => { const a = value(left.cells[index]); const b = value(right.cells[index]); if (a.missing || b.missing) return a.missing === b.missing ? 0 : a.missing ? 1 : -1; if (a.number !== undefined && b.number !== undefined) return direction * (a.number - b.number); return direction * a.text.localeCompare(b.text, undefined, { numeric: true }); }); rows.forEach((row) => body.append(row)); table.querySelectorAll("th.pm-sort").forEach((cell) => { cell.dataset.sortDirection = ""; cell.setAttribute("aria-sort", "none"); }); header.dataset.sortDirection = direction === 1 ? "asc" : "desc"; header.setAttribute("aria-sort", direction === 1 ? "ascending" : "descending"); }; document.addEventListener("click", (event) => { const header = event.target.closest("th.pm-sort"); if (header) sort(header); }); document.addEventListener("keydown", (event) => { if (event.key !== "Enter" && event.key !== " ") return; const header = event.target.closest("th.pm-sort"); if (!header) return; event.preventDefault(); sort(header); }); } """ def _page_text(name: str) -> str: """One ``pages/.md`` per prose tab, named after the tab it fills. Editing the site's words never means touching Python. The data pages (Boards, Board, Tasks) are generated markup instead. """ return (PAGES_DIR / f"{name}.md").read_text() def _registry_by_id() -> dict[str, dict]: """Scoreable tasks keyed by task_id (dataset present in the public manifest). ``PRIMO_FIXTURE=1`` swaps the token-gated fetch for the test registry, so the app runs offline for local smoke-testing. """ if os.environ.get("PRIMO_FIXTURE"): from conftest import REGISTRY return {task_id: dict(task) for task_id, task in REGISTRY.items()} datasets = manifest_ids(fetch_manifest(TOKEN)) registry = scoreable_tasks(fetch_tasks_registry(TOKEN), datasets) return {_norm_id(task["task_id"]): task for task in registry} PageState = tuple[dict[str, dict], list[Board], pd.DataFrame] def _page_state() -> PageState: """Registry, boards and persisted results -- one fetch per render. A failed fetch yields empty structures so the page renders a "come back later" state instead of a stack trace. Every page is built from one of these, threaded through rather than refetched, so a page load is one round trip. """ try: by_id = _registry_by_id() return by_id, build_boards(by_id), read_results(TOKEN) except Exception: # noqa: BLE001 traceback.print_exc() return {}, [], pd.DataFrame(columns=RESULT_COLUMNS) def _about_text(by_id: dict[str, dict]) -> str: """Methodology + the archives the cohorts live in, never their accessions. The placeholder is substituted, not ``.format``-ed: a page of prose is free to contain a brace, and a stray one must not blow up the tab. An unreachable registry leaves ``by_id`` empty and the sentence falls back to a generic one. """ repositories = source_repositories(list(by_id.values())) return _page_text("about").replace( "{repositories}", " and ".join(repositories) or "public archives" ) def _summary(result: dict, model_name: str) -> str: lines = [ f"**{model_name}**: covered {result['n_datasets_scored']}/" f"{result['n_datasets_total']} datasets", "", ] for category, stats in sorted(result["categories"].items()): lines.append(f"- **{category}** ({stats['metric']}) = {stats['mean']:.3f}") if result["full_coverage"]: lines.append("\n✅ **full coverage.** You are ranked on every board.") else: lines.append( "\n⚠️ **partial coverage.** You are ranked on the boards whose tasks you " "covered in full, and your scores always appear in each board's " "**per-task** table." ) if result["missing"]: lines.append(f"- missing from file: {result['missing']}") if result["incomplete"]: lines.append(f"- could not score: {result['incomplete']}") return "\n".join(lines) def _claimed_by(model: str) -> str: """Who already owns this model name, or ``""``. A results fetch that fails leaves the name free: a Hugging Face hiccup must not block a submission, and the worst case is the collision we had before. """ try: return owner_of(read_results(TOKEN), model) except Exception: # noqa: BLE001 traceback.print_exc() return "" def _rendered_board(slug: str | None) -> str: """The Board page for ``slug`` -- re-rendered after a submission saves.""" by_id, boards, df = _page_state() return render_board(by_slug(boards, slug), df, by_id) def evaluate( submission_path: str, model_name: str, email: str, paper_link: str, hf_model_link: str, notes: str, slug: str | None, profile: gr.OAuthProfile | None, ): def _refuse(message: str): return message, _rendered_board(slug) if profile is None: return _refuse("Please sign in with Hugging Face to submit.") if not submission_path: return _refuse("Please upload a submission file.") if not model_name or not model_name.strip(): return _refuse("Please enter a model name.") if not email or not email.strip(): return _refuse("Please enter a contact email.") model = model_name.strip() if BASELINE_TAG in model.lower(): return _refuse( f"`{BASELINE_TAG}` is reserved for our reference submissions. Please " "pick another model name." ) if model in RESERVED_COLUMNS: return _refuse( f"`{model}` is a column of the per-task table. Please pick another " "model name." ) claimed = _claimed_by(model) if claimed and claimed != profile.username: return _refuse( f"The model name `{model}` is already taken by @{claimed}, and the " "board keeps each name's latest submission. Please pick another name." ) try: result = score_all(submission_path, TOKEN) except SubmissionError as error: return _refuse(f"❌ {error}") except EvaluatorError as error: traceback.print_exc() return _refuse( "⚠️ We couldn't evaluate your submission. This is on our side, not your " f"file. Please try again in a moment.\n\n`{error}`" ) except Exception as error: # noqa: BLE001 traceback.print_exc() return _refuse(f"⚠️ Unexpected evaluation error (our side): {error}") summary = _summary(result, model) submitted_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") rows = [ { "model_name": model, "task_id": task.task_id, "score": round(float(task.score), 4), "submitted_at": submitted_at, IS_BASELINE: False, OWNER: profile.username, } for task in result["per_task"] ] meta = { "model_name": model, "submitted_at": submitted_at, OWNER: profile.username, "email": email.strip(), "paper_link": (paper_link or "").strip(), "hf_model_link": (hf_model_link or "").strip(), "notes": (notes or "").strip(), } try: append_results(rows, TOKEN) append_submission(meta, TOKEN) except Exception as error: # noqa: BLE001 traceback.print_exc() summary += f"\n\n⚠️ scored, but the leaderboard was not saved: {error}" return summary, _rendered_board(slug) def _landing_tab(params: dict, has_board: bool) -> str: """Which page a visitor lands on: ``?tab=`` wins, then ``?board=``, else Boards. An unknown ``?tab=`` falls through to Boards rather than selecting nothing, which would render the Space with every panel collapsed. """ tab = params.get("tab") if tab in TAB_IDS: return tab return "board" if has_board else "boards" def _init(request: gr.Request): """Render every page from one fetch, landing where the query params ask.""" by_id, boards, df = _page_state() params = dict(request.query_params) if request else {} board = by_slug(boards, params.get("board")) selected = _landing_tab(params, bool(params.get("board") and board)) active_slug = board.slug if selected == "board" and board else None active_tab = selected if selected in FOOT_TABS else None return ( gr.Tabs(selected=selected), rail_html(boards, active_slug, active_tab), board.slug if board else None, render_boards(boards, df, by_id), render_tasks(by_id), _about_text(by_id), render_board(board, df, by_id), ) def build_demo() -> gr.Blocks: with gr.Blocks( title="PRIMO Benchmark", theme=THEME, css_paths=[CSS_PATH], js=TABLE_SORT_JS, fill_width=True, ) as demo: active_board = gr.State(None) with gr.Row(elem_id="pm-shell"): with gr.Column(elem_id="pm-rail"): rail = gr.HTML() with gr.Column(elem_id="pm-main"): with gr.Tabs(elem_id="pm-pages") as pages: with gr.Tab("Boards", id="boards"): boards_html = gr.HTML() with gr.Tab("Board", id="board"): board_html = gr.HTML() with gr.Tab("Tasks", id="tasks"): tasks_html = gr.HTML() with gr.Tab("Submit", id="submit"): gr.HTML(SUBMIT_HEAD) with gr.Column(elem_classes=["pm-body"]): with gr.Row(): with gr.Column(scale=3): gr.Markdown(_page_text("submit")) with gr.Column(scale=2, elem_id="pm-form"): gr.LoginButton() model_tb = gr.Textbox( label="Model name", placeholder="e.g. eva-rna-v1", info="Shown on the leaderboard.", ) email_tb = gr.Textbox( label="Email address", placeholder="you@lab.org", info="Contact for this submission, kept private.", ) notes_tb = gr.Textbox( label="Training data / notes (optional)", placeholder="e.g. pretrained on atlas X", info="About the model or its training data.", ) paper_tb = gr.Textbox( label="Paper link (optional)", placeholder="https://arxiv.org/abs/...", ) hf_tb = gr.Textbox( label="Hugging Face model link (optional)", placeholder="https://huggingface.co/...", ) file_in = gr.File( label="Submission (.csv / .tsv / .parquet / .npz)", type="filepath", ) run_btn = gr.Button( "Evaluate", elem_classes=[ "pm-btn", "pm-btn--primary", "pm-btn--block", ], ) result_md = gr.Markdown() with gr.Tab("Contribute", id="contribute"): with gr.Column(elem_classes=["pm-body", "pm-prose"]): gr.Markdown(_page_text("contribute")) with gr.Tab("Method", id="method"): with gr.Column(elem_classes=["pm-body", "pm-prose"]): about_md = gr.Markdown() run_btn.click( evaluate, [file_in, model_tb, email_tb, paper_tb, hf_tb, notes_tb, active_board], [result_md, board_html], ) demo.load( _init, None, [pages, rail, active_board, boards_html, tasks_html, about_md, board_html], ) return demo if __name__ == "__main__": build_demo().launch( server_name="0.0.0.0", server_port=7860, ssr_mode=False, allowed_paths=ALLOWED_PATHS, )