feat: implement subnet metagraphn endpoint

services/dashboard/metagraph_api_server.py

@@ -0,0 +1,69 @@
+import bittensor as bt
+import uvicorn
+from fastapi import FastAPI
+from typing import List, Dict, Any
+from dataclasses import dataclass, field
+from vidaio_subnet_core import CONFIG
+
+config = CONFIG.bandwidth
+threshold = config.min_stake
+
+@dataclass
+class MinerInfo:
+    miners: List[Dict[str, Any]] = field(default_factory=list)
+    
+    def append(self, miner_uid, miner_hotkey, trust, incentive, emission, daily_reward):
+        self.miners.append({
+            "miner_uid": int(miner_uid),
+            "miner_hotkey": miner_hotkey,
+            "trust": float(trust),
+            "incentive": float(incentive),
+            "emission": float(emission),
+            "daily_reward": float(daily_reward)
+        })
+    
+    def to_dict(self):
+        return {
+            "miners": self.miners
+        }
+
+app = FastAPI()
+
+def get_filtered_miners(threshold=threshold):
+    miner_info = MinerInfo()
+    subtensor = bt.subtensor()
+    metagraph = subtensor.metagraph(netuid=85)
+    
+    incentives = metagraph.I
+    trusts = metagraph.T
+    emissions = metagraph.E
+    hotkeys = metagraph.hotkeys
+    stakes = metagraph.S
+    uids = metagraph.uids
+    daily_rewards = emissions * 20
+    
+    for uid, incentive, trust, emission, hotkey, stake, daily_reward in zip(
+        uids, incentives, trusts, emissions, hotkeys, stakes, daily_rewards
+    ):
+        if stake < threshold:
+            miner_info.append(
+                miner_uid=uid,
+                miner_hotkey=hotkey,
+                trust=trust,
+                incentive=incentive,
+                emission=emission,
+                daily_reward=daily_reward
+            )
+    
+    return miner_info
+
+@app.get("/miner_info")
+def get_miner_info():
+
+    miner_info = get_filtered_miners()
+    return miner_info.to_dict()
+
+if __name__ == "__main__":
+    host = config.host
+    port = config.port
+    uvicorn.run(app, host=host, port=port)

services/dashboard/model.py

@@ -1,6 +1,6 @@
-from typing import List
 from dataclasses import dataclass
-from datetime import datetime
+from dataclasses import dataclass
+from typing import List, Dict, Any
 
 @dataclass
 class MinerData:
@@ -59,3 +59,26 @@ class MinerData:
             "task_url": self.task_urls,
             "processed_url": self.processed_urls
         }
+
+@dataclass
+class MinerInfo:
+    """
+    Class representing information about a miner.
+    """
+    miner_uid: int
+    miner_hotkey: str
+    trust: float
+    incentive: float
+    emission: float
+    daily_reward: float
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert the MinerInfo instance to a dictionary"""
+        return {
+            "miner_uid": self.miner_uid,
+            "miner_hotkey": self.miner_hotkey,
+            "trust": self.trust,
+            "incentive": self.incentive,
+            "emission": self.emission,
+            "daily_reward": self.daily_reward
+        }

services/dashboard/server.py

@@ -80,7 +80,7 @@ def main():
             "0xcdef1234567890abcdef1234567890abcdef1234"
         ],
         timestamp=datetime.now().isoformat(),
-        request_type="image_processing",
+        request_type="test dashboard api",
         vmaf_scores=[95.2, 87.6, 92.1],
         pieapp_scores=[88.7, 79.3, 85.9],
         final_scores=[91.95, 83.45, 89.0],

vidaio_subnet_core/configs/dashboard.py

@@ -8,4 +8,6 @@ load_dotenv()
 class DashboardConfig(BaseModel):
     endpoint: str = Field(default=os.getenv("DASHBOARD_ENDPOINT", "localhost:9001"))
     timeout: int = Field(default=10)
-    max_retries: int = Field(default = 3)
+    max_retries: int = Field(default = 3)
+    host: str = Field(default="0.0.0.0")
+    port: int = Field(default=29117)