add runtime apis to reg

bittensor_cli/src/__init__.py

@@ -278,10 +278,31 @@ class WalletValidationTypes(Enum):
                     "params": [],
                     "type": "Vec<u8>",
                 },
+                "get_subnet_info_v2": {
+                    "params": [
+                        {
+                            "name": "netuid",
+                            "type": "u16",
+                        },
+                    ],
+                    "type": "Vec<u8>",
+                },
+                "get_subnets_info_v2": {
+                    "params": [],
+                    "type": "Vec<u8>",
+                },
                 "get_all_dynamic_info": {
                     "params": [],
                     "type": "Vec<u8>",
                 },
+                "get_dynamic_info": {
+                    "params": [{"name": "netuid", "type": "u16"}],
+                    "type": "Vec<u8>",
+                },
+                "get_subnet_state": {
+                    "params": [{"name": "netuid", "type": "u16"}],
+                    "type": "Vec<u8>",
+                },
             }
         },
         "SubnetRegistrationRuntimeApi": {

bittensor_cli/src/bittensor/async_substrate_interface.py

@@ -4,7 +4,7 @@
 from collections import defaultdict
 from dataclasses import dataclass
 from hashlib import blake2b
-from typing import Optional, Any, Union, Callable, Awaitable, cast
+from typing import Optional, Any, Union, Callable, Awaitable, cast, Iterable
 
 from bt_decode import PortableRegistry, decode as decode_by_type_string, MetadataV15
 from async_property import async_property
@@ -1792,7 +1792,7 @@ async def query_multiple(
         runtime = await self.init_runtime(block_hash=block_hash)
         preprocessed: tuple[Preprocessed] = await asyncio.gather(
             *[
-                self._preprocess([x], block_hash, storage_function, module)
+                self._preprocess([x] if not isinstance(x, Iterable) else list(x), block_hash, storage_function, module)
                 for x in params
             ]
         )

bittensor_cli/src/bittensor/extrinsics/registration.py

@@ -493,17 +493,23 @@ async def get_neuron_for_pubkey_and_subnet():
         )
         if uid is None:
             return NeuronInfo.get_null_neuron()
+
 
-        params = [netuid, uid]
-        json_body = await subtensor.substrate.rpc_request(
-            method="neuronInfo_getNeuron",
-            params=params,
+        hex_bytes_result = await subtensor.substrate.query_runtime_api(
+            runtime_api="NeuronInfoRuntimeApi",
+            method="get_neuron",
+            params=[netuid, uid],
         )
 
-        if not (result := json_body.get("result", None)):
+        if not (result := hex_bytes_result):
             return NeuronInfo.get_null_neuron()
+
+        if result.startswith("0x"):
+            bytes_result = bytes.fromhex(result[2:])
+        else:
+            bytes_result = bytes.fromhex(result)
 
-        return NeuronInfo.from_vec_u8(bytes(result))
+        return NeuronInfo.from_vec_u8(bytes_result)
 
     print_verbose("Checking subnet status")
     if not await subtensor.subnet_exists(netuid):

bittensor_cli/src/bittensor/subtensor_interface.py

@@ -431,14 +431,15 @@ async def get_total_stake_for_hotkey(
 
         :return: {address: Balance objects}
         """
-        results = await self.substrate.query_multiple(
-            params=[s for s in ss58_addresses],
+        netuids = await self.get_all_subnet_netuids(block_hash=block_hash)
+        results: dict[tuple[str, int], int] = await self.substrate.query_multiple(
+            params=[p for p in zip(ss58_addresses, netuids)],
             module="SubtensorModule",
-            storage_function="TotalHotkeyStake",
+            storage_function="TotalHotkeyAlpha",
             block_hash=block_hash,
             reuse_block_hash=reuse_block,
         )
-        return {k: Balance.from_rao(r or 0) for (k, r) in results.items()}
+        return {k[0]: Balance.from_rao(r or 0) for (k, r) in results.items()}
 
     async def current_take(
         self,
@@ -730,17 +731,23 @@ async def neuron_for_uid(
         """
         if uid is None:
             return NeuronInfo.get_null_neuron()
-
-        params = [netuid, uid, block_hash] if block_hash else [netuid, uid]
-        json_body = await self.substrate.rpc_request(
-            method="neuronInfo_getNeuron",
-            params=params,  # custom rpc method
+
+        hex_bytes_result = await self.query_runtime_api(
+            runtime_api="NeuronInfoRuntimeApi",
+            method="get_neuron",
+            params=[netuid, uid],
+            block_hash=block_hash,
         )
 
-        if not (result := json_body.get("result", None)):
+
+        if not (result := hex_bytes_result):
             return NeuronInfo.get_null_neuron()
 
-        bytes_result = bytes(result)
+        if result.startswith("0x"):
+            bytes_result = bytes.fromhex(result[2:])
+        else:
+            bytes_result = bytes.fromhex(result)
+
         return NeuronInfo.from_vec_u8(bytes_result)
 
     async def get_delegated(
@@ -769,15 +776,23 @@ async def get_delegated(
             else (self.substrate.last_block_hash if reuse_block else None)
         )
         encoded_coldkey = ss58_to_vec_u8(coldkey_ss58)
-        json_body = await self.substrate.rpc_request(
-            method="delegateInfo_getDelegated",
-            params=([block_hash, encoded_coldkey] if block_hash else [encoded_coldkey]),
+
+        hex_bytes_result = await self.query_runtime_api(
+            runtime_api="DelegateInfoRuntimeApi",
+            method="get_delegated",
+            params=[encoded_coldkey],
+            block_hash=block_hash,
         )
 
-        if not (result := json_body.get("result")):
+        if not (result := hex_bytes_result):
             return []
 
-        return DelegateInfo.delegated_list_from_vec_u8(bytes(result))
+        if result.startswith("0x"):
+            bytes_result = bytes.fromhex(result[2:])
+        else:
+            bytes_result = bytes.fromhex(result)
+
+        return DelegateInfo.delegated_list_from_vec_u8(bytes_result)
 
     async def query_identity(
         self,
@@ -1153,7 +1168,7 @@ async def get_delegates_by_netuid_light(
             A list of DelegateInfo objects detailing each delegate's characteristics.
 
         """
-
+        # TODO (Ben): doesn't exist
         params = [netuid] if not block_hash else [netuid, block_hash]
         json_body = await self.substrate.rpc_request(
             method="delegateInfo_getDelegatesLight",  # custom rpc method
@@ -1170,10 +1185,22 @@ async def get_delegates_by_netuid_light(
     async def get_subnet_dynamic_info(
         self, netuid: int, block_hash: Optional[str] = None
     ) -> "DynamicInfo":
-        json = await self.substrate.rpc_request(
-            method="subnetInfo_getDynamicInfo", params=[netuid, block_hash]
+        hex_bytes_result = await self.query_runtime_api(
+            runtime_api="SubnetInfoRuntimeApi",
+            method="get_dynamic_info",
+            params=[netuid],
+            block_hash=block_hash,
         )
-        subnets = DynamicInfo.from_vec_u8(json["result"])
+
+        if hex_bytes_result is None:
+            return None 
+
+        if hex_bytes_result.startswith("0x"):
+            bytes_result = bytes.fromhex(hex_bytes_result[2:])
+        else:
+            bytes_result = bytes.fromhex(hex_bytes_result)
+
+        subnets = DynamicInfo.from_vec_u8(bytes_result)
         return subnets
 
     async def get_stake_for_coldkey_and_hotkey_on_netuid(

bittensor_cli/src/commands/subnets.py

@@ -269,16 +269,22 @@ async def subnets_list(
 async def show(subtensor: "SubtensorInterface", netuid: int, prompt: bool = True):
     async def show_root():
         all_subnets = await subtensor.get_all_subnet_dynamic_info()
-        root_state: "SubnetState" = SubnetState.from_vec_u8(
-            (
-                await subtensor.substrate.rpc_request(
-                    method="subnetInfo_getSubnetState", params=[0, None]
-                )
-            )["result"]
+
+        hex_bytes_result = await subtensor.query_runtime_api(
+            runtime_api="SubnetInfoRuntimeApi",
+            method="get_subnet_state",
+            params=[0],
         )
-        if root_state is None:
+        if (bytes_result := hex_bytes_result) is None:
             err_console.print("The root subnet does not exist")
             return
+
+        if bytes_result.startswith("0x"):
+            bytes_result = bytes.fromhex(bytes_result[2:])
+
+        root_state: "SubnetState" = SubnetState.from_vec_u8(
+           bytes_result
+        )
         if len(root_state.hotkeys) == 0:
             err_console.print(
                 "The root-subnet is currently empty with 0 UIDs registered."
@@ -373,12 +379,20 @@ async def show_root():
 
     async def show_subnet(netuid_: int):
         subnet_info = await subtensor.get_subnet_dynamic_info(netuid_)
+        hex_bytes_result = await subtensor.query_runtime_api(
+            runtime_api="SubnetInfoRuntimeApi",
+            method="get_subnet_state",
+            params=[netuid_],
+        )
+        if (bytes_result := hex_bytes_result) is None:
+            err_console.print(f"Subnet {netuid_} does not exist")
+            return
+
+        if bytes_result.startswith("0x"):
+            bytes_result = bytes.fromhex(bytes_result[2:])
+
         subnet_state: "SubnetState" = SubnetState.from_vec_u8(
-            (
-                await subtensor.substrate.rpc_request(
-                    method="subnetInfo_getSubnetState", params=[netuid_, None]
-                )
-            )["result"]
+            bytes_result
         )
         if subnet_info is None:
             err_console.print(f"Subnet {netuid_} does not exist")
@@ -789,12 +803,21 @@ async def metagraph_cmd(
                 ),
                 subtensor.substrate.get_block_number(block_hash=block_hash),
             )
+
+            hex_bytes_result = await subtensor.query_runtime_api(
+                runtime_api="SubnetInfoRuntimeApi",
+                method="get_subnet_state",
+                params=[netuid],
+            )
+            if not (bytes_result := hex_bytes_result):
+                err_console.print(f"Subnet {netuid} does not exist")
+                return
+
+            if bytes_result.startswith("0x"):
+                bytes_result = bytes.fromhex(bytes_result[2:])
+
             subnet_state: "SubnetState" = SubnetState.from_vec_u8(
-                (
-                    await subtensor.substrate.rpc_request(
-                        method="subnetInfo_getSubnetState", params=[netuid, None]
-                    )
-                )["result"]
+                bytes_result
             )
 
         difficulty = int(difficulty_)

bittensor_cli/src/commands/wallets.py

@@ -1114,17 +1114,12 @@ async def _fetch_neuron_for_netuid(
     """
 
     async def neurons_lite_for_uid(uid: int) -> dict[Any, Any]:
-        call_definition = TYPE_REGISTRY["runtime_api"]["NeuronInfoRuntimeApi"][
-            "methods"
-        ]["get_neurons_lite"]
-        data = await subtensor.encode_params(
-            call_definition=call_definition, params=[uid]
-        )
         block_hash = subtensor.substrate.last_block_hash
-        hex_bytes_result = await subtensor.substrate.rpc_request(
-            method="state_call",
-            params=["NeuronInfoRuntimeApi_get_neurons_lite", data, block_hash],
-            reuse_block_hash=True,
+        hex_bytes_result = await subtensor.query_runtime_api(
+            runtime_api="NeuronInfoRuntimeApi",
+            method="get_neurons_lite",
+            params=[uid],
+            block_hash=block_hash,
         )
 
         return hex_bytes_result