used aiohttp for getting the http requests

app.py

@@ -1,51 +1,41 @@
 import gradio as gr
 import requests
 import asyncio
-from typing import Any
+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 typing import Iterable
 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"
 
-
-def get_com_price() -> Any:
+async def get_com_price(session: aiohttp.ClientSession) -> Any:
     retries = 5
-    for i in range(0, retries):
+    for i in range(retries):
         try:
-            price = float(
-                requests.get(
-                    "https://api.mexc.com/api/v3/avgPrice?symbol=COMAIUSDT"
-                ).json()["price"]
-            )
-            print(f"Fetched COM price: {price}")
-            return price
+            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}")
-            if i == retries - 1:
-                raise
-            _ = asyncio.sleep(retries)
+            await asyncio.sleep(retries)
     raise RuntimeError("Failed to fetch COM price")
 
-
-def make_query(client: CommuneClient) -> tuple[dict[int, int], dict[int, str]]:
-    request_dict: dict[Any, Any] = {
+async def make_query(client: CommuneClient) -> tuple[dict[int, int], dict[int, str]]:
+    request_dict = {
         "SubspaceModule": [
             ("Name", [netuid]),
             ("Emission", []),
@@ -53,8 +43,8 @@ def make_query(client: CommuneClient) -> tuple[dict[int, int], dict[int, str]]:
             ("Dividends", []),
         ],
     }
-    emission_dict: dict[int, int] = {}
-    name_dict: dict[int, str] = {}
+    emission_dict = {}
+    name_dict = {}
     result = client.query_batch_map(request_dict)
 
     emission = result["Emission"]
@@ -67,63 +57,47 @@ def make_query(client: CommuneClient) -> tuple[dict[int, int], dict[int, str]]:
     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():
-    try:
-        com_price = get_com_price()
-        blocks_in_day = 10_800
-        client = CommuneClient(node_url)
-        emission_dict, name_dict = make_query(client)
-        print("got tha emission")
-        scores = {}
-        for uid, emi in emission_dict.items():
-            # we are doing ** 11 because emission are for modules have 2 extra units,
-            # (otherwise it would be more intuive to do ** 9, hence from_nano)
-            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 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()
-    # Convert tuples to lists
     leaderboard_data = [list(row) for row in leaderboard_data]
-    # Add emojis to the leaderboard data
     for row in leaderboard_data:
         row[0] = f"{row[0]} 🏆"
-
-    # Calculate the total USD value
     total_usd_value = sum(float(row[4][1:]) for row in leaderboard_data)
-
-    # Calculate rewards per week and per month
     rewards_per_week = total_usd_value * 7
     rewards_per_month = total_usd_value * 30
-
-    return leaderboard_data, f''' 
+    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>
@@ -131,7 +105,6 @@ async def update_leaderboard_table():
     </div>
     '''
 
-
 stone_gray = Color(
     name="stone_gray",
     c50="#f5f5f5",
@@ -147,11 +120,9 @@ stone_gray = Color(
     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,
@@ -186,12 +157,10 @@ class Seafoam(Base):
             button_shadow="*shadow_drop_lg",
             button_large_padding="32px",
             button_primary_background_fill_hover="*button_primary_background_fill",
-            # Add the following lines to set the button colors
-            button_primary_background_fill="#333333",  # Dark background color
-            button_primary_text_color="#ffffff",  # White text color
+            button_primary_background_fill="#333333",
+            button_primary_text_color="#ffffff",
         )
 
-
 seafoam = Seafoam()
 
 with gr.Blocks(theme=seafoam, analytics_enabled=True) as demo:
@@ -222,9 +191,8 @@ with gr.Blocks(theme=seafoam, analytics_enabled=True) as demo:
     refresh_button.click(fn=update_leaderboard_table, outputs=[
                          leaderboard_table, total_usd_value_html])
 
-    # Initial load of leaderboard data
     demo.load(update_leaderboard_table, inputs=None, outputs=[
               leaderboard_table, total_usd_value_html])
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()