[Revolution] call runtime apis using state call

bittensor/__init__.py

@@ -111,8 +111,39 @@ def debug(on: bool = True):
 }
 
 # --- Type Registry ---
-__type_registry__ = {"types": {"Balance": "u64"}}  # Need to override default u128
-
+__type_registry__ = {
+    "types": {
+        "Balance": "u64",  # Need to override default u128
+    },
+    "runtime_api": {
+        "NeuronInfoRuntimeApi": {
+            "methods": {
+                "get_neuron_lite": {
+                    'params': [
+                        {
+                            "name": "netuid",
+                            "type": "u16",
+                        }, 
+                        {
+                            "name": "uid",
+                            "type": "u16",
+                        },
+                    ],
+                    "type": "Vec<u8>",
+                },
+                "get_neurons_lite": {
+                    'params': [
+                        {
+                            "name": "netuid",
+                            "type": "u16",
+                        },
+                    ],
+                    "type": "Vec<u8>",
+                },
+            }
+        },
+    }
+}
 
 from .errors import *
 

bittensor/chain_data.py

@@ -21,7 +21,7 @@
 from enum import Enum
 from dataclasses import dataclass
 from scalecodec.types import GenericCall
-from typing import List, Tuple, Dict, Optional, Any, TypedDict
+from typing import List, Tuple, Dict, Optional, Any, TypedDict, overload, Union
 from scalecodec.base import RuntimeConfiguration, ScaleBytes
 from scalecodec.type_registry import load_type_registry_preset
 from scalecodec.utils.ss58 import ss58_encode
@@ -225,15 +225,52 @@ class ChainDataType(Enum):
 U16_MAX = 65535
 U64_MAX = 18446744073709551615
 
+@overload
+def from_scale_encoding(
+    input: List[int],
+    type_name: ChainDataType,
+    is_vec: bool = False,
+    is_option: bool = False,
+) -> Optional[Dict]:
+    ...
 
+@overload
 def from_scale_encoding(
-    vec_u8: List[int],
+    input: bytes,
     type_name: ChainDataType,
     is_vec: bool = False,
     is_option: bool = False,
 ) -> Optional[Dict]:
-    as_bytes = bytes(vec_u8)
-    as_scale_bytes = ScaleBytes(as_bytes)
+    ...
+
+@overload
+def from_scale_encoding(
+    input: ScaleBytes,
+    type_name: ChainDataType,
+    is_vec: bool = False,
+    is_option: bool = False,
+) -> Optional[Dict]:
+    ...
+
+def from_scale_encoding(
+    input: Union[List[int], bytes, ScaleBytes],
+    type_name: ChainDataType,
+    is_vec: bool = False,
+    is_option: bool = False,
+) -> Optional[Dict]:
+    if isinstance(input, ScaleBytes):
+        as_scale_bytes = input
+    else:
+        if isinstance(input, list) and all([isinstance(i, int) for i in input]):
+            vec_u8 = input
+            as_bytes = bytes(vec_u8)
+        elif isinstance(input, bytes):
+            as_bytes = input
+        else:
+            raise TypeError("input must be a List[int], bytes, or ScaleBytes")
+
+        as_scale_bytes = ScaleBytes(as_bytes)
+
     rpc_runtime_config = RuntimeConfiguration()
     rpc_runtime_config.update_type_registry(load_type_registry_preset("legacy"))
     rpc_runtime_config.update_type_registry(custom_rpc_type_registry)

bittensor/subtensor.py

@@ -25,8 +25,10 @@
 
 from retry import retry
 from loguru import logger
-from typing import List, Dict, Union, Optional, Tuple
+from typing import List, Dict, Union, Optional, Tuple, TypedDict, Any
 from substrateinterface.base import QueryMapResult, SubstrateInterface
+from scalecodec.base import RuntimeConfiguration
+from scalecodec.type_registry import load_type_registry_preset
 
 # Local imports.
 from .chain_data import (
@@ -38,6 +40,7 @@
     AxonInfo,
     ProposalVoteData,
     ProposalCallData,
+    custom_rpc_type_registry,
 )
 from .errors import *
 from .extrinsics.network import register_subnetwork_extrinsic
@@ -69,6 +72,10 @@
 
 logger = logger.opt(colors=True)
 
+class ParamWithTypes(TypedDict):
+    name: str # Name of the parameter.
+    type: str # ScaleType string of the parameter.
+
 
 class subtensor:
     """Factory Class for bittensor.subtensor
@@ -1189,6 +1196,84 @@ def make_substrate_call_with_retry():
                 )
 
         return make_substrate_call_with_retry()
+
+    def state_call(
+        self,
+        method: str,
+        data: str,
+        block: Optional[int] = None,
+    ) -> Optional[object]:
+        @retry(delay=2, tries=3, backoff=2, max_delay=4)
+        def make_substrate_call_with_retry():
+            with self.substrate as substrate:
+                block_hash = None if block == None else substrate.get_block_hash(block)
+                params = [method, data]
+                if block_hash:
+                    params = params + [block_hash]
+                return substrate.rpc_request(
+                    method="state_call",
+                    params=params
+                )
+
+        return make_substrate_call_with_retry()
+
+    def query_runtime_api(
+        self,
+        runtime_api: str,
+        method: str,
+        params: Optional[List[ParamWithTypes]],
+        block: Optional[int] = None,
+    ) -> Optional[bytes]:
+        """
+        Returns a Scale Bytes type that should be decoded. 
+        """
+        call_definition = bittensor.__type_registry__['runtime_api'][runtime_api]['methods'][method]
+        json_result = self.state_call(
+            method=f"{runtime_api}_{method}",
+            data='0x' if params is None else self._encode_params(
+                call_definition=call_definition,
+                params=params
+            ),
+            block=block
+        )
+
+        if json_result is None:
+            return None
+
+        return_type = call_definition['type']
+
+        as_scale_bytes = scalecodec.ScaleBytes( json_result['result'] )
+
+        rpc_runtime_config = RuntimeConfiguration()
+        rpc_runtime_config.update_type_registry(load_type_registry_preset('legacy'))
+        rpc_runtime_config.update_type_registry(custom_rpc_type_registry)
+
+        obj = rpc_runtime_config.create_scale_object(return_type)
+
+        return obj.decode(as_scale_bytes)
+
+
+    def _encode_params(
+        self,
+        call_definition: List[ParamWithTypes],
+        params: Union[List[Any], Dict[str, str]],
+    ) -> str:
+        """
+        Returns a hex encoded string of the params using their types.
+        """
+        param_data = scalecodec.ScaleBytes(b'')
+
+        for i, param in enumerate(call_definition['params']):
+            scale_obj = self.substrate.create_scale_object(param['type'])
+            if type(params) is list:
+                param_data += scale_obj.encode(params[i])
+            else:
+                if param['name'] not in params:
+                    raise ValueError(f"Missing param {param['name']} in params dict.")
+
+                param_data += scale_obj.encode(params[param['name']])
+
+        return param_data.to_hex()
 
     #####################################
     #### Hyper parameter calls. ####
@@ -1858,26 +1943,23 @@ def neuron_for_uid_lite(
         """
         if uid == None:
             return NeuronInfoLite._null_neuron()
+
+        hex_bytes_result = self.query_runtime_api(
+            runtime_api="NeuronInfoRuntimeApi",
+            method="get_neuron_lite",
+            params={ 
+                "netuid": netuid,
+                "uid": uid,
+            },
+            block=block,
+        )
 
-        @retry(delay=2, tries=3, backoff=2, max_delay=4)
-        def make_substrate_call_with_retry():
-            with self.substrate as substrate:
-                block_hash = None if block == None else substrate.get_block_hash(block)
-                params = [netuid, uid]
-                if block_hash:
-                    params = params + [block_hash]
-                return substrate.rpc_request(
-                    method="neuronInfo_getNeuronLite",  # custom rpc method
-                    params=params,
-                )
-
-        json_body = make_substrate_call_with_retry()
-        result = json_body["result"]
-
-        if result in (None, []):
+        if hex_bytes_result == None:
             return NeuronInfoLite._null_neuron()
+
+        bytes_result = bytes.fromhex(hex_bytes_result[2:])
 
-        return NeuronInfoLite.from_vec_u8(result)
+        return NeuronInfoLite.from_vec_u8(bytes_result)
 
     def neurons_lite(
         self, netuid: int, block: Optional[int] = None
@@ -1892,26 +1974,19 @@ def neurons_lite(
             neuron (List[NeuronInfoLite]):
                 List of neuron lite metadata objects.
         """
+        hex_bytes_result = self.query_runtime_api(
+            runtime_api="NeuronInfoRuntimeApi",
+            method="get_neurons_lite",
+            params=[netuid],
+            block=block,
+        )
 
-        @retry(delay=2, tries=3, backoff=2, max_delay=4)
-        def make_substrate_call_with_retry():
-            with self.substrate as substrate:
-                block_hash = None if block == None else substrate.get_block_hash(block)
-                params = [netuid]
-                if block_hash:
-                    params = params + [block_hash]
-                return substrate.rpc_request(
-                    method="neuronInfo_getNeuronsLite",  # custom rpc method
-                    params=params,
-                )
-
-        json_body = make_substrate_call_with_retry()
-        result = json_body["result"]
-
-        if result in (None, []):
-            return []
+        if hex_bytes_result == None:
+            return None
+
+        bytes_result = bytes.fromhex(hex_bytes_result[2:])
 
-        return NeuronInfoLite.list_from_vec_u8(result)
+        return NeuronInfoLite.list_from_vec_u8(bytes_result)
 
     def metagraph(
         self, netuid: int, lite: bool = True, block: Optional[int] = None