Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import pandas as pd | |
| import wandb | |
| import time | |
| from datetime import datetime, timedelta | |
| import requests | |
| def get_competitions(CONFIG_URL): | |
| """ | |
| Get competition names and their evaluation times from the config URL. | |
| """ | |
| competitions = [] | |
| try: | |
| config = requests.get(CONFIG_URL).json() | |
| for competition in config: | |
| # Extract competition name and evaluation times | |
| competition_name = competition["competition_id"] | |
| evaluation_times = competition.get("evaluation_times", []) | |
| # Store competition name and evaluation times as a tuple | |
| competitions.append((competition_name, evaluation_times)) | |
| except Exception as e: | |
| print(f"Error loading competition data: {str(e)}") | |
| return competitions | |
| def get_latest_evaluation_time(evaluation_times): | |
| """ | |
| Get the latest UTC evaluation time as a pandas Timestamp (with dtype=datetime64[ns, UTC]). | |
| Args: | |
| evaluation_times (list): List of evaluation times as strings (in HH:MM format). | |
| Returns: | |
| pd.Timestamp: Latest evaluation time as a pandas Timestamp with timezone UTC. | |
| """ | |
| # Get the current UTC date and time | |
| current_utc_datetime = datetime.utcnow() | |
| current_utc_time = current_utc_datetime.time() | |
| # Convert evaluation times to datetime.time objects | |
| eval_times = [datetime.strptime(time, "%H:%M").time() for time in evaluation_times] | |
| # Sort the evaluation times to ensure they are in chronological order | |
| eval_times.sort() | |
| # Loop through the evaluation times in reverse to find the latest time that has passed | |
| for eval_time in reversed(eval_times): | |
| if current_utc_time >= eval_time: | |
| # If the time has passed today, return today's date with that time as pd.Timestamp | |
| latest_time = current_utc_datetime.replace(hour=eval_time.hour, minute=eval_time.minute, second=0, microsecond=0) | |
| return pd.Timestamp(latest_time, tz='UTC') | |
| # If none of the evaluation times have passed, return the latest one from the previous day | |
| yesterday = current_utc_datetime - timedelta(days=1) | |
| latest_eval_time = eval_times[-1] | |
| # Return the latest evaluation time from yesterday as pd.Timestamp | |
| latest_time = yesterday.replace(hour=latest_eval_time.hour, minute=latest_eval_time.minute, second=0, microsecond=0) | |
| return pd.Timestamp(latest_time, tz='UTC') | |
| def fetch_competition_summary(api, entity, project): | |
| data = [] | |
| # entity = projects[selected_project]["entity"] | |
| # project = projects[selected_project]["project"] | |
| runs = api.runs(f"{entity}/{project}") | |
| for run in runs: | |
| try: | |
| summary = run.summary | |
| if summary.get("validator_hotkey") and summary.get("winning_hotkey"): | |
| data.append({ | |
| "Created At": run.created_at, | |
| "Validator ID": summary.get("validator_hotkey"), | |
| "Winning Hotkey": summary.get("winning_hotkey"), | |
| "Run Time (s)": summary.get("run_time_s"), | |
| }) | |
| except Exception as e: | |
| st.write(f"Error processing run {run.id}: {str(e)}") | |
| df = pd.DataFrame(data) | |
| if not df.empty: | |
| df['Created At'] = pd.to_datetime(df['Created At'], utc=True) | |
| df = df.sort_values(by="Created At", ascending=False) | |
| return df | |
| def fetch_models_evaluation(api, entity, project): | |
| data = [] | |
| # entity = projects[selected_project]["entity"] | |
| # project = projects[selected_project]["project"] | |
| runs = api.runs(f"{entity}/{project}") | |
| for run in runs: | |
| try: | |
| summary = run.summary | |
| if summary.get("score") is not None: # Assuming runs with score are model evaluations | |
| model_link = summary.get("hf_model_link", "") | |
| if model_link: | |
| # Create clickable link | |
| model_link_html = f'<a href="{model_link}" target="_blank">Model Link</a>' | |
| else: | |
| model_link_html = "N/A" | |
| data.append({ | |
| "Created At": run.created_at, | |
| "Miner hotkey": summary.get("miner_hotkey", "N/A"), | |
| "F1-beta": summary.get("fbeta"), | |
| "Accuracy": summary.get("accuracy"), | |
| "Recall": summary.get("recall"), | |
| "Precision": summary.get("precision"), | |
| "Tested entries": summary.get("tested_entries"), | |
| "ROC AUC": summary.get("roc_auc"), | |
| "Confusion Matrix": summary.get("confusion_matrix"), | |
| "Model link": model_link_html, | |
| "Score": summary.get("score"), | |
| #TODO link to huggingface model | |
| }) | |
| except Exception as e: | |
| st.write(f"Error processing run {run.id}: {str(e)}") | |
| df = pd.DataFrame(data) | |
| if not df.empty: | |
| df['Created At'] = pd.to_datetime(df['Created At'], utc=True) | |
| df = df.sort_values(by="Created At", ascending=False) | |
| return df | |
| def highlight_score_column(s): | |
| """ | |
| Highlight the 'Score' column with a custom background color. | |
| """ | |
| return ['background-color: yellow' if s.name == 'Score' else '' for _ in s] | |