Create src.results module

app.py

@@ -4,8 +4,9 @@ import gradio as gr
 import pandas as pd
 from huggingface_hub import HfFileSystem
 
-from src.constants import DETAILS_DATASET_ID, DETAILS_FILENAME, RESULTS_DATASET_ID, SUBTASKS, TASKS
-
+from src.constants import DETAILS_DATASET_ID, DETAILS_FILENAME, SUBTASKS, TASKS
+from src.results import fetch_result_paths, filter_latest_result_path_per_model, update_load_results_component, \
+    load_results_dataframes, display_results, update_tasks_component, clear_results
 
 fs = HfFileSystem()
 

src/results.py

@@ -0,0 +1,104 @@
+import json
+
+import gradio as gr
+import pandas as pd
+
+from app import fs, latest_result_path_per_model
+from src.constants import RESULTS_DATASET_ID, TASKS
+
+
+def fetch_result_paths():
+    paths = fs.glob(f"{RESULTS_DATASET_ID}/**/**/*.json")
+    return paths
+
+
+def filter_latest_result_path_per_model(paths):
+    from collections import defaultdict
+
+    d = defaultdict(list)
+    for path in paths:
+        model_id, _ = path[len(RESULTS_DATASET_ID) + 1:].rsplit("/", 1)
+        d[model_id].append(path)
+    return {model_id: max(paths) for model_id, paths in d.items()}
+
+
+def get_result_path_from_model(model_id, result_path_per_model):
+    return result_path_per_model[model_id]
+
+
+def update_load_results_component():
+    return gr.Button("Load Results", interactive=True)
+
+
+def load_data(result_path) -> pd.DataFrame:
+    with fs.open(result_path, "r") as f:
+        data = json.load(f)
+    return data
+
+
+def load_results_dataframe(model_id):
+    if not model_id:
+        return
+    result_path = get_result_path_from_model(model_id, latest_result_path_per_model)
+    data = load_data(result_path)
+    model_name = data.get("model_name", "Model")
+    df = pd.json_normalize([{key: value for key, value in data.items()}])
+    # df.columns = df.columns.str.split(".")  # .split return a list instead of a tuple
+    return df.set_index(pd.Index([model_name])).reset_index()
+
+
+def load_results_dataframes(*model_ids):
+    return [load_results_dataframe(model_id) for model_id in model_ids]
+
+
+def display_results(task, *dfs):
+    dfs = [df.set_index("index") for df in dfs if "index" in df.columns]
+    if not dfs:
+        return None, None
+    df = pd.concat(dfs)
+    df = df.T.rename_axis(columns=None)
+    return display_tab("results", df, task), display_tab("configs", df, task)
+
+
+def display_tab(tab, df, task):
+    df = df.style.format(na_rep="")
+    df.hide(
+        [
+            row
+            for row in df.index
+            if (
+                not row.startswith(f"{tab}.")
+                or row.startswith(f"{tab}.leaderboard.")
+                or row.endswith(".alias")
+                or (not row.startswith(f"{tab}.{task}") if task != "All" else False)
+            )
+        ],
+        axis="index",
+    )
+    start = len(f"{tab}.leaderboard_") if task == "All" else len(f"{tab}.{task} ")
+    df.format_index(lambda idx: idx[start:].removesuffix(",none"), axis="index")
+    return df.to_html()
+
+
+def update_tasks_component():
+    return gr.Radio(
+            ["All"] + list(TASKS.values()),
+            label="Tasks",
+            info="Evaluation tasks to be displayed",
+            value="All",
+            interactive=True,
+        )
+
+
+def clear_results():
+    # model_id_1, model_id_2, dataframe_1, dataframe_2, task
+    return (
+        None, None, None, None,
+        gr.Radio(
+            ["All"] + list(TASKS.values()),
+            label="Tasks",
+            info="Evaluation tasks to be displayed",
+            value="All",
+            interactive=False,
+        ),
+    )