app.py

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_timexd = get_latest_evaluation_time(competition_info[1])
 print(type(evaluation_timexd))
 evaluation_time = pd.Timestamp(evaluation_timexd)
 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("<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])
 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()