app.py

"""
Streamlit dashboard for the Dippy Speech Subnet Leaderboard
"""
import requests
import streamlit as st
import pandas as pd

st.set_page_config(layout="wide")

import pandas as pd
import numpy as np
REMOTE_LEADERBOARD_URL = "https://sn58.dippy-bittensor-subnet.com/minerboard"
def iswin(score_i, score_j, block_i, block_j):
    MAX_PENALTY = 0.03  # Adjust this value as needed
    penalty = MAX_PENALTY
    score_i = (1 - penalty) * score_i if block_i > block_j else score_i
    score_j = (1 - penalty) * score_j if block_j > block_i else score_j
    return score_i > score_j

def calculate_win_rate(df):
    n = len(df)
    win_counts = np.zeros(n)
    
    for i in range(n):
        for j in range(n):
            if i != j:
                if iswin(df.loc[i, 'total_score'], df.loc[j, 'total_score'],
                         df.loc[i, 'block'], df.loc[j, 'block']) and df.loc[i, 'real_status'] == 'COMPLETED':
                    win_counts[i] += 1
    
    return win_counts / (n - 1)  # Divide by (n-1) as each row isn't compared with itself



def leaderboard_dashboard():
    # load the logo from image.txt file as base64
    with open("image.txt", "r") as f:
        image = f.read()

    st.markdown(
        f"""
        <div style="text-align: center;">
            <img src="data:image/png;base64,{image}" alt="Dippy Roleplay Logo" width="600" height="300" style="margin-bottom: 2rem;">
            <h1 style="margin-top: 0;">SN58-Dippy-Speech Leaderboard</h1>
            <div style="font-size: 18px;">This is the leaderboard for the Dippy voice validation API hosted by SN58.</div>
        </div>
        """,
        unsafe_allow_html=True,
    )
    
    
    # Add emojis based on the status
    status_emojis = {
        'COMPLETED': '✅COMPLETED',
        'FAILED': '❌FAILED',
        'QUEUED': '🕒QUEUED',
        'RUNNING': '🏃RUNNING'
    }

    
    # Get the minerboard data from the API
    response = requests.get(REMOTE_LEADERBOARD_URL)
    if response.status_code != 200:
        st.error("Failed to fetch minerboard data.")
        return

    # Parse the response JSON data
    minerboard_data = response.json()
    if len(minerboard_data) < 1:
        st.markdown(
        f"""
        <div style="text-align: center;">
            <h2 style="margin-top: 0;">In progress!</h2>
        </div>
        """,
        unsafe_allow_html=True,
    )
        return

    # Convert the data to a DataFrame
    minerboard = pd.DataFrame(minerboard_data)
    
    

    minerboard['status'] = minerboard['status'].map(lambda status: status_emojis.get(status, status))
    # Sort the minerboard_winrate by the total_score column
    minerboard = minerboard.sort_values(by='total_score', ascending=False, ignore_index=True)

    # front_order = ['repo_name','uid', 'hotkey', 'total_score', 'status', 'hash']
    front_order = ['uid', 'repo_name', 'repo_namespace', 'hotkey', 'total_score', 'status', 'hash']

    # move status column to the front
    column_order = front_order + [column for column in minerboard.columns if column not in front_order]

    minerboard = minerboard[column_order]
    

    minerboard_winrate = pd.DataFrame(minerboard_data)

    minerboard_winrate['real_status'] = minerboard_winrate['status']
    
    minerboard_winrate.loc[minerboard_winrate['real_status'] == 'FAILED', 'total_score'] = 0
    

    minerboard_winrate['status'] = minerboard_winrate['status'].map(lambda status: status_emojis.get(status, status))

    

    minerboard_winrate['win_rate'] = calculate_win_rate(minerboard_winrate)

    minerboard_winrate = minerboard_winrate.sort_values(by='win_rate', ascending=False, ignore_index=True)
            
    # column_order = ['uid', 'win_rate', 'hotkey', 'hash', 'total_score', 'block', 'status']

    column_order = ['uid', 'repo_name', 'repo_namespace', 'win_rate', 'hotkey', 'hash', 'total_score', 'block', 'status']


    # Create a new DataFrame with only the specified columns
    minerboard_winrate = minerboard_winrate[column_order]
    st.header("Leaderboard by Win Rate ")
    st.dataframe(minerboard_winrate, hide_index=True)
    with st.expander("See detailed calculation method"):
        st.write("The win rate is calculated by comparing each miner against every other miner. Note that this board is only an approximation as queued miners have a score of 0, validators are omitted, etc.")
        st.code("""
    Example of calculating a win:
    def iswin(score_i, score_j, block_i, block_j):
        penalty = 0.03
        score_i = (1 - penalty) * score_i if block_i > block_j else score_i
        score_j = (1 - penalty) * score_j if block_j > block_i else score_j
        return score_i > score_j
    """)
    st.markdown("---")
    st.header("Minerboard")
    st.dataframe(minerboard, hide_index=True)
    st.markdown("---")


if __name__ == '__main__':
    leaderboard_dashboard()