src/validator_states.py

from datetime import timedelta

import gradio as gr
import pandas as pd
from packaging import version

from chain_data import get_neurons
from wandb_data import get_current_runs, Run

AVERAGE_BENCHMARKING_TIME_WARNING_THRESHOLD = 180  # 3 minutes
ETA_WARNING_THRESHOLD = 43200  # 12 hours
UPDATED_WARNING_THRESHOLD = 1000
VTRUST_WARNING_THRESHOLD = 0.75


def get_latest_version(runs: list[Run]) -> str:
    latest_version = version.parse("0.0.0")
    for run in runs:
        current_version = version.parse(run.version)
        if current_version > latest_version:
            latest_version = current_version
    return str(latest_version)


def colorize(val, color: str) -> str:
    return f"<span style='color: {color}'>{val}</span>"


def create_validator_states() -> gr.Dataframe:
    data: list[list] = []
    runs = sorted(get_current_runs(), key=lambda run: run.uid)
    latest_version = get_latest_version(runs)
    for run in runs:
        if not run.hotkey in get_neurons():
            continue
        vtrust = get_neurons()[run.hotkey].validator_trust
        updated = get_neurons()[run.hotkey].updated
        data.append([
            run.uid,
            run.name,
            colorize(run.version, "springgreen" if run.version == latest_version else "red"),
            colorize(run.status.name(), run.status.color()),
            f"{min(run.total_submissions, len(run.submissions) + len(run.invalid_submissions))}/{run.total_submissions}",
            len(run.invalid_submissions),
            colorize(f"{timedelta(seconds=int(run.average_benchmarking_time))}", "orange" if run.average_benchmarking_time > AVERAGE_BENCHMARKING_TIME_WARNING_THRESHOLD else "springgreen" if run.average_benchmarking_time > 0 else "gray"),
            colorize(f"{timedelta(seconds=run.eta)}", "orange" if run.eta > ETA_WARNING_THRESHOLD else "springgreen" if run.eta > 0 else "gray"),
            run.step,
            colorize(f"{vtrust:.4f}", "springgreen" if vtrust > VTRUST_WARNING_THRESHOLD else "red"),
            colorize(updated, "springgreen" if updated < UPDATED_WARNING_THRESHOLD else "red"),
        ])

    return gr.Dataframe(
        pd.DataFrame(data, columns=["UID", "Name", "Version", "Status", "Tested", "Invalid", "Avg. Benchmark Time", "ETA", "Step", "VTrust", "Updated"]),
        datatype=["number", "markdown", "markdown", "markdown", "number", "number", "markdown", "markdown", "number", "markdown", "markdown"],
        interactive=False,
        max_height=800,
        row_count=1,
    )