import os import json import datetime from email.utils import parseaddr import gradio as gr import pandas as pd import numpy as np from datasets import load_dataset, DatasetDict from apscheduler.schedulers.background import BackgroundScheduler from huggingface_hub import HfApi # InfoStrings from scorer import question_scorer from content import ( format_error, format_warning, format_log, TITLE, INTRODUCTION_TEXT, CITATION_BUTTON_LABEL, CITATION_BUTTON_TEXT, model_hyperlink, ) TOKEN = os.environ.get("TOKEN", None) OWNER = "stemdataset" INTERNAL_DATA_DATASET = f"{OWNER}/STEM-Labels-Private" SUBMISSION_DATASET = f"{OWNER}/submissions_internal" CONTACT_DATASET = f"{OWNER}/contact_info" RESULTS_DATASET = f"{OWNER}/results" LEADERBOARD_PATH = f"{OWNER}/stem-leaderboard" api = HfApi() os.makedirs("scored", exist_ok=True) # Display the results eval_results = load_dataset( RESULTS_DATASET, token=TOKEN, download_mode="force_redownload", verification_mode="no_checks", ) contact_infos = load_dataset( CONTACT_DATASET, token=TOKEN, download_mode="force_redownload", verification_mode="no_checks", ) def get_dataframe_from_results(eval_results: DatasetDict, split): local_df = eval_results[split] local_df = local_df.map( lambda row: {"model": model_hyperlink(row["url"], row["model"])} ) local_df = local_df.remove_columns(["url"]) local_df = local_df.rename_column("model", "Model Name") local_df = local_df.rename_column("model_family", "Model Family") local_df = local_df.rename_column("average", "Average") local_df = local_df.rename_column("science", "Science") local_df = local_df.rename_column("technology", "Technology") local_df = local_df.rename_column("engineering", "Engineering") local_df = local_df.rename_column("math", "Math") local_df = local_df.rename_column("organisation", "Organisation") local_df = local_df.rename_column("submit_date", "Submit Date") df = pd.DataFrame(local_df) df = df[[ "Model Name", "Model Family", "Science", "Technology", "Engineering", "Math", "Average", "Organisation", "Submit Date", ]] df = df.sort_values(by=["Average"], ascending=False) numeric_cols = ["Science", "Technology", "Engineering", "Math", "Average"] df[numeric_cols] = df[numeric_cols].round(decimals=1) for col in numeric_cols: df[col] = df[col].apply(lambda x: f"{x:.1f}") return df eval_dataframe_test = get_dataframe_from_results( eval_results=eval_results, split="basic" ) # Gold answers gold_dataset = load_dataset(INTERNAL_DATA_DATASET, token=TOKEN)["labels"] def restart_space(): api.restart_space(repo_id=LEADERBOARD_PATH, token=TOKEN) TYPES = ["markdown", "number", "number", "number", "number", "str", "str"] def calc_test_acc(preds: list[int]) -> dict[str, float]: tmp_accs = { "science": [0, 0], "technology": [0, 0], "engineer": [0, 0], "math": [0, 0], } labels = gold_dataset for pred, label in zip(preds, labels): subject = label["subject"] tmp_accs[subject][1] += 1 if pred == label["answer_idx"]: tmp_accs[subject][0] += 1 accs = {k: v[0] / v[1] for k, v in tmp_accs.items()} accs["average"] = np.mean(list(accs.values())) accs = {k: round(v * 100, 1) for k, v in accs.items()} return accs def add_new_eval( val_or_test: str, model: str, model_family: str, url: str, path_to_file: gr.File, organisation: str, mail: str, ): curr_timestamp = datetime.datetime.today() # Very basic email parsing _, parsed_mail = parseaddr(mail) if not "@" in parsed_mail: return format_warning("Please provide a valid email adress.") if model == "": return format_warning("Please provide a model name.") if model_family == "": return format_warning("Please provide a model family.") print( json.dumps( { "val_or_test": val_or_test, "model": model, "model_family": model_family, "url": url, "path_to_file": path_to_file, "organisation": organisation, "mail": mail, }, indent=2, ) ) print("Adding new eval") # Check if the combination model/org already exists and prints a warning message if yes if model.lower() in set( [m.lower() for m in eval_results["basic"]["model"]] ) and organisation.lower() in set( [l.lower() for l in eval_results["basic"]["organisation"]] ): return format_warning("This model has been already submitted.") if path_to_file is None: return format_warning("Please attach a file.") # Save submitted file api.upload_file( repo_id=SUBMISSION_DATASET, path_or_fileobj=path_to_file.name, path_in_repo=f"{organisation}/{model}/{val_or_test}_raw_{curr_timestamp}.txt", repo_type="dataset", token=TOKEN, ) # Compute score file_path = path_to_file.name with open(f"scored/{organisation}_{model}.json", "w") as scored_file: with open(file_path, "r") as f: preds = [] for ix, line in enumerate(f): try: pred_idx = int(line.strip()) except Exception: return format_error( f"Line {ix} is incorrectly formatted. Please fix it and resubmit your file." ) preds.append(pred_idx) stem_scores = calc_test_acc(preds) scored_file.write(json.dumps(stem_scores, indent=2)) # Save scored file api.upload_file( repo_id=SUBMISSION_DATASET, path_or_fileobj=f"scored/{organisation}_{model}.json", path_in_repo=f"{organisation}/{model}/{val_or_test}_scored_{curr_timestamp}.json", repo_type="dataset", token=TOKEN, ) # Actual submission eval_entry = { "model": model, "model_family": model_family, "url": url, "organisation": organisation, "submit_date": "\n".join(str(curr_timestamp).split(" ")), "science": stem_scores["science"], "technology": stem_scores["technology"], "engineering": stem_scores["engineer"], "math": stem_scores["math"], "average": stem_scores["average"], } eval_results["basic"] = eval_results["basic"].add_item(eval_entry) print(eval_results) eval_results.push_to_hub(RESULTS_DATASET, token=TOKEN) contact_info = { "model": model, "model_family": model_family, "url": url, "organisation": organisation, "mail": mail, "submit_date": "\n".join(str(curr_timestamp).split(" ")), } contact_infos["basic"] = contact_infos["basic"].add_item(contact_info) contact_infos.push_to_hub(CONTACT_DATASET, token=TOKEN) return format_log( f"Model {model} submitted by {organisation} successfully. \nPlease refresh the leaderboard, and wait a bit to see the score displayed" ) def refresh(): eval_results = load_dataset( RESULTS_DATASET, token=TOKEN, download_mode="force_redownload", verification_mode="no_checks", ) eval_dataframe_test = get_dataframe_from_results( eval_results=eval_results, split="basic" ) return eval_dataframe_test def upload_file(files): file_paths = [file.name for file in files] return file_paths demo = gr.Blocks() with demo: gr.HTML(TITLE) gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text") with gr.Row(): with gr.Accordion("📙 Citation", open=False): citation_button = gr.Textbox( value=CITATION_BUTTON_TEXT, label=CITATION_BUTTON_LABEL, elem_id="citation-button", ) with gr.Tab("Results: Test"): leaderboard_table_test = gr.components.Dataframe( value=eval_dataframe_test, datatype=TYPES, interactive=False, wrap=True, ) refresh_button = gr.Button("Refresh") refresh_button.click( refresh, inputs=[], outputs=[ leaderboard_table_test, ], ) with gr.Accordion("Submit a new model for evaluation"): with gr.Row(): with gr.Column(): level_of_test = gr.Radio(["test"], value="test", label="Split") model_name_textbox = gr.Textbox(label="Model name") model_family_textbox = gr.Textbox(label="Model family") url_textbox = gr.Textbox(label="Url to model information") with gr.Column(): organisation = gr.Textbox(label="Organisation") mail = gr.Textbox( label="Contact email (will be stored privately, & used if there is an issue with your submission)" ) file_output = gr.File() submit_button = gr.Button("Submit Eval") submission_result = gr.Markdown() submit_button.click( add_new_eval, [ level_of_test, model_name_textbox, model_family_textbox, url_textbox, file_output, organisation, mail, ], submission_result, ) scheduler = BackgroundScheduler() scheduler.add_job(restart_space, "interval", seconds=3600) scheduler.start() demo.launch(debug=True, server_name="0.0.0.0")