Feat/ subnet mechanisms

bittensor_cli/src/__init__.py

@@ -700,6 +700,10 @@ class WalletValidationTypes(Enum):
         "GOVERNANCE": "Governance",
         "TAKE": "Delegate take configuration",
     },
+    "MECHANISMS": {
+        "CONFIG": "Mechanism Configuration",
+        "EMISSION": "Mechanism Emission",
+    },
     "SUBNETS": {
         "INFO": "Subnet Information",
         "CREATION": "Subnet Creation & Management",

bittensor_cli/src/bittensor/chain_data.py

@@ -13,6 +13,7 @@
     u16_normalized_float as u16tf,
     u64_normalized_float as u64tf,
     decode_account_id,
+    get_netuid_and_subuid_by_storage_index,
 )
 
 
@@ -1084,12 +1085,13 @@ class MetagraphInfo(InfoBase):
     alpha_dividends_per_hotkey: list[
         tuple[str, Balance]
     ]  # List of dividend payout in alpha via subnet.
+    subuid: int = 0
 
     @classmethod
     def _fix_decoded(cls, decoded: dict) -> "MetagraphInfo":
         """Returns a MetagraphInfo object from decoded chain data."""
         # Subnet index
-        _netuid = decoded["netuid"]
+        _netuid, _subuid = get_netuid_and_subuid_by_storage_index(decoded["netuid"])
 
         # Name and symbol
         decoded.update({"name": bytes(decoded.get("name")).decode()})
@@ -1102,6 +1104,7 @@ def _fix_decoded(cls, decoded: dict) -> "MetagraphInfo":
         return cls(
             # Subnet index
             netuid=_netuid,
+            subuid=_subuid,
             # Name and symbol
             name=decoded["name"],
             symbol=decoded["symbol"],

bittensor_cli/src/bittensor/subtensor_interface.py

@@ -1175,6 +1175,55 @@ async def get_subnet_hyperparameters(
 
         return SubnetHyperparameters.from_any(result)
 
+    async def get_subnet_mechanisms(
+        self, netuid: int, block_hash: Optional[str] = None
+    ) -> int:
+        """Return the number of mechanisms that belong to the provided subnet."""
+
+        result = await self.query(
+            module="SubtensorModule",
+            storage_function="MechanismCountCurrent",
+            params=[netuid],
+            block_hash=block_hash,
+        )
+
+        if result is None:
+            return 0
+        return int(result)
+
+    async def get_all_subnet_mechanisms(
+        self, block_hash: Optional[str] = None
+    ) -> dict[int, int]:
+        """Return mechanism counts for every subnet with a recorded value."""
+
+        results = await self.substrate.query_map(
+            module="SubtensorModule",
+            storage_function="MechanismCountCurrent",
+            params=[],
+            block_hash=block_hash,
+        )
+        res = {}
+        async for netuid, count in results:
+            res[int(netuid)] = int(count.value)
+        return res
+
+    async def get_mechanism_emission_split(
+        self, netuid: int, block_hash: Optional[str] = None
+    ) -> list[int]:
+        """Return the emission split configured for the provided subnet."""
+
+        result = await self.query(
+            module="SubtensorModule",
+            storage_function="MechanismEmissionSplit",
+            params=[netuid],
+            block_hash=block_hash,
+        )
+
+        if not result:
+            return []
+
+        return [int(value) for value in result]
+
     async def burn_cost(self, block_hash: Optional[str] = None) -> Optional[Balance]:
         result = await self.query_runtime_api(
             runtime_api="SubnetRegistrationRuntimeApi",
@@ -1300,37 +1349,51 @@ async def get_stake_for_coldkey_and_hotkey_on_netuid(
         else:
             return Balance.from_rao(fixed_to_float(_result)).set_unit(int(netuid))
 
+    async def get_mechagraph_info(
+        self, netuid: int, mech_id: int, block_hash: Optional[str] = None
+    ) -> Optional[MetagraphInfo]:
+        """
+        Returns the metagraph info for a given subnet and mechanism id.
+        And yes, it is indeed 'mecha'graph
+        """
+        query = await self.query_runtime_api(
+            runtime_api="SubnetInfoRuntimeApi",
+            method="get_mechagraph",
+            params=[netuid, mech_id],
+            block_hash=block_hash,
+        )
+
+        if query is None:
+            return None
+
+        return MetagraphInfo.from_any(query)
+
     async def get_metagraph_info(
         self, netuid: int, block_hash: Optional[str] = None
     ) -> Optional[MetagraphInfo]:
-        hex_bytes_result = await self.query_runtime_api(
+        query = await self.query_runtime_api(
             runtime_api="SubnetInfoRuntimeApi",
             method="get_metagraph",
             params=[netuid],
             block_hash=block_hash,
         )
 
-        if hex_bytes_result is None:
+        if query is None:
             return None
 
-        try:
-            bytes_result = bytes.fromhex(hex_bytes_result[2:])
-        except ValueError:
-            bytes_result = bytes.fromhex(hex_bytes_result)
-
-        return MetagraphInfo.from_any(bytes_result)
+        return MetagraphInfo.from_any(query)
 
     async def get_all_metagraphs_info(
         self, block_hash: Optional[str] = None
     ) -> list[MetagraphInfo]:
-        hex_bytes_result = await self.query_runtime_api(
+        query = await self.query_runtime_api(
             runtime_api="SubnetInfoRuntimeApi",
             method="get_all_metagraphs",
             params=[],
             block_hash=block_hash,
         )
 
-        return MetagraphInfo.list_from_any(hex_bytes_result)
+        return MetagraphInfo.list_from_any(query)
 
     async def multi_get_stake_for_coldkey_and_hotkey_on_netuid(
         self,

bittensor_cli/src/bittensor/utils.py

@@ -35,6 +35,7 @@
 
 BT_DOCS_LINK = "https://docs.learnbittensor.org"
 
+GLOBAL_MAX_SUBNET_COUNT = 4096
 
 console = Console()
 json_console = Console()
@@ -1466,6 +1467,26 @@ def get_hotkey_pub_ss58(wallet: Wallet) -> str:
         return wallet.hotkey.ss58_address
 
 
+def get_netuid_and_subuid_by_storage_index(storage_index: int) -> tuple[int, int]:
+    """Returns the netuid and subuid from the storage index.
+
+    Chain APIs (e.g., SubMetagraph response) returns netuid which is storage index that encodes both the netuid and
+    subuid. This function reverses the encoding to extract these components.
+
+    Parameters:
+        storage_index: The storage index of the subnet.
+
+    Returns:
+        tuple[int, int]:
+            - netuid subnet identifier.
+            - subuid identifier.
+    """
+    return (
+        storage_index % GLOBAL_MAX_SUBNET_COUNT,
+        storage_index // GLOBAL_MAX_SUBNET_COUNT,
+    )
+
+
 async def print_extrinsic_id(
     extrinsic_receipt: Optional[AsyncExtrinsicReceipt],
 ) -> None: