app.py

import json

import requests

from datasets import load_dataset

import gradio as gr
from apscheduler.schedulers.background import BackgroundScheduler


from huggingface_hub import HfApi, hf_hub_download
from huggingface_hub.repocard import metadata_load
import pandas as pd

from utils import *


block = gr.Blocks()

# Containing the data
rl_envs = [
{
"rl_env_beautiful": "LunarLander-v2 🚀",
"rl_env": "LunarLander-v2",
"video_link": "",
"global": None
},    
{
"rl_env_beautiful": "CartPole-v1",
"rl_env": "CartPole-v1",
"video_link": "https://huggingface.co/sb3/ppo-CartPole-v1/resolve/main/replay.mp4",
"global": None
},
{
"rl_env_beautiful": "FrozenLake-v1-4x4-no_slippery ❄️",
"rl_env": "FrozenLake-v1-4x4-no_slippery",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "FrozenLake-v1-8x8-no_slippery ❄️",
"rl_env": "FrozenLake-v1-8x8-no_slippery",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "FrozenLake-v1-4x4 ❄️",
"rl_env": "FrozenLake-v1-4x4",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "FrozenLake-v1-8x8 ❄️",
"rl_env": "FrozenLake-v1-8x8",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "Taxi-v3 🚖",
"rl_env": "Taxi-v3",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "CarRacing-v0 🏎️",
"rl_env": "CarRacing-v0",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "MountainCar-v0 ⛰️",
"rl_env": "MountainCar-v0",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "SpaceInvadersNoFrameskip-v4 👾",
"rl_env": "SpaceInvadersNoFrameskip-v4",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "PongNoFrameskip-v4 🎾",
"rl_env": "PongNoFrameskip-v4",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "BreakoutNoFrameskip-v4 🧱",
"rl_env": "BreakoutNoFrameskip-v4",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "QbertNoFrameskip-v4 🐦",
"rl_env": "QbertNoFrameskip-v4",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "BipedalWalker-v3",
"rl_env": "BipedalWalker-v3",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "Walker2DBulletEnv-v0",
"rl_env": "Walker2DBulletEnv-v0",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "AntBulletEnv-v0",
"rl_env": "AntBulletEnv-v0",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "HalfCheetahBulletEnv-v0",
"rl_env": "HalfCheetahBulletEnv-v0",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "PandaReachDense-v2",
"rl_env": "PandaReachDense-v2",
"video_link": "",
"global": None
},
{
"rl_env_beautiful": "Pixelcopter-PLE-v0",
"rl_env": "Pixelcopter-PLE-v0",
"video_link": "",
"global": None
}
]



def get_metadata(model_id):
    try:
        readme_path = hf_hub_download(model_id, filename="README.md")
        return metadata_load(readme_path)
    except requests.exceptions.HTTPError:
        # 404 README.md not found
        return None
        
def parse_metrics_accuracy(meta):
    if "model-index" not in meta:
        return None
    result = meta["model-index"][0]["results"]
    metrics = result[0]["metrics"]
    accuracy = metrics[0]["value"]
    return accuracy

# We keep the worst case episode
def parse_rewards(accuracy):
    default_std = -1000
    default_reward=-1000
    if accuracy !=  None:
        accuracy = str(accuracy)
        parsed =  accuracy.split(' +/- ')
        if len(parsed)>1:
            mean_reward = float(parsed[0])
            std_reward =  float(parsed[1])
        elif len(parsed)==1: #only mean reward   
            mean_reward = float(parsed[0])
            std_reward =  float(0)
             
        else: 
            mean_reward = float(default_std)
            std_reward = float(default_reward)

    else:
        mean_reward = float(default_std)
        std_reward = float(default_reward)
    return mean_reward, std_reward


def get_model_ids(rl_env):
    api = HfApi()
    models = api.list_models(filter=rl_env)
    model_ids = [x.modelId for x in models]
    #print(model_ids)
    return model_ids

def get_model_dataframe(rl_env):
    # Get model ids associated with rl_env
    model_ids = get_model_ids(rl_env)
    #print(model_ids)
    data = []
    for model_id in model_ids:
        """
        readme_path = hf_hub_download(model_id, filename="README.md")
        meta = metadata_load(readme_path)
        """
        meta = get_metadata(model_id)
        #LOADED_MODEL_METADATA[model_id] = meta if meta is not None else ''
        if meta is None:
            continue
        user_id = model_id.split('/')[0]
        row = {}
        row["User"] = make_clickable_user(user_id)
        row["Model"] = make_clickable_model(model_id)
        accuracy = parse_metrics_accuracy(meta)
        mean_reward, std_reward = parse_rewards(accuracy)
        mean_reward = mean_reward if not pd.isna(mean_reward) else 0
        std_reward = std_reward if not pd.isna(std_reward) else 0
        row["Results"] = mean_reward - std_reward
        row["Mean Reward"] = mean_reward
        row["Std Reward"] = std_reward
        data.append(row)
    print("DATA", data)
    ranked_dataframe = rank_dataframe(pd.DataFrame.from_records(data))
    print("RANKED", ranked_dataframe)
    return ranked_dataframe
    
    
def rank_dataframe(dataframe):
    #print("DATAFRAME", dataframe)
    dataframe = dataframe.sort_values(by=['Results'], ascending=False)
    if not 'Ranking' in dataframe.columns:
        dataframe.insert(0, 'Ranking', [i for i in range(1,len(dataframe)+1)])
    else:
        dataframe['Ranking'] =   [i for i in range(1,len(dataframe)+1)]
    return dataframe


with block:
    gr.Markdown(f"""
    # 🏆 The Deep Reinforcement Learning Course Leaderboard 🏆 
    
    This is the leaderboard of trained agents during the Deep Reinforcement Learning Course. A free course from beginner to expert.
    
    Just choose which environment you trained your agent on and with Ctrl+F find your rank 🏆

    **The leaderboard is updated every hour. If you don't find your model, go to the bottom of the page and click on the refresh button**
    
    We use **lower bound result to sort the models: mean_reward - std_reward.**

    You **can click on the model's name** to be redirected to its model card which includes documentation.
    
    🤖 You want to try to train your agents? <a href="https://huggingface.co/deep-rl-course/unit0/introduction?fw=pt" target="_blank"> Check the Hugging Face free Deep Reinforcement Learning Course 🤗 </a>.
    
    You want to compare two agents? <a href="https://huggingface.co/spaces/ThomasSimonini/Compare-Reinforcement-Learning-Agents" target="_blank">It's possible using this Spaces demo 👀 </a>.
    
    🔧 There is an **environment missing?** Please open an issue.

    For the RL course progress check out <a href="https://huggingface.co/spaces/ThomasSimonini/Check-my-progress-Deep-RL-Course" target="_blank"> User Progress App </a>
    """)
    
    #for rl_env in RL_ENVS:
    for i in range(0, len(rl_envs)):
        rl_env = rl_envs[i]
       
        with gr.TabItem(rl_env["rl_env_beautiful"]) as rl_tab:
            with gr.Row():
                markdown = """
                    # {name_leaderboard}
                    
                    """.format(name_leaderboard = rl_env["rl_env_beautiful"], video_link = rl_env["video_link"])
                gr.Markdown(markdown)
            with gr.Row():
                rl_env["global"] = gr.components.Dataframe(value= get_model_dataframe(rl_env["rl_env"]), headers=["Ranking 🏆", "User 🤗", "Model id 🤖", "Results", "Mean Reward", "Std Reward"], datatype=["number", "markdown", "markdown", "number", "number", "number"])
            with gr.Row():
                data_run = gr.Button("Refresh")
                #print("rl_env", rl_env["rl_env"])
                val = gr.Variable(value=[rl_env["rl_env"]])
                data_run.click(get_model_dataframe, inputs=[val], outputs =rl_env["global"])
            
                       
block.launch()

def refresh_leaderboard():
    """
    Here we refresh the leaderboard:
    we update the rl_env["global"] for each rl_envs in rl_env
    """
    for i in range(0, len(rl_envs)):
        rl_env = rl_envs[i]
        temp = get_model_dataframe(rl_env)
        rl_env["global"] = temp
    print("The leaderboard has been updated")

scheduler = BackgroundScheduler()
# Refresh every hour
scheduler.add_job(func=refresh_leaderboard, trigger="interval", seconds=3600)
scheduler.start()