import huggingface_hub as hf_hub import uuid import glob import json import constants hf_api = hf_hub.HfApi( endpoint="https://huggingface.co", token=constants.TOKEN_HUB, ) # ## ### ## # def is_model_on_hub(model_id): """Check if a model is on the hub and return a failure message if not.""" try: author = model_id.split("/")[0] model_name = model_id.split("/")[1] if len(author) == 0 or len(model_name) == 0: return "is not a valid model name. Please use the format `author/model_name`." except: return "is not a valid model name. Please use the format `author/model_name`." try: models = list(hf_api.list_models(author=author, search=model_name)) matched = [model_id for m in models if m.modelId == model_id] if len(matched) != 1: return "was not found on the hub!" else: return None except: return "was not found on hub!" def request_model(model_id, evaluated_models, requested_models): def styled_error(error): return f"
{error}
" def styled_message(message): return f"{message}
" # Check if the model is on the hub error_msg = is_model_on_hub(model_id) if error_msg != None: return styled_error(f"{model_id} {error_msg}") # Check if the model was already evaluated if model_id in evaluated_models: return styled_error(f"{model_id} has already been evaluated.") # Check if the model was already requested if model_id in requested_models: return styled_error(f"A request for {model_id} was already made.") # Add the model to the local evaluation queue with open("requested_models.txt", "a") as f: f.write(model_id + "\n") # Push the request file to the private dataset repo unique_request_filename = str(uuid.uuid4()) + ".json" with open(unique_request_filename, "w") as f: json.dump(model_id, f) hf_api.upload_file( path_or_fileobj=unique_request_filename, #local_filepath path_in_repo=unique_request_filename, repo_id=constants.DATASET_REPO, token=constants.TOKEN_HUB, repo_type="dataset", commit_message=f"{model_id} added to the evaluation queue.", ) # Record the request requested_models.append(model_id) return styled_message("🤗 Your request has been submitted and will be evaluated as soon as possible!") def get_requested_models(): """Returns the list of requested models present in the private dataset repo.""" path_ = hf_api.snapshot_download( repo_id=constants.DATASET_REPO, repo_type="dataset", allow_patterns="*.json", # token=constants.TOKEN_HUB, ) requested_models = [] for json_file in glob.glob(path_ + "/*.json"): with open(json_file, "r") as f: model_id = json.load(f) requested_models.append(model_id) return requested_models