src/populate.py

import pandas as pd

import gradio as gr

from src.about import Tasks
from src.display.columns import localize_dataset_leaderboard_df, localize_main_leaderboard_df
from src.display.formatting import has_no_nan_values, make_clickable_model
from src.display.dataset_readmes import get_dataset_readme_markdown
from src.display.i18n import t
from src.display.utils import AutoEvalColumn, get_dataset_metric_cols
from src.leaderboard.read_evals import get_dataset_metrics_for_row, get_stored_eval_rows
from src.leaderboard.sync import sync_results_from_hub


def get_leaderboard_df(cols: list, benchmark_cols: list) -> pd.DataFrame:
    rows = get_stored_eval_rows()
    if not rows:
        return localize_main_leaderboard_df(pd.DataFrame(columns=cols))

    display_rows = [{k: v for k, v in row.items() if not k.startswith("_")} for row in rows]
    df = pd.DataFrame.from_records(display_rows)
    df = df.sort_values(by=[AutoEvalColumn.average.name], ascending=False)
    df = df[cols].round(decimals=2)
    df = df[has_no_nan_values(df, benchmark_cols)]
    return localize_main_leaderboard_df(df)


def get_dataset_leaderboard_df(task_name: str) -> pd.DataFrame:
    task = Tasks[task_name]
    task_val = task.value
    metric_cols = get_dataset_metric_cols(task)
    columns = ["model", "team"] + metric_cols
    sort_col = metric_cols[0] if metric_cols else None

    rows = get_stored_eval_rows()
    if not rows:
        return localize_dataset_leaderboard_df(pd.DataFrame(columns=columns))

    dataset_rows = []
    for row in rows:
        dataset_score = row.get("_results", {}).get(task_val.benchmark)
        if dataset_score is None:
            continue

        metrics = get_dataset_metrics_for_row(row, task)
        if not any(v is not None for v in metrics.values()):
            continue

        dataset_row = {
            "model": make_clickable_model(row["_model"]),
            "team": row[AutoEvalColumn.team.name],
        }
        for col_name, value in metrics.items():
            dataset_row[col_name] = round(value, 2) if value is not None else None
        dataset_rows.append(dataset_row)

    if not dataset_rows:
        return localize_dataset_leaderboard_df(pd.DataFrame(columns=columns))

    df = pd.DataFrame.from_records(dataset_rows, columns=columns)
    if sort_col:
        df = df.sort_values(by=sort_col, ascending=False)
    return localize_dataset_leaderboard_df(df.reset_index(drop=True))


def get_dataset_info_markdown(task_name: str) -> str:
    return get_dataset_readme_markdown(task_name)


def get_empty_leaderboard_notice() -> str:
    return t("empty_leaderboard")


def get_empty_leaderboard_notice_update(main_df: pd.DataFrame):
    if main_df.empty:
        return gr.update(value=get_empty_leaderboard_notice(), visible=True)
    return gr.update(value="", visible=False)


def get_all_leaderboard_outputs():
    """Reload main and per-dataset tables from results/."""
    from src.display.utils import BENCHMARK_COLS, COLS

    main_df = get_leaderboard_df(COLS, BENCHMARK_COLS)
    dataset_dfs = [get_dataset_leaderboard_df(task.name) for task in Tasks]
    return (main_df, *dataset_dfs)


def reload_leaderboard_tables():
    """Reload all tables, syncing persisted results on HF Spaces first."""
    sync_results_from_hub()
    outputs = get_all_leaderboard_outputs()
    return (*outputs, get_empty_leaderboard_notice_update(outputs[0]))