app.py

import gradio as gr
import requests
import asyncio
from typing import Any, Iterable
from gradio.themes.base import Base
from gradio.themes.utils import colors, fonts, sizes
from gradio.themes.utils.colors import Color
from communex.client import CommuneClient
from communex.misc import get_map_modules
from typing import Any, cast
import aiohttp
import time

FONT = """<link href="https://fonts.googleapis.com/css2?family=Roboto+Mono&display=swap" rel="stylesheet">"""
HEADER = """
<h2 align="center" class="typewriter">Welcome to the Nya - CommuneAI Compute Subnet Leaderboard!</h2>
<p align="center">This leaderboard showcases the top-performing miners in the Nya - CommuneAI Compute Subnet. The models are ranked based on their daily rewards.</p>
<p align="center">Nya is a subnet dedicated to computing. Nya means purpose in Swahili. Training models like ChatGPT and Llama costs approximately $300 million, requiring billion-dollar datacenters. This expense is a significant barrier for many researchers. Inspired by initiatives like Folding at Home and Learning at Home, we envision a decentralized network that democratizes AI research and development through a shared mission and purpose. During the global pandemic, Folding at Home saw such an influx of participants that its computing power exceeded the world's largest supercomputers. Recognizing this potential, we are developing technology to enable the decentralized training of large models, benefiting all participants. We believe that the network's value will grow as more participants join and as we continue to release new models and technologies.</p>
<p align="center">In this subnet, miners are rewarded for completing assigned tasks, which involve machine learning computations instead of solving cryptographic puzzles like in Bitcoin mining. The results are validated to ensure correctness and maintain network integrity. Rewards are distributed based on each miner's share of the total computation performed, meaning the more computation a miner completes, the more rewards they receive.</p>
"""
EVALUATION_HEADER = """<h3 align="center">Evaluation Details</h3>"""
EVALUATION_DETAILS = """<p align="center"><b>Name</b> represents the model name. <b>Rewards / Day</b> indicates the expected daily rewards for each model in <b>$COMAI</b>. <b>UID</b> is the unique identifier of the miner. <b>$USD Value</b> is the estimated dollar value of the daily rewards.</p>"""

netuid = 8
node_url = "wss://commune-api-node-2.communeai.net"

def get_validator_uids(client, netuid):
    modules = cast(dict[str, Any], get_map_modules(client, netuid=netuid))
    modules = [value for _, value in modules.items()]
    validator_uids = []
    for module in modules:
        if not (module["incentive"] == module["dividends"] == 0 or module["incentive"] > module["dividends"]):
            validator_uids.append(int(module['uid']))
    return validator_uids

async def get_com_price(session: aiohttp.ClientSession) -> float:
    try:
        async with session.get("https://api.mexc.com/api/v3/avgPrice?symbol=COMAIUSDT") as response:
            response.raise_for_status()
            price = float((await response.json())["price"])
            print(f"Fetched COM price: {price}")
            return price
    except Exception as e:
        print(f"Error fetching COM price: {e}")
        raise RuntimeError("Failed to fetch COM price")

async def make_query(client: CommuneClient) -> tuple[dict[int, int], dict[int, str]]:
    request_dict = {
        "SubspaceModule": [
            ("Name", [netuid]),
            ("Emission", []),
            ("Incentive", []),
            ("Dividends", []),
        ],
    }
    emission_dict = {}
    name_dict = {}
    result = client.query_batch_map(request_dict)

    print("Query result:", result)

    emission = result["Emission"]
    netuid_emission = emission[netuid]
    incentive = result["Incentive"]
    netuid_incentive = incentive[netuid]
    dividends = result["Dividends"]
    netuid_dividends = dividends[netuid]

    names = result["Name"]
    highest_uid = max(names.keys())
    validator_uids = get_validator_uids(client, netuid)
    for uid in range(highest_uid + 1):
        if uid in validator_uids:
            continue
        emission = netuid_emission[uid]
        if emission != 0:
            incentive = netuid_incentive[uid]
            dividends = netuid_dividends[uid]
            if incentive > 0:
                emission_dict[uid] = netuid_emission[uid]
                name_dict[uid] = names[uid]
    print("Emission dict:", emission_dict)
    print("Name dict:", name_dict)
    return emission_dict, name_dict

async def get_leaderboard_data():
    async with aiohttp.ClientSession() as session:
        com_price = await get_com_price(session)
        blocks_in_day = 10_800
        client = CommuneClient(node_url)
        emission_dict, name_dict = await make_query(client)
        print("Got the emission")
        scores = {}
        for uid, emi in emission_dict.items():
            scores[uid] = (emi / 10**11) * blocks_in_day
        sorted_scores = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        leaderboard_data = []
        for rank, (uid, score) in enumerate(sorted_scores, start=1):
            name = name_dict[uid]
            units = score
            usd_value = score * com_price
            leaderboard_data.append((rank, uid, name, units, f"${usd_value:.2f}"))
        print("Leaderboard data:", leaderboard_data)
        return leaderboard_data

async def update_leaderboard_table():
    start_time = time.time()
    leaderboard_data = await get_leaderboard_data()
    leaderboard_data = [list(row) for row in leaderboard_data]
    for row in leaderboard_data:
        row[0] = f"{row[0]} 🏆"
    total_usd_value = sum(float(row[4][1:]) for row in leaderboard_data)
    rewards_per_week = total_usd_value * 7
    rewards_per_month = total_usd_value * 30
    print(f"Updated leaderboard in {time.time() - start_time:.2f} seconds")
    return leaderboard_data, f'''
    <div style="display: flex; justify-content: space-between; align-items: center; font-size: 16px; font-weight: bold;">
        <div>𝙏𝙤𝙩𝙖𝙡 $ 𝙈𝙞𝙣𝙞𝙣𝙜 𝙍𝙚𝙬𝙖𝙧𝙙𝙨 𝙋𝙚𝙧 𝙒𝙚𝙚𝙠: ${rewards_per_week:,.0f}</div>
        <div>𝙏𝙤𝙩𝙖𝙡 $ 𝙈𝙞𝙣𝙞𝙣𝙜 𝙍𝙚𝙬𝙖𝙧𝙙𝙨 𝙋𝙚𝙧 𝘿𝙖𝙮: ${total_usd_value:,.0f}</div>
        <div>𝙏𝙤𝙩𝙖𝙡 $ 𝙈𝙞𝙣𝙞𝙣𝙜 𝙍𝙚𝙬𝙖𝙧𝙙𝙨 𝙋𝙚𝙧 𝙈𝙤𝙣𝙩𝙝: ${rewards_per_month:,.0f}</div>
    </div>
    '''

stone_gray = Color(
    name="stone_gray",
    c50="#f5f5f5",
    c100="#e9e9e9",
    c200="#d9d9d9",
    c300="#c4c4c4",
    c400="#a6a6a6",
    c500="#8f8f8f",
    c600="#737373",
    c700="#595959",
    c800="#464646",
    c900="#262626",
    c950="#1a1a1a",
)

class Seafoam(Base):
    def __init__(
        self,
        primary_hue: colors.Color | str = stone_gray,
        secondary_hue: colors.Color | str = stone_gray,
        neutral_hue: colors.Color | str = stone_gray,
        spacing_size: sizes.Size | str = sizes.spacing_md,
        radius_size: sizes.Size | str = sizes.radius_md,
        text_size: sizes.Size | str = sizes.text_lg,
        font: fonts.Font | str | Iterable[fonts.Font | str] = (
            fonts.GoogleFont("Quicksand"),
            "ui-sans-serif",
            "sans-serif",
        ),
        font_mono: fonts.Font | str | Iterable[fonts.Font | str] = (
            fonts.GoogleFont("IBM Plex Mono"),
            "ui-monospace",
            "monospace",
        ),
    ):
        super().__init__(
            primary_hue=primary_hue,
            secondary_hue=secondary_hue,
            neutral_hue=neutral_hue,
            spacing_size=spacing_size,
            radius_size=radius_size,
            text_size=text_size,
            font=font,
            font_mono=font_mono,
        )
        super().set(
            block_title_text_weight="600",
            block_border_width="3px",
            block_shadow="*shadow_drop_lg",
            button_shadow="*shadow_drop_lg",
            button_large_padding="32px",
            button_primary_background_fill_hover="*button_primary_background_fill",
            button_primary_background_fill="#333333",
            button_primary_text_color="#ffffff",
        )

seafoam = Seafoam()

custom_css = """
#custom-image {
    display: block;
    margin-left: auto;
    margin-right: auto;
    width: 50%;
}
"""

with gr.Blocks(theme=seafoam, analytics_enabled=True, css=custom_css) as demo:
    gr.HTML(FONT)
    gr.Image("comuneXnakamoto", elem_id="custom-image")
    gr.HTML(HEADER)
    gr.HTML(EVALUATION_HEADER)
    gr.HTML(EVALUATION_DETAILS)
    gr.HTML(LANGUAGES_SUPPORTED)


    total_usd_value_html = gr.HTML(
        elem_id="total-usd-value",
        value='''
            <div style="display: flex; justify-content: space-between; align-items: center; font-size: 16px; font-weight: bold;">
                <div>𝙏𝙤𝙩𝙖𝙡 $ 𝙈𝙞𝙣𝙞𝙣𝙜 𝙍𝙚𝙬𝙖𝙧𝙙𝙨 𝙋𝙚𝙧 𝙒𝙚𝙚𝙠: $0.00</div>
                <div>𝙏𝙤𝙩𝙖𝙡 $ 𝙈𝙞𝙣𝙞𝙣𝙜 𝙍𝙚𝙬𝙖𝙧𝙙𝙨 𝙋𝙚𝙧 𝘿𝙖𝙮: $0.00</div>
                <div>𝙏𝙤𝙩𝙖𝙡 $ 𝙈𝙞𝙣𝙞𝙣𝙜 𝙍𝙚𝙬𝙖𝙧𝙙𝙨 𝙋𝙚𝙧 𝙈𝙤𝙣𝙩𝙝: $0.00</div>
            </div>
        '''
    )
    leaderboard_table = gr.Dataframe(
        headers=["Rank 🏆", "UID", "Name", "Rewards / Day", "$USD Value"],
        datatype=["str", "str", "str", "str", "str"],
        interactive=False,
        visible=True,
        elem_id="leaderboard-table",
    )
    refresh_button = gr.Button("Refresh Leaderboard")
    refresh_button.click(fn=update_leaderboard_table, outputs=[
                         leaderboard_table, total_usd_value_html])

    demo.load(update_leaderboard_table, inputs=None, outputs=[
              leaderboard_table, total_usd_value_html])

if __name__ == "__main__":
    demo.launch()