feat: Update with new results every 30 mins

app.py

@@ -10,6 +10,15 @@ from pydantic import BaseModel
 import gradio as gr
 import requests
 import random
+import logging
+import datetime as dt
+
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("radial_plot_generator")
+
+
+UPDATE_FREQUENCY_MINUTES = 30
 
 
 class Task(BaseModel):
@@ -130,46 +139,9 @@ DATASETS = [
 def main() -> None:
     """Produce a radial plot."""
 
-    # Download all the newest records
-    response = requests.get("https://scandeval.com/scandeval_benchmark_results.jsonl")
-    response.raise_for_status()
-    records = [
-        json.loads(dct_str)
-        for dct_str in response.text.split("\n")
-        if dct_str.strip("\n")
-    ]
-
-    # Build a dictionary of languages -> results-dataframes, whose indices are the
-    # models and columns are the tasks.
-    results_dfs = dict()
-    for language in {dataset.language for dataset in DATASETS}:
-        possible_dataset_names = {
-            dataset.name for dataset in DATASETS if dataset.language == language
-        }
-        data_dict = defaultdict(dict)
-        for record in records:
-            model_name = record["model"]
-            dataset_name = record["dataset"]
-            if dataset_name in possible_dataset_names:
-                dataset = next(
-                    dataset for dataset in DATASETS if dataset.name == dataset_name
-                )
-                results_dict = record['results']['total']
-                score = results_dict.get(
-                    f"test_{dataset.task.metric}", results_dict.get(dataset.task.metric)
-                )
-                if dataset.task in data_dict[model_name]:
-                    data_dict[model_name][dataset.task].append(score)
-                else:
-                    data_dict[model_name][dataset.task] = [score]
-        results_df = pd.DataFrame(data_dict).T.map(
-            lambda list_or_nan:
-            np.mean(list_or_nan) if list_or_nan == list_or_nan else list_or_nan
-        ).dropna()
-        if any(task not in results_df.columns for task in ALL_TASKS):
-            results_dfs[language] = pd.DataFrame()
-        else:
-            results_dfs[language] = results_df
+    global last_fetch
+    results_dfs = fetch_results()
+    last_fetch = dt.datetime.now()
 
     all_languages: list[str | int | float | tuple[str, str | int | float]] | None = [
         language.name for language in ALL_LANGUAGES.values()
@@ -251,7 +223,6 @@ def main() -> None:
             outputs=plot,
         )
 
-
     demo.launch()
 
 
@@ -272,6 +243,8 @@ def update_model_ids_dropdown(
     if results_dfs is None or len(language_names) == 0:
         return gr.update(choices=[], value=[])
 
+    # Download the newest records if it has been more than 5 minutes since the last
+
     filtered_results_dfs = {
         language: df
         for language, df in results_dfs.items()
@@ -300,7 +273,7 @@ def produce_radial_plot(
     model_ids: list[str],
     language_names: list[str],
     use_win_ratio: bool,
-    results_dfs: dict[Language, pd.DataFrame] | None
+    results_dfs: dict[Language, pd.DataFrame] | None,
 ) -> go.Figure:
     """Produce a radial plot as a plotly figure.
 
@@ -320,6 +293,17 @@ def produce_radial_plot(
     if results_dfs is None or len(language_names) == 0 or len(model_ids) == 0:
         return go.Figure()
 
+    global last_fetch
+    minutes_since_last_fetch = (dt.datetime.now() - last_fetch).total_seconds() / 60
+    if minutes_since_last_fetch > UPDATE_FREQUENCY_MINUTES:
+        results_dfs = fetch_results()
+        last_fetch = dt.datetime.now()
+
+    logger.info(
+        f"Producing radial plot for models {model_ids!r} on languages "
+        f"{language_names!r}..."
+    )
+
     tasks = ALL_TASKS
     languages = [ALL_LANGUAGES[language_name] for language_name in language_names]
 
@@ -386,7 +370,62 @@ def produce_radial_plot(
         polar=dict(radialaxis=dict(visible=True)), showlegend=True, title=title
     )
 
+    logger.info("Successfully produced radial plot.")
+
     return fig
 
+def fetch_results() -> dict[Language, pd.DataFrame]:
+    """Fetch the results from the ScandEval benchmark.
+
+    Returns:
+        A dictionary of languages -> results-dataframes, whose indices are the
+        models and columns are the tasks.
+    """
+    logger.info("Fetching results from ScandEval benchmark...")
+
+    response = requests.get("https://scandeval.com/scandeval_benchmark_results.jsonl")
+    response.raise_for_status()
+    records = [
+        json.loads(dct_str)
+        for dct_str in response.text.split("\n")
+        if dct_str.strip("\n")
+    ]
+
+    # Build a dictionary of languages -> results-dataframes, whose indices are the
+    # models and columns are the tasks.
+    results_dfs = dict()
+    for language in {dataset.language for dataset in DATASETS}:
+        possible_dataset_names = {
+            dataset.name for dataset in DATASETS if dataset.language == language
+        }
+        data_dict = defaultdict(dict)
+        for record in records:
+            model_name = record["model"]
+            dataset_name = record["dataset"]
+            if dataset_name in possible_dataset_names:
+                dataset = next(
+                    dataset for dataset in DATASETS if dataset.name == dataset_name
+                )
+                results_dict = record['results']['total']
+                score = results_dict.get(
+                    f"test_{dataset.task.metric}", results_dict.get(dataset.task.metric)
+                )
+                if dataset.task in data_dict[model_name]:
+                    data_dict[model_name][dataset.task].append(score)
+                else:
+                    data_dict[model_name][dataset.task] = [score]
+        results_df = pd.DataFrame(data_dict).T.map(
+            lambda list_or_nan:
+            np.mean(list_or_nan) if list_or_nan == list_or_nan else list_or_nan
+        ).dropna()
+        if any(task not in results_df.columns for task in ALL_TASKS):
+            results_dfs[language] = pd.DataFrame()
+        else:
+            results_dfs[language] = results_df
+
+    logger.info("Successfully fetched results from ScandEval benchmark.")
+
+    return results_dfs
+
 if __name__ == "__main__":
     main()