Spaces:
Sleeping
Sleeping
import streamlit as st | |
import wandb | |
import pandas as pd | |
import os | |
import time | |
import datetime | |
from typing import Dict, Any | |
from utils import fetch_competition_summary, fetch_models_evaluation, highlight_score_column, get_competition_id | |
# from dotenv import load_dotenv | |
from cfg import * | |
# load_dotenv() | |
st.set_page_config(layout="wide") | |
table_font_size = 16 | |
st.markdown(TABLE_STYLE, unsafe_allow_html=True) | |
### WANDB | |
# Access the API key from the environment variable | |
wandb_api_key = os.getenv('WANDB_API_KEY') | |
# Log in to wandb using the API key | |
if wandb_api_key: | |
wandb.login(key=wandb_api_key) | |
# wandb.login() | |
else: | |
st.error("WANDB_API_KEY not found in environment variables.") | |
wandb_api = wandb.Api() | |
def update_leader_info( | |
leader_info: Dict[str, Dict[str, Any]], | |
competition: str, | |
best_model: Dict[str, Any] | |
) -> Dict[str, Any]: | |
if leader_info.get(competition) is None: | |
leader_info[competition] = { | |
"Miner hotkey": best_model["Miner hotkey"], | |
"Date": time.strftime("%Y-%m-%d"), | |
"Days on Top": 1 | |
} | |
else: | |
if leader_info[competition]["Miner hotkey"] == best_model["Miner hotkey"]: | |
# count the number of days on top | |
start_date = datetime.datetime.strptime(leader_info[competition]["Date"], "%Y-%m-%d") | |
current_date = datetime.datetime.now() | |
days_on_top = (current_date - start_date).days | |
leader_info[competition]["Days on Top"] = days_on_top + 1 | |
else: | |
leader_info[competition]["Miner hotkey"] = best_model["Miner hotkey"] | |
leader_info[competition]["Date"] = time.strftime("%Y-%m-%d") | |
leader_info[competition]["Days on Top"] = 1 | |
return leader_info[competition] | |
def load_competition_data(last_update_time=None): | |
competition_summaries = {} | |
model_evaluations = {} | |
competitions_id = get_competition_id(CONFIG_URL) | |
for competition in COMPETITIONS: | |
if COMPETITIONS[competition]["project"] in competitions_id: | |
competition_summaries[competition] = fetch_competition_summary(wandb_api, COMPETITIONS, competition) | |
model_evaluations[competition] = fetch_models_evaluation(wandb_api, COMPETITIONS, competition) | |
last_update_time = time.time() | |
return competition_summaries, model_evaluations, last_update_time | |
# Streamlit app main function | |
def main(): | |
st.markdown(HEADER, unsafe_allow_html=True) | |
competitions_id = get_competition_id(CONFIG_URL) | |
if 'last_update_time' not in st.session_state: | |
st.session_state.last_update_time = None | |
if "leader_info" not in st.session_state: | |
st.session_state.leader_info = {} | |
if 'selected_competition' not in st.session_state: | |
st.session_state.selected_competition = None | |
if st.session_state.last_update_time is None or (time.time() - st.session_state.last_update_time > UPDATE_INTERVAL): | |
competition_summaries, model_evaluations, st.session_state.last_update_time = load_competition_data(st.session_state.last_update_time) | |
for competition in COMPETITIONS: | |
if COMPETITIONS[competition]["project"] not in competitions_id: | |
continue | |
if not competition_summaries[competition].empty: | |
# get all winning hotkeys and number of wins | |
winning_hotkeys = competition_summaries[competition]["Winning Hotkey"].value_counts() | |
# if not empty, get the best hotkey | |
if not winning_hotkeys.empty: | |
best_hotkey = winning_hotkeys.idxmax() | |
# Filter models for the best hotkey | |
best_model_filtered = model_evaluations[competition][model_evaluations[competition]["Miner hotkey"] == best_hotkey] | |
# Check if the filtered DataFrame is not empty | |
if not best_model_filtered.empty: | |
best_model = best_model_filtered.iloc[0] | |
st.session_state.leader_info[competition] = update_leader_info(st.session_state.leader_info, competition, best_model) | |
else: | |
st.warning(f"No model found for the best hotkey: {best_hotkey} in competition {competition}.") | |
else: | |
st.session_state.leader_info[competition] = { | |
"Miner hotkey": "N/A", | |
"Date": "N/A", | |
"Days on Top": "N/A" | |
} | |
else: | |
competition_summaries, model_evaluations, _ = load_competition_data(st.session_state.last_update_time) | |
st.subheader("###") | |
st.markdown("<h2 style='text-align: center; font-size: 28px;'>Competitions</h2>", unsafe_allow_html=True) | |
st.write("#### Select a competition to view more details and rankings.") | |
# Create a header for the table | |
cols = st.columns([1, 3, 2, 2, 3, 1]) | |
headers = ["Index", "Competition Name", "Date", "Miner hotkey", "Days on Top"] | |
for col, header in zip(cols, headers): | |
col.write(header) | |
for index, (competition, details) in enumerate(COMPETITIONS.items(), start=1): | |
if COMPETITIONS[competition]["project"] not in competitions_id: | |
continue | |
leader_info = st.session_state.get("leader_info", {}).get(competition, {}) | |
cols = st.columns([1, 3, 2, 2, 3, 1]) | |
cols[0].write(index) | |
if cols[1].button(competition): | |
st.session_state.selected_competition = competition | |
cols[2].write(leader_info.get("Date", "N/A")) | |
cols[3].write(leader_info.get("Miner hotkey", "N/A")) | |
cols[4].write(leader_info.get("Days on Top", "N/A")) | |
if st.session_state.selected_competition: | |
competition_name = st.session_state.selected_competition | |
competition_details = COMPETITIONS.get(competition_name, {}) | |
description = competition_details.get("description", "No description available.") | |
st.subheader(f"Competition: {competition_name}") | |
st.write(description) | |
# Add search bar for miner hotkey | |
miner_hotkey_search = st.text_input("Search for Miner Hotkey", "") | |
st.write(description) | |
st.subheader("Competition Summary") | |
competition_summary_df = competition_summaries.get(competition_name, pd.DataFrame()) | |
# Filter the competition summary dataframe by miner hotkey if a search term is entered | |
if miner_hotkey_search: | |
competition_summary_df = competition_summary_df[ | |
competition_summary_df["Winning Hotkey"].str.contains(miner_hotkey_search, na=False, case=False) | |
] | |
if not competition_summary_df.empty: | |
st.dataframe(competition_summary_df, height=500, hide_index=True) | |
else: | |
st.warning("No competition summary data available.") | |
st.subheader("Models Evaluation") | |
models_evaluation_df = model_evaluations.get(competition_name, pd.DataFrame()) | |
# Filter the models evaluation dataframe by miner hotkey if a search term is entered | |
if miner_hotkey_search: | |
models_evaluation_df = models_evaluation_df[ | |
models_evaluation_df["Miner hotkey"].str.contains(miner_hotkey_search, na=False, case=False) | |
] | |
if not models_evaluation_df.empty: | |
st.dataframe(models_evaluation_df.style.apply(highlight_score_column, axis=0), height=500, hide_index=True) | |
else: | |
st.warning("No models evaluation data available.") | |
else: | |
st.write("Please select a competition to view details.") | |
# Run the app | |
if __name__ == "__main__": | |
main() |