diff --git a/async_substrate_interface/async_substrate.py b/async_substrate_interface/async_substrate.py index dae91e5..de6b82f 100644 --- a/async_substrate_interface/async_substrate.py +++ b/async_substrate_interface/async_substrate.py @@ -3261,7 +3261,9 @@ async def _do_runtime_call_old( param_data = b"" if "encoder" in runtime_call_def: - param_data = runtime_call_def["encoder"](params) + if runtime is None: + runtime = await self.init_runtime(block_hash=block_hash) + param_data = runtime_call_def["encoder"](params, runtime.registry) else: for idx, param in enumerate(runtime_call_def["params"]): param_type_string = f"{param['type']}" diff --git a/async_substrate_interface/sync_substrate.py b/async_substrate_interface/sync_substrate.py index 8ddd90b..affc773 100644 --- a/async_substrate_interface/sync_substrate.py +++ b/async_substrate_interface/sync_substrate.py @@ -2527,28 +2527,21 @@ def _do_runtime_call_old( runtime_call_def = _TYPE_REGISTRY["runtime_api"][api]["methods"][method] # Encode params - param_data = b"" + param_data: Union[ScaleBytes, bytes] = b"" - if "encoder" in runtime_call_def: - param_data = runtime_call_def["encoder"](params) - else: - for idx, param in enumerate(runtime_call_def["params"]): - param_type_string = f"{param['type']}" - if isinstance(params, list): - param_data += self.encode_scale(param_type_string, params[idx]) - else: - if param["name"] not in params: - raise ValueError( - f"Runtime Call param '{param['name']}' is missing" - ) + runtime = self.init_runtime(block_hash=block_hash) - param_data += self.encode_scale( - param_type_string, params[param["name"]] - ) + if "encoder" in runtime_call_def and runtime.registry is not None: + # only works if we have metadata v15 + param_data = runtime_call_def["encoder"](params, runtime.registry) + param_hex = param_data.hex() + else: + param_data = self._encode_scale_legacy(runtime_call_def, params, runtime) + param_hex = param_data.to_hex() # RPC request result_data = self.rpc_request( - "state_call", [f"{api}_{method}", param_data.hex(), block_hash] + "state_call", [f"{api}_{method}", param_hex, block_hash] ) result_vec_u8_bytes = hex_to_bytes(result_data["result"]) result_bytes = self.decode_scale("Vec", result_vec_u8_bytes) diff --git a/async_substrate_interface/type_registry.py b/async_substrate_interface/type_registry.py index 0f224e8..7e2e246 100644 --- a/async_substrate_interface/type_registry.py +++ b/async_substrate_interface/type_registry.py @@ -1,3 +1,5 @@ +from typing import Union +from collections import namedtuple from bt_decode import ( NeuronInfo, NeuronInfoLite, @@ -8,7 +10,54 @@ SubnetInfoV2, encode, ) -from scalecodec import ss58_encode +from scalecodec import ss58_decode + + +def stake_info_decode_vec_legacy_compatibility( + item, +) -> list[dict[str, Union[str, int, bytes, bool]]]: + stake_infos: list[StakeInfo] = StakeInfo.decode_vec(item) + NewStakeInfo = namedtuple( + "NewStakeInfo", + [ + "netuid", + "hotkey", + "coldkey", + "stake", + "locked", + "emission", + "drain", + "is_registered", + ], + ) + output = [] + for stake_info in stake_infos: + output.append( + NewStakeInfo( + 0, + stake_info.hotkey, + stake_info.coldkey, + stake_info.stake, + 0, + 0, + 0, + False, + ) + ) + return output + + +def preprocess_get_stake_info_for_coldkeys(addrs): + output = [] + if isinstance(addrs[0], list): # I think + for addr in addrs[0]: + output.append(list(bytes.fromhex(ss58_decode(addr)))) + else: + if isinstance(addrs[0], dict): + for addr in addrs[0]["coldkey_accounts"]: + output.append(list(bytes.fromhex(ss58_decode(addr)))) + return output + _TYPE_REGISTRY: dict[str, dict] = { "types": { @@ -24,7 +73,9 @@ "type": "Vec", }, ], - "encoder": lambda addr: encode(ss58_encode(addr), "Vec"), + "encoder": lambda addr, reg: encode( + "Vec", reg, list(bytes.fromhex(ss58_decode(addr))) + ), "type": "Vec", "decoder": DelegateInfo.decode_delegated, }, @@ -97,8 +148,20 @@ }, ], "type": "Vec", - "encoder": lambda addr: encode(ss58_encode(addr), "Vec"), - "decoder": StakeInfo.decode_vec, + "encoder": lambda addr, reg: encode( + "Vec", + reg, + list( + bytes.fromhex( + ss58_decode( + addr[0] + if isinstance(addr, list) + else addr["coldkey_account"] + ) + ) + ), + ), + "decoder": stake_info_decode_vec_legacy_compatibility, }, "get_stake_info_for_coldkeys": { "params": [ @@ -108,8 +171,10 @@ }, ], "type": "Vec", - "encoder": lambda addrs: encode( - [ss58_encode(addr) for addr in addrs], "Vec>" + "encoder": lambda addrs, reg: encode( + "Vec>", + reg, + preprocess_get_stake_info_for_coldkeys(addrs), ), "decoder": StakeInfo.decode_vec_tuple_vec, }, diff --git a/async_substrate_interface/types.py b/async_substrate_interface/types.py index b9fbfae..931b6c9 100644 --- a/async_substrate_interface/types.py +++ b/async_substrate_interface/types.py @@ -965,6 +965,23 @@ def _encode_scale( result = bytes(encode_by_type_string(type_string, runtime.registry, value)) return result + @staticmethod + def _encode_scale_legacy(call_definition: list[dict], params: Union[list[Any], dict[str, Any]], runtime:Runtime) -> bytes: + """Returns a hex encoded string of the params using their types.""" + param_data = scalecodec.ScaleBytes(b"") + + for i, param in enumerate(call_definition["params"]): # type: ignore + scale_obj = runtime.runtime_config.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 + @staticmethod def _encode_account_id(account) -> bytes: """Encode an account ID into bytes. diff --git a/tests/integration_tests/test_substrate_interface.py b/tests/integration_tests/test_substrate_interface.py index f6cf0eb..b574104 100644 --- a/tests/integration_tests/test_substrate_interface.py +++ b/tests/integration_tests/test_substrate_interface.py @@ -163,3 +163,17 @@ def test_get_payment_info(): assert partial_fee_all_options > partial_fee_no_era assert partial_fee_all_options > partial_fee_era print("test_get_payment_info succeeded") + + +def test_old_runtime_calls(): + from bittensor import SubtensorApi + + sub = SubtensorApi(network="archive", legacy_methods=True, async_subtensor=False) + # will pass + assert sub.get_stake_info_for_coldkey( + "5CQ6dMW8JZhKCZX9kWsZRqa3kZRKmNHxbPPVFEt6FgyvGv2G", 4943592 + ) + # needs to use legacy + assert sub.get_stake_info_for_coldkey( + "5CQ6dMW8JZhKCZX9kWsZRqa3kZRKmNHxbPPVFEt6FgyvGv2G", 4670227 + )