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_competitions, get_latest_evaluation_time # 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() @st.cache_data 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] @st.cache_data() def load_competition_data(last_update_time=None): competition_summaries = {} model_evaluations = {} competitions = get_competitions(CONFIG_URL) for competition_info in competitions: competition = competition_info[0] competition_summaries[competition] = fetch_competition_summary(wandb_api, ENTITY, competition) model_evaluations[competition] = fetch_models_evaluation(wandb_api, ENTITY, 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 = get_competitions(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_info in competitions: competition = competition_info[0] if not competition_summaries[competition].empty: # get only competition sumaries that heppen after latest evaluation time evaluation_time = get_latest_evaluation_time(competition_info[1]) evaluation_time = pd.Timestamp(evaluation_time) filtered_competition_summaries = competition_summaries[competition][competition_summaries[competition]["Created At"] > evaluation_time] # get all winning hotkeys and number of wins winning_hotkeys = filtered_competition_summaries["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("

Competitions

", 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]) headers = ["Index", "Competition Name", "Date", "Miner hotkey"] for col, header in zip(cols, headers): col.write(header) for index, competition_info in enumerate(competitions, start=1): competition = competition_info[0] leader_info = st.session_state.get("leader_info", {}).get(competition, {}) cols = st.columns([1, 3, 2, 2, 3]) 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 st.subheader(f"Competition: {competition_name}") # Add search bar for miner hotkey miner_hotkey_search = st.text_input("Search for Miner Hotkey", "") 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()