import os import json import datetime from email.utils import parseaddr import gradio as gr import pandas as pd from datasets import load_dataset from apscheduler.schedulers.background import BackgroundScheduler from huggingface_hub import HfApi, login from scorer import instruction_scorer from content import format_error, format_warning, format_log, TITLE, INTRODUCTION_TEXT, SUBMISSION_TEXT, CITATION_BUTTON_LABEL, CITATION_BUTTON_TEXT, model_hyperlink TOKEN = os.environ.get("TOKEN", None) if TOKEN: login(token=TOKEN) else: print("Token not found. Please set the HUGGINGFACE_TOKEN environment variable.") OWNER="Taejin" REF_JSONS_SET = f"{OWNER}/speaker_tagging_lb_refs" RESULTS_DATASET = f"{OWNER}/spk_tag_results" LEADERBOARD_PATH = f"{OWNER}/leaderboard" SUBMISSION_DATASET = f"{OWNER}/submission_leaderboard" api = HfApi() YEAR_VERSION = "2024" def read_json_file(filepath): with open(filepath) as infile: data_dict = json.load(infile) return data_dict def save_json_file(filepath, data_dict): with open(filepath, "w") as outfile: json.dump(data_dict, outfile) os.makedirs("scored", exist_ok=True) print(f"Token loaded-1 : {TOKEN}") results_data_files = {"dev": "dev_set_data.csv", "eval": "eval_set_data.csv"} results = load_dataset(RESULTS_DATASET, data_files=results_data_files, token=TOKEN, download_mode="force_redownload", ignore_verifications=True) print(f"Token loaded-2 : {TOKEN}") ref_json_files = {"dev_src": "err_dev.src.seglst.json", "dev_ref": "err_dev.ref.seglst.json", "eval_src": "err_eval.src.seglst.json", "eval_ref": "err_eval.ref.seglst.json"} ref_jsons = load_dataset(REF_JSONS_SET, data_files=ref_json_files, token=TOKEN, download_mode="force_redownload", ignore_verifications=True) def get_dataframe_from_results(results, split): df = results[split].to_pandas() df = df.sort_values(by=["cpWER"], ascending=True) return df def restart_space(): api.restart_space(repo_id=LEADERBOARD_PATH, token=TOKEN) TYPES = ["markdown", "markdown", "markdown", "number", "number", "number"] dev_dataset_dataframe= get_dataframe_from_results(results=results, split="dev") eval_dataset_dataframe= get_dataframe_from_results(results=results, split="eval") def add_new_eval( system_name: str, method: str, path_to_file: str, organisation: str, mail: str, ): print(f"printing all inputs system_name: {system_name}, method: {method}, path_to_file: {path_to_file}, organisation: {organisation}, mail: {mail}") if len(system_name)==0: print("system_name none") raise gr.Error("Please provide a system_name name. Field empty!") if len(method)==0: print("method none") raise gr.Error("Please provide a method. Field empty!") if len(organisation)==0: print("org none") raise gr.Error("Please provide organisation information. Field empty!") # Very basic email parsing _, parsed_mail = parseaddr(mail) if not "@" in parsed_mail: print("email here") raise gr.Error("Please provide a valid email address.") # Check if the combination system_name/org already exists and prints a warning message if yes # if system_name.lower() in set([m.lower() for m in results["dev"]["System_name"]]) and organisation.lower() in set([o.lower() for o in results["dev"]["Organisation"]]): # print("system_name org combo here") # raise gr.Error("This system_name has been already submitted.") if path_to_file is None: print("file missing here") raise gr.Error("Please attach a file.") # Save submitted file time_atm = datetime.datetime.today() api.upload_file( repo_id=SUBMISSION_DATASET, path_or_fileobj=path_to_file.name, path_in_repo=f"{organisation}/{system_name}/{YEAR_VERSION}_raw_{time_atm}.json", repo_type="dataset", token=TOKEN ) # Compute score if "err_dev.hyp.seglst.json" in path_to_file.name: ref_file_path="seglst_files/err_dev.ref.seglst.json" mode = "dev" elif "err_eval.hyp.seglst.json" in path_to_file.name: ref_file_path="seglst_files/err_eval.ref.seglst.json" mode = "eval" else: basename = os.path.basename(path_to_file.name) raise gr.Error(f"{basename} is NOT a valid name. It should be either err_dev.hyp.seglst.json or err_eval.hyp.seglst.json") scores = instruction_scorer(file_path_input= path_to_file.name, ref_file_path=ref_file_path, system_name=system_name) path_or_fileobj=f"scored/{organisation}_{system_name}.json" scores_and_info = { "system_name": system_name, "method":method, "organisation": organisation, "email": mail, "cpWER": scores["cpWER"], "errors": scores["errors"], "length": scores["length"], } save_json_file(path_or_fileobj, data_dict=scores_and_info) # Save scored file api.upload_file( repo_id=SUBMISSION_DATASET, path_or_fileobj=path_or_fileobj, path_in_repo=f"{organisation}/{system_name}/{YEAR_VERSION}_scored_{time_atm}.json", repo_type="dataset", token=TOKEN ) # Actual submission eval_entry = { "system_name": system_name, "method":method, "organisation": organisation, "cpWER":scores["cpWER"], "errors":scores["errors"], "length":scores["length"], } if mode == "dev": dev_set_data_csv = "dev_set_data.csv" dev_dataset_dataframe = get_dataframe_from_results(results=results, split="dev") dev_dataset_dataframe = pd.concat([dev_dataset_dataframe, pd.DataFrame([eval_entry])], ignore_index=True) dev_dataset_dataframe.to_csv(dev_set_data_csv, index=False) api.upload_file( repo_id=RESULTS_DATASET, path_or_fileobj=dev_set_data_csv, path_in_repo=dev_set_data_csv, repo_type="dataset", token=TOKEN ) elif mode == "eval": eval_set_data_csv = "eval_set_data.csv" eval_dataset_dataframe = get_dataframe_from_results(results=results, split="eval") eval_dataset_dataframe = pd.concat([eval_dataset_dataframe, pd.DataFrame([eval_entry])], ignore_index=True) eval_dataset_dataframe.to_csv(eval_set_data_csv, index=False) api.upload_file( repo_id=RESULTS_DATASET, path_or_fileobj=eval_set_data_csv, path_in_repo=eval_set_data_csv, repo_type="dataset", token=TOKEN ) return format_log(f"system_name {system_name} submitted by {organisation} successfully! \nPlease refresh the val leaderboard, and wait a bit to see the score displayed") def refresh(): results_data_files = {"dev": "dev_set_data.csv", "eval": "eval_set_data.csv"} results = load_dataset(RESULTS_DATASET, data_files=results_data_files, token=TOKEN, download_mode="force_redownload", ignore_verifications=True) dev_results_dataframe = get_dataframe_from_results(results=results, split="dev") eval_results_dataframe = get_dataframe_from_results(results=results, split="eval") return dev_results_dataframe, eval_results_dataframe def upload_file(files): file_paths = [file.name for file in files] return file_paths for file_key in ['dev_src', 'dev_ref', 'eval_src', 'eval_ref']: ref_jsons[file_key].to_json(path_or_buf=f"seglst_files/{file_key}.json") buff_list = [x.strip() for x in open(f"seglst_files/{file_key}.json").readlines()] buff_str = ",\n".join(buff_list) seglst_json = f"[\n{buff_str}\n]" split, datatype = file_key.split("_") with open(f"seglst_files/err_{split}.{datatype}.seglst.json", "w") as f: f.write(seglst_json) demo = gr.Blocks() with demo: gr.HTML(TITLE) gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text") with gr.Row(): with gr.Accordion("🧐 Introduction", open=False): gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text") with gr.Row(): with gr.Accordion("🎯 Submission Guidelines", open=False): gr.Markdown(SUBMISSION_TEXT, elem_classes="markdown-text") with gr.Row(): with gr.Accordion("📙 Citation", open=False): citation_button = gr.TextArea( value=CITATION_BUTTON_TEXT, label=CITATION_BUTTON_LABEL, elem_id="citation-button", ) with gr.Tab("Results: Dev"): leaderboard_table_dev = gr.components.Dataframe( value=dev_dataset_dataframe, datatype=TYPES, interactive=False, column_widths=["20%"] ) with gr.Tab("Results: Eval"): leaderboard_table_eval = gr.components.Dataframe( value=eval_dataset_dataframe, datatype=TYPES, interactive=False, column_widths=["20%"] ) refresh_button = gr.Button("Refresh") refresh_button.click( refresh, inputs=[], outputs=[ leaderboard_table_dev, leaderboard_table_eval, ], ) with gr.Accordion("Submit a new system_name for evaluation"): with gr.Row(): with gr.Column(): system_name_textbox = gr.Textbox(label="System name", type='text') method_textbox = gr.Textbox(label="Method (LLM with prompt, beam-search, etc)", type='text') with gr.Column(): organisation = gr.Textbox(label="Organisation or Team Name", type='text') mail = gr.Textbox(label="Contact email (will be stored privately, & used if there is an issue with your submission)", type='email') file_output = gr.File() submit_button = gr.Button("Submit Eval") submission_result = gr.Markdown() submit_button.click( add_new_eval, [ system_name_textbox, method_textbox, file_output, organisation, mail ], submission_result, ) scheduler = BackgroundScheduler() scheduler.add_job(restart_space, "interval", seconds=3600) scheduler.start() demo.launch(debug=True)