Skip to content

[fix] change legacy (old) runtimeApi params encoding #194

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
camfairchild wants to merge 7 commits into
base: staging
Choose a base branch
from
Open

[fix] change legacy (old) runtimeApi params encoding #194

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
46a21fb
change legacy (old) runtimeApi params encoding
camfairchild Oct 7, 2025
e083cfc
Small fixes, still WIP
thewhaleking Oct 8, 2025
936d6e8
Added test (failing currently, needs to pass)
thewhaleking Oct 8, 2025
e85cec2
Add Legacy conversion namedtuple
thewhaleking Oct 8, 2025
33a9f18
WIP
thewhaleking Oct 8, 2025
9ec5cb6
Merge branch 'staging' into fix/fix-legacy-runtime-api-params-encoding
thewhaleking Jan 8, 2026
1b8143f
Checkin
thewhaleking Jan 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion async_substrate_interface/async_substrate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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']}"
Expand Down
27 changes: 10 additions & 17 deletions async_substrate_interface/sync_substrate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<u8>", result_vec_u8_bytes)
Expand Down
77 changes: 71 additions & 6 deletions async_substrate_interface/type_registry.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from typing import Union
from collections import namedtuple
from bt_decode import (
NeuronInfo,
NeuronInfoLite,
Expand All @@ -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": {
Expand All @@ -24,7 +73,9 @@
"type": "Vec<u8>",
},
],
"encoder": lambda addr: encode(ss58_encode(addr), "Vec<u8>"),
"encoder": lambda addr, reg: encode(
"Vec<u8>", reg, list(bytes.fromhex(ss58_decode(addr)))
),
"type": "Vec<u8>",
"decoder": DelegateInfo.decode_delegated,
},
Expand Down Expand Up @@ -97,8 +148,20 @@
},
],
"type": "Vec<u8>",
"encoder": lambda addr: encode(ss58_encode(addr), "Vec<u8>"),
"decoder": StakeInfo.decode_vec,
"encoder": lambda addr, reg: encode(
"Vec<u8>",
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": [
Expand All @@ -108,8 +171,10 @@
},
],
"type": "Vec<u8>",
"encoder": lambda addrs: encode(
[ss58_encode(addr) for addr in addrs], "Vec<Vec<u8>>"
"encoder": lambda addrs, reg: encode(
"Vec<Vec<u8>>",
reg,
preprocess_get_stake_info_for_coldkeys(addrs),
),
"decoder": StakeInfo.decode_vec_tuple_vec,
},
Expand Down
17 changes: 17 additions & 0 deletions async_substrate_interface/types.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
14 changes: 14 additions & 0 deletions tests/integration_tests/test_substrate_interface.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Loading