import json
import os
from dataclasses import dataclass
from datetime import datetime
from zoneinfo import ZoneInfo
import bittensor as bt
import gradio as gr
import numpy as np
import plotly.graph_objects as go
import wandb
from substrateinterface import Keypair
from wandb.apis.public import Run
WANDB_RUN_PATH = os.environ["WANDB_RUN_PATH"]
SOURCE_VALIDATOR_UID = int(os.environ["SOURCE_VALIDATOR_UID"])
START_DATE = datetime(2024, 9, 17)
NET_UID = 39
REFRESH_RATE = 120
METAGRAPH_REFRESH_RATE = 43200 # 12 hours
GRAPH_HISTORY_DAYS = 30
MAX_GRAPH_ENTRIES = 10
demo = gr.Blocks(css=".typewriter {font-family: 'JMH Typewriter', sans-serif;}", fill_height=True, fill_width=True)
subtensor = bt.subtensor()
metagraph = bt.metagraph(netuid=NET_UID)
bt.logging.disable_logging()
runs: dict[int, list[Run]] = {}
validator_identities: dict[int, str] = {}
last_refresh: datetime = datetime.fromtimestamp(0, tz=ZoneInfo("America/New_York"))
last_metagraph_refresh: datetime = datetime.fromtimestamp(0, tz=ZoneInfo("America/New_York"))
@dataclass
class LeaderboardEntry:
uid: int
winner: bool
repository: str
score: float
similarity: float
hotkey: str
block: int
baseline_generation_time: float
generation_time: float
size: int
vram_used: float
watts_used: float
@dataclass
class GraphEntry:
dates: list[datetime]
baseline_generation_times: list[float]
generation_times: list[float]
similarities: list[float]
scores: list[float]
models: list[str]
best_time: float
def is_valid_run(run: Run):
required_config_keys = ["hotkey", "uid", "contest", "signature"]
for key in required_config_keys:
if key not in run.config:
return False
uid = run.config["uid"]
validator_hotkey = run.config["hotkey"]
contest_name = run.config["contest"]
signing_message = f"{uid}:{validator_hotkey}:{contest_name}"
try:
return Keypair(validator_hotkey).verify(signing_message, run.config["signature"])
except Exception:
return False
def calculate_score(baseline_generation_time: float, generation_time: float, similarity_score: float) -> float:
return (baseline_generation_time - generation_time) * similarity_score
def date_from_run(run: Run) -> datetime:
return datetime.strptime(run.created_at, "%Y-%m-%dT%H:%M:%SZ").astimezone(ZoneInfo("America/New_York"))
def get_graph_entries(runs: list[Run]) -> dict[int, GraphEntry]:
entries: dict[int, GraphEntry] = {}
for run in reversed(runs[:GRAPH_HISTORY_DAYS]):
date = date_from_run(run)
for summary_key, summary_value in run.summary.items():
if not summary_key.startswith("benchmarks"):
continue
for key, value in summary_value.items():
if "score" in value:
continue
uid = int(key)
baseline_generation_time = value["baseline_generation_time"]
generation_time = value["generation_time"]
similarity = min(1, value["similarity"])
score = calculate_score(baseline_generation_time, generation_time, similarity)
model = run.summary["submissions"][str(uid)]["repository"]
if uid not in entries:
entries[uid] = GraphEntry([date], [baseline_generation_time], [generation_time], [similarity], [score], [model], generation_time)
else:
if generation_time < entries[uid].best_time:
entries[uid].best_time = generation_time
data = entries[uid]
data.dates.append(date)
data.baseline_generation_times.append(baseline_generation_time)
data.generation_times.append(data.best_time)
data.similarities.append(similarity)
data.scores.append(score)
data.models.append(model)
entries = dict(sorted(entries.items(), key=lambda entry: entry[1].scores, reverse=True)[:MAX_GRAPH_ENTRIES])
return dict(sorted(entries.items(), key=lambda entry: entry[1].best_time))
def create_graph(validator_uid: int) -> go.Figure:
entries = get_graph_entries(runs[validator_uid])
fig = go.Figure()
for uid, data in entries.items():
fig.add_trace(go.Scatter(
x=data.dates,
y=data.generation_times,
customdata=np.stack((data.similarities, data.scores, data.models), axis=-1),
mode="lines+markers",
name=uid,
hovertemplate=(
"Date: %{x|%Y-%m-%d}
" +
"Generation Time: %{y}s
" +
"Similarity: %{customdata[0]}
" +
"Score: %{customdata[1]}
" +
"Model: %{customdata[2]}
"
),
))
date_range = max(entries.values(), key=lambda entry: len(entry.dates)).dates
average_baseline_generation_times = sum(entry.baseline_generation_times[0] for entry in entries.values()) / len(entries)
fig.add_trace(go.Scatter(
x=date_range,
y=[average_baseline_generation_times] * len(date_range),
line=dict(color="#ff0000", width=3),
mode="lines",
name="Baseline",
))
background_color = gr.themes.default.colors.slate.c800
fig.update_layout(
title="Generation Time Improvements",
yaxis_title="Generation Time (s)",
plot_bgcolor=background_color,
paper_bgcolor=background_color,
template="plotly_dark"
)
return fig
def get_run_validator_uid(run: Run) -> int:
json_config = json.loads(run.json_config)
uid = int(json_config["uid"]["value"])
return uid
def fetch_wandb_data():
wandb_api = wandb.Api()
wandb_runs = wandb_api.runs(
WANDB_RUN_PATH,
filters={"config.type": "validator", "created_at": {'$gt': str(START_DATE)}},
order="-created_at",
)
wandb_runs = [run for run in wandb_runs if "benchmarks" in run.summary]
global runs
runs.clear()
for run in wandb_runs:
if not is_valid_run(run):
continue
uid = get_run_validator_uid(run)
if not metagraph.validator_permit[uid]:
continue
if uid not in runs:
runs[uid] = []
runs[uid].append(run)
runs = dict(sorted(runs.items(), key=lambda item: item[0]))
def fetch_identities():
validator_identities.clear()
for uid in runs.keys():
identity = subtensor.substrate.query('SubtensorModule', 'Identities', [metagraph.coldkeys[uid]])
if identity != None:
validator_identities[uid] = identity.value["name"]
def get_validator_name(validator_uid: int) -> str:
if validator_uid in validator_identities:
return validator_identities[validator_uid]
else:
return metagraph.hotkeys[validator_uid]
def try_refresh():
global last_refresh
global last_metagraph_refresh
now = datetime.now(tz=ZoneInfo("America/New_York"))
if (now - last_refresh).total_seconds() > REFRESH_RATE:
print(f"Refreshing Leaderboard at {now.strftime('%Y-%m-%d %H:%M:%S')}")
last_refresh = now
fetch_wandb_data()
if (now - last_metagraph_refresh).total_seconds() > METAGRAPH_REFRESH_RATE:
metagraph.sync(subtensor=subtensor)
fetch_identities()
last_metagraph_refresh = now
last_refresh = now
def create_leaderboard(validator_uid: int) -> gr.Dataframe:
try_refresh()
entries: dict[int, LeaderboardEntry] = {}
for run in runs[validator_uid]:
has_data = False
for summary_key, summary_value in run.summary.items():
if not summary_key == "benchmarks":
continue
for key, value in summary_value.items():
has_data = True
uid = int(key)
generation_time = value["generation_time"]
baseline_generation_time = value["baseline_generation_time"]
similarity = min(1, value["similarity"])
entries[uid] = LeaderboardEntry(
uid=uid,
winner="winner" in value,
repository=run.summary["submissions"][str(uid)]["repository"],
score=calculate_score(baseline_generation_time, generation_time, similarity),
similarity=similarity,
baseline_generation_time=baseline_generation_time,
generation_time=generation_time,
size=value["size"],
vram_used=value["vram_used"],
watts_used=value["watts_used"],
hotkey=value["hotkey"],
block=run.summary["submissions"][str(uid)]["block"],
)
if has_data:
break
sorted_entries = [(
entry.uid,
f"{entry.winner}",
entry.repository,
round(entry.score, 3),
f"{entry.generation_time:.3f}s",
f"{entry.similarity:.3f}",
f"{entry.size / 1_000_000_000:.3f}GB",
f"{entry.vram_used / 1_000_000_000:.3f}GB",
f"{entry.watts_used:.3f}W",
entry.hotkey,
entry.block,
) for entry in sorted(entries.values(), key=lambda entry: (entry.winner, entry.score), reverse=True)]
return gr.Dataframe(
sorted_entries,
headers=["Uid", "Winner", "Model", "Score", "Gen Time", "Similarity", "Size", "VRAM Usage", "Power Usage", "Hotkey", "Block"],
datatype=["number", "markdown", "markdown", "number", "markdown", "number", "markdown", "markdown", "markdown", "markdown", "number"],
label=f"Last updated: {last_refresh.strftime('%Y-%m-%d %I:%M:%S %p')} EST",
interactive=False,
)
def create_dropdown() -> gr.Dropdown:
choices: list[tuple[str, int]] = []
for uid, run in runs.items():
if run[0].state != "running":
continue
benchmarks = dict(run[0].summary.get("benchmarks", {}))
finished = any("winner" in value for value in benchmarks.values())
progress_text = "Finished" if finished else "In Progress"
choices.append((f"{uid} - {get_validator_name(uid)} ({progress_text})", uid))
return gr.Dropdown(
choices,
value=SOURCE_VALIDATOR_UID,
interactive=True,
label="Source Validator"
)
def main():
try_refresh()
with demo:
gr.Image(
"cover.png",
show_label=False,
show_download_button=False,
interactive=False,
show_fullscreen_button=False,
show_share_button=False,
container=False,
)
gr.Markdown(
"""