app.py

import json
import math
import subprocess
import tempfile
from pathlib import Path

import numpy as np
import pandas as pd

import gradio as gr
from huggingface_hub import snapshot_download

from src.about import (
    CITATION_BUTTON_LABEL,
    CITATION_BUTTON_TEXT,
    EVALUATION_QUEUE_TEXT,
    INTRODUCTION_TEXT,
    LLM_BENCHMARKS_TEXT,
    TITLE,
    Tasks,
)
from src.display.css_html_js import custom_css
from src.display.utils import (
    BENCHMARK_COLS,
    COLS,
    EVAL_COLS,
    EVAL_TYPES,
    NUMERIC_INTERVALS,
    TYPES,
    AutoEvalColumn,
    ModelType,
    fields,
    WeightType,
    Precision
)
from src.envs import API, DATA_PATH
from src.populate import get_leaderboard_df


def restart_space():
    API.restart_space(repo_id=REPO_ID)


raw_data, original_df = get_leaderboard_df(DATA_PATH, COLS, BENCHMARK_COLS)
leaderboard_df = original_df.copy()



def export_csv(df):
    csv_filename = Path(tempfile._get_default_tempdir()) / f"scandeval_leaderboard_{next(tempfile._get_candidate_names())}.csv"
    df = df.copy()
    df[AutoEvalColumn.model.name] = df[AutoEvalColumn.model.name].apply(lambda x: x.split(">")[1][:-3])
    df.to_csv(csv_filename)
    return str(csv_filename)
    
    
def plot_stats(data_path, plotting_library="plotly", columns=None, table=None):
    plots = {}
    files = Path(data_path).rglob("*.jsonl")
    models = None
    if table is not None:
        models = table.data[AutoEvalColumn.model.name].apply(lambda x: x.split(">")[1][:-3]).values.tolist()
    if columns is not None:
        scores = {(task.value.benchmark, task.value.metric): task.value.col_name for task in Tasks if
                  task.value.col_name in columns}
    else:
        scores = {(task.value.benchmark, task.value.metric): task.value.col_name for task in Tasks}
    model_names = []
    for file in files:
        with open(file) as f:
            for line in f:
                if not line.strip():
                    continue
                line = json.loads(line)
                if line["model"] not in model_names:
                    model_names.append(line["model"])
                if models is not None and line["model"] not in models:
                    continue
                metrics = {}
                for r in line["results"]["raw"]["test"]:
                    for k, v in r.items():
                        key = (line["dataset"], k)
                        if key not in scores:
                            continue
                        val = plots.get(key, {})
                        val[line["model"]] = val.get(line["model"], []) + [v]
                        plots[key] = val
                        metrics.setdefault(k, []).append(v)

    # Boxplot
    # target_size = math.ceil(len(plots) ** 0.5)
    ncols = 2  # target_size if target_size ** 2 == len(plots) else target_size + 1
    nrows = len(plots) // 2 + len(plots) % 2  # target_size
    if plotting_library == "matplotlib":
        import matplotlib.pyplot as plt
        if not plots:
            return plt.subplots(1, 1)[0]

        fig, axs = plt.subplots(nrows, ncols, figsize=(10 * ncols, 10 * nrows))
        for i, (k, v) in enumerate(plots.items()):
            ax = axs[i // ncols, i % ncols]
            vk, vv = zip(*sorted(v.items()))
            ax.boxplot(vv, tick_labels=vk)
            # Tilt the x-axis labels slightly
            for tick in ax.get_xticklabels():
                tick.set_rotation(5)
            ax.set_title(scores[k])
        # fig.show()
        fig.tight_layout()
        # fig.savefig("results.png")
    elif plotting_library == "plotly":
        import plotly.express as px
        import plotly.graph_objects as go
        from plotly.subplots import make_subplots

        if not plots:
            return make_subplots(rows=1, cols=1)
        colors = dict(zip(model_names, px.colors.qualitative.Dark24))
        plot_titles = [scores[k] for k in plots.keys()]
        fig = make_subplots(rows=nrows + 1, cols=ncols, subplot_titles=[AutoEvalColumn.average.name] + plot_titles,
                            specs=[[{"colspan": ncols, "type": "scatterpolar"}] + [None] * (ncols - 1)] + [[{}] * ncols] * nrows,
                            start_cell="top-left",
                            vertical_spacing=0.05, horizontal_spacing=0.1)
        scatters = {}
        # print(fig.print_grid())
        for i, (k, v) in enumerate(plots.items()):
            vk, vv = zip(*sorted(v.items()))
            for j, (label, data) in enumerate(zip(vk, vv)):
                # Adding a box trace for each label in the subplot
                # print(i // ncols + 2, i % ncols + 1)
                fig.add_trace(go.Box(y=data, name=label, boxpoints=False, marker_color=colors[label], showlegend=False, legendgroup=f'group-{label}'),
                              row=i // ncols + 2, col=i % ncols + 1)
                if label not in scatters:
                    scatters[label] = {}
                scatters[label][plot_titles[i]] = np.mean(data) if max(data) < 1 else np.mean(data) / 100.0
        for label, data in scatters.items():
            fig.add_trace(go.Scatterpolar(
                r=tuple(data.values()),
                theta=tuple(data.keys()),
                fill="toself",
                name=label,
                line_color=colors[label],
                legendgroup=f'group-{label}',
            ), row=1, col=1)
        # Update xaxis properties for each subplot to rotate and center labels
        for i in range(nrows * ncols):
            fig.update_xaxes(tickangle=-15, row=i // ncols + 2, col=i % ncols + 1)

        fig.update_layout(
            height=500 * (nrows + 1),
            width=700 * ncols,
            showlegend=True,
            # Prevent plot from getting its top cut off, not showing the titles
            # margin keyword has no effect, instead we do:
            title_yanchor="top",
        )
        # fig.show()
    else:
        raise ValueError(f"Unknown plotting library: {plotting_library}")
    return fig


# Searching and filtering
def update_table_and_plot(
        hidden_df: pd.DataFrame,
        columns: list,
        type_query: list,
        precision_query: str,
        size_query: list,
        show_deleted: bool,
        query: str,
):
    filtered_df = filter_models(hidden_df, type_query, size_query, precision_query, show_deleted)
    filtered_df = filter_queries(query, filtered_df)
    df = select_columns(filtered_df, columns)
    export_filename = export_csv(df)
    df = (df.style
          .format(precision=2, thousands=",", decimal=".")
          .highlight_max(props="background-color: lightgreen; color: black;", axis=0, subset=df.columns[1:])
          .highlight_between(props="color: red;", axis=0, subset=df.columns[1:], left=-np.inf, right=-np.inf)
          )
    fig = plot_stats(DATA_PATH, columns=df.data.columns, table=df)
    return df, fig, export_filename


def search_table(df: pd.DataFrame, query: str) -> pd.DataFrame:
    if query.lower().startswith("not "):
        return df[~(df[AutoEvalColumn.model.name].str.contains(query[4:], case=False))]
    else:
        return df[(df[AutoEvalColumn.model.name].str.contains(query, case=False))]


def select_columns(df: pd.DataFrame, columns: list) -> pd.DataFrame:
    always_here_cols = [
        # AutoEvalColumn.model_type_symbol.name,
        AutoEvalColumn.model.name,
    ]
    # We use COLS to maintain sorting
    filtered_df = df[
        always_here_cols + [c for c in COLS if c in df.columns and c in columns]
        ]
    return filtered_df


def filter_queries(query: str, filtered_df: pd.DataFrame) -> pd.DataFrame:
    final_df = []
    if query != "":
        queries = [q.strip() for q in query.split(";")]
        for _q in queries:
            if _q != "" and not _q.lower().startswith("not "):
                temp_filtered_df = search_table(filtered_df, _q)
                if len(temp_filtered_df) > 0:
                    final_df.append(temp_filtered_df)
        if len(final_df) > 0:
            filtered_df = pd.concat(final_df)
            # filtered_df = filtered_df.drop_duplicates(
            #     subset=[AutoEvalColumn.model.name, AutoEvalColumn.precision.name, AutoEvalColumn.revision.name]
            # )
            filtered_df = filtered_df.drop_duplicates(
                subset=[AutoEvalColumn.model.name]
            )
        for _q in queries:
            if _q != "" and  _q.lower().startswith("not "):
                filtered_df = search_table(filtered_df, _q)
    return filtered_df


def filter_models(
        df: pd.DataFrame, type_query: list, size_query: list, precision_query: list, show_deleted: bool
) -> pd.DataFrame:
    # Show all models
    if show_deleted:
        filtered_df = df
    else:  # Show only still on the hub models
        filtered_df = df
        # filtered_df = df[df[AutoEvalColumn.still_on_hub.name] == True]

    # type_emoji = [t[0] for t in type_query]
    # filtered_df = filtered_df.loc[df[AutoEvalColumn.model_type_symbol.name].isin(type_emoji)]
    # filtered_df = filtered_df.loc[df[AutoEvalColumn.precision.name].isin(precision_query + ["None"])]

    # numeric_interval = pd.IntervalIndex(sorted([NUMERIC_INTERVALS[s] for s in size_query]))
    # params_column = pd.to_numeric(df[AutoEvalColumn.params.name], errors="coerce")
    # mask = params_column.apply(lambda x: any(numeric_interval.contains(x)))
    # filtered_df = filtered_df.loc[mask]

    return filtered_df


demo = gr.Blocks(css=custom_css)
with demo:
    gr.HTML(TITLE)
    gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text")

    with gr.Row():
        with gr.Column():
            with gr.Row():
                search_bar = gr.Textbox(
                    placeholder=" 🔍 Search for your model (separate multiple queries with `;`) and press ENTER...",
                    show_label=False,
                    elem_id="search-bar",
                )
            with gr.Row():
                shown_columns = gr.CheckboxGroup(
                    choices=[
                        c.name
                        for c in fields(AutoEvalColumn)
                        if not c.hidden and not c.never_hidden
                    ],
                    value=[
                        c.name
                        for c in fields(AutoEvalColumn)
                        if c.displayed_by_default and not c.hidden and not c.never_hidden
                    ],
                    label="Select columns to show",
                    elem_id="column-select",
                    interactive=True,
                )
            with gr.Row(visible=False):
                deleted_models_visibility = gr.Checkbox(
                    value=False, label="Show gated/private/deleted models", interactive=True
                )
        with gr.Column(min_width=320, visible=False):
            # with gr.Box(elem_id="box-filter"):
            filter_columns_type = gr.CheckboxGroup(
                label="Model types",
                choices=[t.to_str() for t in ModelType],
                value=[t.to_str() for t in ModelType],
                interactive=True,
                elem_id="filter-columns-type",
            )
            filter_columns_precision = gr.CheckboxGroup(
                label="Precision",
                choices=[i.value.name for i in Precision],
                value=[i.value.name for i in Precision],
                interactive=True,
                elem_id="filter-columns-precision",
            )
            filter_columns_size = gr.CheckboxGroup(
                label="Model sizes (in billions of parameters)",
                choices=list(NUMERIC_INTERVALS.keys()),
                value=list(NUMERIC_INTERVALS.keys()),
                interactive=True,
                elem_id="filter-columns-size",
            )
    with gr.Tabs(elem_classes="tab-buttons") as tabs:
        with gr.TabItem("🏅 LLM Benchmark", elem_id="llm-benchmark-tab-table", id=0):
            cols_to_show = [c.name for c in fields(AutoEvalColumn) if c.never_hidden] + shown_columns.value
            leaderboard_table = gr.components.Dataframe(
                value=(leaderboard_df[cols_to_show].style
                       .format(precision=2, thousands=",", decimal=".")
                       .highlight_max(props="background-color: lightgreen; color: black;", axis=0,
                                      subset=cols_to_show[1:])
                       .highlight_between(props="color: red;", axis=0,
                                          subset=cols_to_show[1:], left=-np.inf, right=-np.inf)
                       ),
                headers=[c.name for c in fields(AutoEvalColumn) if c.never_hidden] + shown_columns.value,
                datatype=TYPES,
                elem_id="leaderboard-table",
                interactive=False,
                visible=True,
            )
            leaderboard_file = gr.File(interactive=False, value=export_csv(leaderboard_df[cols_to_show]), visible=True)

        with gr.TabItem("📊 LLM Plots", elem_id="llm-benchmark-tab-plot", id=1):
            leaderboard_plot = gr.components.Plot(plot_stats(DATA_PATH))

    # Dummy leaderboard for handling the case when the user uses backspace key
    hidden_leaderboard_table_for_search = gr.components.Dataframe(
        value=original_df[COLS],
        headers=COLS,
        datatype=TYPES,
        visible=False,
    )
    search_bar.submit(
        update_table_and_plot,
        [
            hidden_leaderboard_table_for_search,
            shown_columns,
            filter_columns_type,
            filter_columns_precision,
            filter_columns_size,
            deleted_models_visibility,
            search_bar,
        ],
        [
            leaderboard_table,
            leaderboard_plot,
            leaderboard_file,
        ],
    )
    for selector in [shown_columns, filter_columns_type, filter_columns_precision, filter_columns_size,
                     deleted_models_visibility]:
        selector.change(
            update_table_and_plot,
            [
                hidden_leaderboard_table_for_search,
                shown_columns,
                filter_columns_type,
                filter_columns_precision,
                filter_columns_size,
                deleted_models_visibility,
                search_bar,
            ],
            [
                leaderboard_table,
                leaderboard_plot,
                leaderboard_file,
            ],
            queue=True,
        )

        # with gr.TabItem("📝 About", elem_id="llm-benchmark-tab-table", id=2):
        # gr.Markdown(LLM_BENCHMARKS_TEXT, elem_classes="markdown-text")

        # with gr.TabItem("🚀 Submit here! ", elem_id="llm-benchmark-tab-table", id=3):
        #     with gr.Column():
        #         with gr.Row():
        #             gr.Markdown(EVALUATION_QUEUE_TEXT, elem_classes="markdown-text")

        #         with gr.Column():
        #             with gr.Accordion(
        #                 f"✅ Finished Evaluations ({len(finished_eval_queue_df)})",
        #                 open=False,
        #             ):
        #                 with gr.Row():
        #                     finished_eval_table = gr.components.Dataframe(
        #                         value=finished_eval_queue_df,
        #                         headers=EVAL_COLS,
        #                         datatype=EVAL_TYPES,
        #                         row_count=5,
        #                     )
        #             with gr.Accordion(
        #                 f"🔄 Running Evaluation Queue ({len(running_eval_queue_df)})",
        #                 open=False,
        #             ):
        #                 with gr.Row():
        #                     running_eval_table = gr.components.Dataframe(
        #                         value=running_eval_queue_df,
        #                         headers=EVAL_COLS,
        #                         datatype=EVAL_TYPES,
        #                         row_count=5,
        #                     )

        #             with gr.Accordion(
        #                 f"⏳ Pending Evaluation Queue ({len(pending_eval_queue_df)})",
        #                 open=False,
        #             ):
        #                 with gr.Row():
        #                     pending_eval_table = gr.components.Dataframe(
        #                         value=pending_eval_queue_df,
        #                         headers=EVAL_COLS,
        #                         datatype=EVAL_TYPES,
        #                         row_count=5,
        #                     )
        #     with gr.Row():
        #         gr.Markdown("# ✉️✨ Submit your model here!", elem_classes="markdown-text")

        #     with gr.Row():
        #         with gr.Column():
        #             model_name_textbox = gr.Textbox(label="Model name")
        #             revision_name_textbox = gr.Textbox(label="Revision commit", placeholder="main")
        #             model_type = gr.Dropdown(
        #                 choices=[t.to_str(" : ") for t in ModelType if t != ModelType.Unknown],
        #                 label="Model type",
        #                 multiselect=False,
        #                 value=None,
        #                 interactive=True,
        #             )

        #         with gr.Column():
        #             precision = gr.Dropdown(
        #                 choices=[i.value.name for i in Precision if i != Precision.Unknown],
        #                 label="Precision",
        #                 multiselect=False,
        #                 value="float16",
        #                 interactive=True,
        #             )
        #             weight_type = gr.Dropdown(
        #                 choices=[i.value.name for i in WeightType],
        #                 label="Weights type",
        #                 multiselect=False,
        #                 value="Original",
        #                 interactive=True,
        #             )
        #             base_model_name_textbox = gr.Textbox(label="Base model (for delta or adapter weights)")

        #     submit_button = gr.Button("Submit Eval")
        #     submission_result = gr.Markdown()
        #     submit_button.click(
        #         add_new_eval,
        #         [
        #             model_name_textbox,
        #             base_model_name_textbox,
        #             revision_name_textbox,
        #             precision,
        #             weight_type,
        #             model_type,
        #         ],
        #         submission_result,
        #     )

    # with gr.Row():
    #     with gr.Accordion("📙 Citation", open=False):
    #         citation_button = gr.Textbox(
    #             value=CITATION_BUTTON_TEXT,
    #             label=CITATION_BUTTON_LABEL,
    #             lines=20,
    #             elem_id="citation-button",
    #             show_copy_button=True,
    #         )

# scheduler = BackgroundScheduler()
# scheduler.add_job(restart_space, "interval", seconds=1800)
# scheduler.start()
demo.queue(default_concurrency_limit=40).launch()