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
import aiohttp

FONT = """<link href="https://fonts.googleapis.com/css2?family=Roboto+Mono&display=swap" rel="stylesheet">"""
IMAGE = """<p align="center" style="font-size: 72px;">𐌔𐌙𐌍𐌕𐋅𐌉𐌀</p>"""
HEADER = """
<h2 align="center" class="typewriter">Welcome to the Synthia Commune Leaderboard!</h2>
<p align="center">This leaderboard showcases the top-performing miners in the Synthia Commune Subnet. The models are ranked based on their daily rewards.</p>
<p align="center">The Synthia subnet leverages Commune's incentives to create a permissionless mining market around distilling knowledge out of SOTA closed-source model APIs into a public dataset to accelerate the OpenSource AI space. Targeted models and strategy will adapt based on the current SOTA.</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 = 3
node_url = "wss://commune-api-node-2.communeai.net"

async def get_com_price(session: aiohttp.ClientSession) -> Any:
    retries = 5
    for i in range(retries):
        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}")
            await asyncio.sleep(retries)
    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)

    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())
    for uid in range(highest_uid + 1):
        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]
    return emission_dict, name_dict

async def get_leaderboard_data():
    async with aiohttp.ClientSession() as session:
        try:
            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}"))
            return leaderboard_data
        except Exception as e:
            print(f"Error fetching leaderboard data: {e}")
            return []

async def update_leaderboard_table():
    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
    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()

with gr.Blocks(theme=seafoam, analytics_enabled=True) as demo:
    gr.HTML(FONT)
    gr.HTML(IMAGE)
    gr.HTML(HEADER)
    gr.HTML(EVALUATION_HEADER)
    gr.HTML(EVALUATION_DETAILS)

    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.components.Dataframe(
        headers=["Rank 🏆", "UID", "Name", "Rewards / Day", "$USD Value"],
        datatype=["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()