|
import os |
|
import random |
|
import pandas as pd |
|
from datetime import datetime |
|
from huggingface_hub import HfApi |
|
|
|
|
|
DATASET_REPO_URL = "https://huggingface.co/datasets/CarlCochet/BotFightData" |
|
ELO_FILENAME = "elo.csv" |
|
ELO_DIR = "soccer_elo" |
|
ELO_FILE = os.path.join(ELO_DIR, ELO_FILENAME) |
|
HF_TOKEN = os.environ.get("HF_TOKEN") |
|
|
|
|
|
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")) |
|
|
|
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 = {"rank": [], "author": [], "model": [], "elo": [], "games_played": []} |
|
sorted_models = sorted(self.models, key=lambda x: x.elo, reverse=True) |
|
for i, model in enumerate(sorted_models): |
|
data_dict["rank"].append(i + 1) |
|
data_dict["author"].append(model.author) |
|
data_dict["model"].append(model.name) |
|
data_dict["elo"].append(model.elo) |
|
data_dict["games_played"].append(model.games_played) |
|
|
|
df = pd.DataFrame(data_dict) |
|
fileobj = open('env_elos/elo.csv', 'w') |
|
df.to_csv(fileobj, index=False) |
|
api.upload_file( |
|
fileobj, |
|
"env_elos/elo.csv", |
|
"CarlCochet/BotFights", |
|
"Update elos", |
|
) |
|
df_matches = pd.DataFrame(self.matches) |
|
date = datetime.now() |
|
df_matches.to_csv(f"match_history/{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 = [] |
|
for i, row in data.iterrows(): |
|
models.append(Model(row["author"], row["model"], row["elo"], row["games_played"])) |
|
models_names.append(row["model"]) |
|
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() |
|
print("Matchmaking done ---", datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")) |
|
|
|
|
|
if __name__ == "__main__": |
|
print("It's running!") |
|
api = HfApi() |
|
init_matchmaking() |
|
|