Spaces:
Running
Running
File size: 4,083 Bytes
de6fc01 0339ef7 de6fc01 0339ef7 de6fc01 0339ef7 1cf1270 0339ef7 1cf1270 0339ef7 1cf1270 0339ef7 1cf1270 0339ef7 1cf1270 0339ef7 1cf1270 0339ef7 1cf1270 0339ef7 1cf1270 0339ef7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 |
import gradio as gr
import asyncio
import time
import threading
from src.retrieve_data import populate_lb_data
leaderboard_data = {}
async def fetch_data():
"""Fetch the leaderboard data asynchronously"""
global leaderboard_data
try:
data = await populate_lb_data()
leaderboard_data = data
return True
except Exception as e:
print(f"Error fetching data: {e}")
return False
def background_update():
"""Background thread function to update data every 5 minutes"""
while True:
print("Updating leaderboard data...")
asyncio.run(fetch_data())
time.sleep(300) # 5 minutes
def create_table_for_lb(lb_name, gpu_name):
"""Create a formatted table for a specific leaderboard and GPU"""
if (
not lb_name
or not gpu_name
or lb_name not in leaderboard_data
or gpu_name not in leaderboard_data[lb_name]
):
return gr.Dataframe(
headers=["Rank", "Submission Name", "User ID", "Score", "Date"],
datatype=["number", "str", "str", "str", "str"],
value=[],
interactive=False,
)
lb_data = leaderboard_data[lb_name][gpu_name]
headers = ["Rank", "Submission Name", "User ID", "Score", "Date"]
rows = []
for i, result in enumerate(lb_data.results, 1):
rows.append(
[
i,
result.submission_name,
result.user_id,
f"{float(result.submission_score):.4f}",
result.submission_time,
]
)
return gr.Dataframe(
headers=headers,
datatype=["number", "str", "str", "str", "str"],
value=rows,
interactive=False,
)
def refresh_ui():
"""Force refresh the UI with latest data"""
asyncio.run(fetch_data())
return "Data refreshed!", get_lb_names(), [], None
def get_lb_names():
"""Get list of available leaderboard names"""
return list(leaderboard_data.keys())
def get_gpu_names(lb_name):
"""Get list of available GPUs for a specific leaderboard"""
if not lb_name or lb_name not in leaderboard_data:
return []
return list(leaderboard_data[lb_name].keys())
def on_lb_change(lb_name):
gpu_choices = get_gpu_names(lb_name)
return (
gr.update(choices=gpu_choices, value=gpu_choices[0] if gpu_choices else None),
update_table(lb_name, gpu_choices[0]),
)
def update_table(lb_name, gpu_name):
"""Update the table based on selected leaderboard and GPU"""
if not lb_name or not gpu_name:
return None
return create_table_for_lb(lb_name, gpu_name)
def build_ui():
with gr.Blocks(title="ML Leaderboards", theme=gr.themes.Soft()) as app:
gr.Markdown("# 🍿 KernelBot Leaderboard 🍿")
asyncio.run(fetch_data())
with gr.Row():
with gr.Column(scale=1):
lb_dropdown = gr.Dropdown(
choices=get_lb_names(),
label="Select Leaderboard",
interactive=True,
)
gpu_dropdown = gr.Dropdown(
choices=get_gpu_names(lb_dropdown.value),
label="Select GPU",
interactive=True,
)
with gr.Row():
results_table = gr.Dataframe(
headers=["Rank", "Submission Name", "User ID", "Score", "Date"],
datatype=["number", "str", "str", "str", "str"],
interactive=False,
label="Results",
)
lb_dropdown.change(
fn=on_lb_change,
inputs=[lb_dropdown],
outputs=[gpu_dropdown, results_table],
)
gpu_dropdown.change(
fn=update_table, inputs=[lb_dropdown, gpu_dropdown], outputs=results_table
)
return app
if __name__ == "__main__":
update_thread = threading.Thread(target=background_update, daemon=True)
update_thread.start()
app = build_ui()
app.launch()
|