Spaces:
Runtime error
Runtime error
| import random | |
| import pandas as pd | |
| from datetime import datetime | |
| from huggingface_hub import HfApi | |
| env = "Soccer" | |
| class Model: | |
| """ | |
| Class containing the info of a model. | |
| :param name: Name of the model | |
| :param elo: Elo rating of the model | |
| :param games_played: Number of games played by the model (useful if we implement sigma uncertainty) | |
| """ | |
| def __init__(self, author, name, elo=1200, games_played=0): | |
| self.author = author | |
| self.name = name | |
| self.elo = elo | |
| self.games_played = games_played | |
| class Matchmaking: | |
| """ | |
| Class managing the matchmaking between the models. | |
| :param models: List of models | |
| :param queue: Temporary list of models used for the matching process | |
| :param k: Dev coefficient | |
| :param max_diff: Maximum difference considered between two models' elo | |
| :param matches: Dictionary containing the match history (to later upload as CSV) | |
| """ | |
| def __init__(self, models): | |
| self.models = models | |
| self.queue = self.models.copy() | |
| self.k = 20 | |
| self.max_diff = 500 | |
| self.matches = { | |
| "model1": [], | |
| "model2": [], | |
| "result": [], | |
| "datetime": [], | |
| "env": [] | |
| } | |
| def run(self): | |
| """ | |
| Run the matchmaking process. | |
| Add models to the queue, shuffle it, and match the models one by one to models with close ratings. | |
| Compute the new elo for each model after each match and add the match to the match history. | |
| """ | |
| self.queue = self.models.copy() | |
| random.shuffle(self.queue) | |
| while len(self.queue) > 1: | |
| model1 = self.queue.pop(0) | |
| model2 = self.queue.pop(self.find_n_closest_indexes(model1, 10)) | |
| result = match(model1, model2) | |
| self.compute_elo(model1, model2, result) | |
| self.matches["model1"].append(model1.name) | |
| self.matches["model2"].append(model2.name) | |
| self.matches["result"].append(result) | |
| self.matches["datetime"].append(datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")) | |
| self.matches["env"].append(env) | |
| def compute_elo(self, model1, model2, result): | |
| """ Compute the new elo for each model based on a match result. """ | |
| delta = model1.elo - model2.elo | |
| win_probability = 1 / (1 + 10 ** (-delta / 500)) | |
| model1.elo += self.k * (result - win_probability) | |
| model2.elo -= self.k * (result - win_probability) | |
| def find_n_closest_indexes(self, model, n) -> int: | |
| """ | |
| Get a model index with a fairly close rating. If no model is found, return the last model in the queue. | |
| We don't always pick the closest rating to add variety to the matchups. | |
| :param model: Model to compare | |
| :param n: Number of close models from which to pick a candidate | |
| :return: id of the chosen candidate | |
| """ | |
| indexes = [] | |
| closest_diffs = [9999999] * n | |
| for i, m in enumerate(self.queue): | |
| if m.name == model.name: | |
| continue | |
| diff = abs(m.elo - model.elo) | |
| if diff < max(closest_diffs): | |
| closest_diffs.append(diff) | |
| closest_diffs.sort() | |
| closest_diffs.pop() | |
| indexes.append(i) | |
| random.shuffle(indexes) | |
| return indexes[0] | |
| def to_csv(self): | |
| """ Save the match history as a CSV file to the hub. """ | |
| data_dict = {"author": [], "name": [], "elo": [], "games_played": []} | |
| for model in self.models: | |
| data_dict["author"].append(model.author) | |
| data_dict["name"].append(model.name) | |
| data_dict["elo"].append(model.elo) | |
| data_dict["games_played"].append(model.games_played) | |
| df = pd.DataFrame(data_dict) | |
| df.to_csv('env_elos/elo.csv', index=False) | |
| df_matches = pd.DataFrame(self.matches) | |
| date = datetime.now() | |
| df_matches.to_csv(f"matches/{date.strftime('%Y-%m-%d_%H-%M-%S_%f')}.csv", index=False) | |
| def match(model1, model2) -> float: | |
| """ | |
| !!! Current code is placeholder !!! | |
| TODO: Launch a Unity process with the 2 models and get the result of the match | |
| :param model1: First Model object | |
| :param model2: Second Model object | |
| :return: match result (0: model1 lost, 0.5: draw, 1: model1 won) | |
| """ | |
| result = random.randint(0, 2) / 2 | |
| model1.games_played += 1 | |
| model2.games_played += 1 | |
| return result | |
| def get_models_list() -> list: | |
| """ | |
| !!! Current code is placeholder !!! | |
| TODO: Create a list of Model objects from the models found on the hub | |
| :return: list of Model objects | |
| """ | |
| models = [] | |
| models_names = [] | |
| data = pd.read_csv("env_elos/elo.csv") | |
| # models_on_hub = api.list_models(filter=["reinforcement-learning", env, "stable-baselines3"]) | |
| models_on_hub = [] | |
| for i, row in data.iterrows(): | |
| models.append(Model(row["author"], row["name"], row["elo"], row["games_played"])) | |
| models_names.append(row["name"]) | |
| for model in models_on_hub: | |
| if model.modelId not in models_names: | |
| models.append(Model(model.author, model.modelId)) | |
| return models | |
| def init_matchmaking(): | |
| models = get_models_list() | |
| matchmaking = Matchmaking(models) | |
| matchmaking.run() | |
| matchmaking.to_csv() | |
| if __name__ == "__main__": | |
| print("It's running!") | |
| api = HfApi() | |
| init_matchmaking() | |