import json import logging import os from tempfile import TemporaryDirectory from typing import Dict, List, Optional import jsonlines from huggingface_hub import CommitOperationAdd # type: ignore[import] from huggingface_hub import Discussion, HfApi, HfFileSystem from tqdm import tqdm from .evaluation import METRICS from .formatting import styled_error, styled_message, styled_warning from .tasks import TASKS_PRETTY_REVERSE class AlreadyExists(Exception): pass class SubmissionUploader: """Class for adding new files to a dataset on a Hub and opening a PR. Heavily influenced by these amazing spaces: * https://huggingface.co/spaces/safetensors/convert * https://huggingface.co/spaces/gaia-benchmark/leaderboard """ def __init__(self, dataset_id: str): self._api = HfApi(token=os.environ["HF_TOKEN"]) self._fs = HfFileSystem(token=os.environ["HF_TOKEN"]) self._dataset_id = dataset_id def _get_previous_pr(self, pr_title: str) -> Optional[Discussion]: """Searches among discussions of dataset repo for a PR with the given title.""" try: discussions = self._api.get_repo_discussions( repo_id=self._dataset_id, repo_type="dataset" ) except Exception: return None for discussion in discussions: if ( discussion.status == "open" and discussion.is_pull_request and discussion.title == pr_title ): return discussion return None def _get_metadata( self, model_name_pretty: str, model_availability: str, urls: str, context_size: str, submitted_by: str, ) -> Dict[str, str]: return { "model_name": model_name_pretty, "model_availability": model_availability, "urls": urls, "context_size": context_size, "submitted_by": submitted_by, } def _upload_predictions( self, task_id: str, model_folder: str, filenames: List[str], ) -> List[CommitOperationAdd]: commit_operations = [ CommitOperationAdd( path_in_repo=f"{task_id}/predictions/{model_folder}/{os.path.basename(filename)}", path_or_fileobj=filename, ) for filename in filenames ] return commit_operations def _compute_metrics_for_predictions( self, task_id: str, filenames: Optional[List[str]], temp_directory: str ) -> None: metrics_module = METRICS[task_id] assert ( metrics_module is not None ), f"Computing metrics for {task_id} is not supported." metrics_module.reset() open(os.path.join(temp_directory, "metrics.jsonl"), "w").close() # compute the metrics for each submitted file for filename in filenames: with jsonlines.open(filename, "r") as reader: for example in tqdm( reader, desc=f"Computing metrics for {os.path.basename(filename)}" ): metrics_module.add_batch( predictions=[example["prediction"]], references=[example["reference"]], ) computed_metrics = metrics_module.compute() metrics_module.reset() with jsonlines.open( os.path.join(temp_directory, "metrics.jsonl"), "a" ) as writer: writer.write(computed_metrics) # aggregate the metrics over submitted files with jsonlines.open( os.path.join(temp_directory, "metrics.jsonl"), "r" ) as reader: metrics_results = [line for line in reader] final_metrics_results = { key: sum(entry[key] for entry in metrics_results) / len(metrics_results) for key in metrics_results[0] } with open(os.path.join(temp_directory, "final_metrics.json"), "w") as f: json.dump(final_metrics_results, f) def _upload_results( self, task_id: str, model_folder: str, model_name_pretty: str, model_availability: str, urls: str, context_size: str, submitted_by: str, temp_directory: str, ) -> List[CommitOperationAdd]: final_results = {} with open(os.path.join(temp_directory, "final_metrics.json"), "r") as f: metrics = json.load(f) final_results.update(metrics) metadata_dict = self._get_metadata( model_name_pretty=model_name_pretty, model_availability=model_availability, urls=urls, context_size=context_size, submitted_by=submitted_by, ) final_results.update(metadata_dict) with jsonlines.open( os.path.join(temp_directory, "final_results.jsonl"), "w" ) as writer: writer.write(final_results) return [ CommitOperationAdd( path_in_repo=f"{task_id}/results/{model_folder}.jsonl", path_or_fileobj=os.path.join(temp_directory, "final_results.jsonl"), ) ] def _verify_arguments( self, task_pretty: str, model_folder: str, model_name_pretty: str, model_availability: str, urls: str, context_size: str, submitted_by: str, filenames: Optional[List[str]], ): assert ( task_pretty and task_pretty in TASKS_PRETTY_REVERSE ), "Please, select one of the supported tasks." assert ( model_folder ), "Please, specify non-empty name for a directory with a model's results." assert model_name_pretty, "Please, specify non-empty name for a model." assert ( model_availability ), "Please, specify non-empty information about a model's availability." assert ( context_size ), "Please, specify non-empty information about a model's context size." try: _ = int(context_size) except: raise ValueError( "Please, specify a model's context size as an integer (e.g., 16000)." ) assert ( submitted_by ), "Please, specify non-empty information about a submission's author(s)." assert filenames, "Please, attach at least one file with predictions." def upload_files( self, task_pretty: str, model_folder: str, model_name_pretty: str, model_availability: str, urls: str, context_size: str, submitted_by: str, filenames: Optional[List[str]], force: bool = False, ) -> str: try: self._verify_arguments( task_pretty=task_pretty, model_folder=model_folder, model_name_pretty=model_name_pretty, model_availability=model_availability, urls=urls, context_size=context_size, submitted_by=submitted_by, filenames=filenames, ) pr_title = f"🚀 New submission to {task_pretty} task: {model_name_pretty} with {context_size} context size from {submitted_by}" logging.info(f"Start processing {pr_title}") task_id = TASKS_PRETTY_REVERSE[task_pretty] logging.info("Checking if this request has already been submitted...") if not force: if model_name_pretty in self._fs.ls( f"datasets/{self._dataset_id}/{task_id}/predictions" ) and all( filename in self._fs.ls( f"datasets/{self._dataset_id}/{task_id}/predictions/{model_name_pretty}" ) for filename in filenames + ["metadata.json"] ): return styled_warning( f"{model_name_pretty} is already present in {self._dataset_id}." ) prev_pr = self._get_previous_pr(pr_title) if prev_pr is not None: url = f"https://huggingface.co/datasets/{self._dataset_id}/discussions/{prev_pr.num}" return styled_warning( f"{self._dataset_id} already has an open PR for this submission: {url}." ) logging.info("Processing predictions...") predictions_commit_operations = self._upload_predictions( task_id=task_id, model_folder=model_folder, filenames=filenames, ) with TemporaryDirectory() as d: logging.info("Computing metrics...") self._compute_metrics_for_predictions( task_id=task_id, filenames=filenames, temp_directory=str(d) ) logging.info("Processing results...") results_commit_operations = self._upload_results( task_id=task_id, model_folder=model_folder, model_name_pretty=model_name_pretty, model_availability=model_availability, urls=urls, context_size=context_size, submitted_by=submitted_by, temp_directory=str(d), ) logging.info("Creating commit...") new_pr = self._api.create_commit( repo_id=self._dataset_id, operations=predictions_commit_operations + results_commit_operations, commit_message=pr_title, commit_description=f"""New submission to {task_pretty} task in 🏟️ Long Code Arena benchmark!\n* Model name: {model_name_pretty}\n* Model availability: {model_availability}\n* Context Size: {context_size}\n* Relevant URLs: {urls}\n* Submitted By: {submitted_by}""", create_pr=True, repo_type="dataset", ) return styled_message(f"🎉 PR created at {new_pr.pr_url}.") except Exception as e: logging.exception(e) if str(e): return styled_error(f"An exception occurred. Please, try again.\n{e}") return styled_error("An exception occurred. Please, try again.")