Skip to content
2 changes: 2 additions & 0 deletions bittensor_cli/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -6551,6 +6551,7 @@ def stake_childkey_take(
wait_for_inclusion: bool = Options.wait_for_inclusion,
wait_for_finalization: bool = Options.wait_for_finalization,
prompt: bool = Options.prompt,
decline: bool = Options.decline,
quiet: bool = Options.quiet,
verbose: bool = Options.verbose,
json_output: bool = Options.json_output,
Expand Down Expand Up @@ -6608,6 +6609,7 @@ def stake_childkey_take(
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
prompt=prompt,
decline=decline,
)
)
if json_output:
Expand Down
128 changes: 31 additions & 97 deletions bittensor_cli/src/bittensor/chain_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
SS58_FORMAT,
u16_normalized_float as u16tf,
u64_normalized_float as u64tf,
decode_account_id,
get_netuid_and_subuid_by_storage_index,
)

Expand All @@ -37,28 +36,9 @@ class ChainDataType(Enum):
SubnetIdentity = 11


def decode_hex_identity(info_dictionary):
decoded_info = {}
for k, v in info_dictionary.items():
if isinstance(v, dict):
item = next(iter(v.values()))
else:
item = v

if isinstance(item, tuple):
try:
decoded_info[k] = bytes(item).decode()
except UnicodeDecodeError:
print(f"Could not decode: {k}: {item}")
else:
decoded_info[k] = item
return decoded_info


def process_stake_data(stake_data, netuid):
decoded_stake_data = {}
for account_id_bytes, stake_ in stake_data:
account_id = decode_account_id(account_id_bytes)
for account_id, stake_ in stake_data:
decoded_stake_data.update(
{account_id: Balance.from_rao(stake_).set_unit(netuid)}
)
Expand All @@ -70,32 +50,6 @@ def _tbwu(val: int, netuid: Optional[int] = 0) -> Balance:
return Balance.from_rao(val).set_unit(netuid)


def _chr_str(codes: tuple[int]) -> str:
"""Converts a tuple of integer Unicode code points into a string."""
return "".join(map(chr, codes))


def process_nested(
data: Sequence[dict[Hashable, tuple[int]]] | dict | Any,
chr_transform: Callable[[tuple[int]], str],
) -> list[dict[Hashable, str]] | dict[Hashable, str] | Any:
"""Processes nested data structures by applying a transformation function to their elements."""
if isinstance(data, Sequence):
if len(data) > 0 and isinstance(data[0], dict):
return [
{k: chr_transform(v) for k, v in item.items()}
if item is not None
else None
for item in data
]
# TODO @abe why do we kind of silently fail here?
return {}
elif isinstance(data, dict):
return {k: chr_transform(v) for k, v in data.items()}
else:
return data


@dataclass
class AxonInfo:
version: int
Expand Down Expand Up @@ -292,8 +246,8 @@ class StakeInfo(InfoBase):

@classmethod
def _fix_decoded(cls, decoded: Any) -> "StakeInfo":
hotkey = decode_account_id(decoded.get("hotkey"))
coldkey = decode_account_id(decoded.get("coldkey"))
hotkey = decoded.get("hotkey")
coldkey = decoded.get("coldkey")
netuid = int(decoded.get("netuid"))
stake = Balance.from_rao(decoded.get("stake")).set_unit(netuid)
locked = Balance.from_rao(decoded.get("locked")).set_unit(netuid)
Expand Down Expand Up @@ -390,8 +344,8 @@ def _fix_decoded(cls, decoded: Any) -> "NeuronInfo":
stake_dict = process_stake_data(decoded.get("stake"), netuid=netuid)
total_stake = sum(stake_dict.values()) if stake_dict else Balance(0)
axon_info = decoded.get("axon_info", {})
coldkey = decode_account_id(decoded.get("coldkey"))
hotkey = decode_account_id(decoded.get("hotkey"))
coldkey = decoded.get("coldkey")
hotkey = decoded.get("hotkey")
return cls(
hotkey=hotkey,
coldkey=coldkey,
Expand Down Expand Up @@ -484,11 +438,11 @@ def get_null_neuron() -> "NeuronInfoLite":
def _fix_decoded(cls, decoded: Union[dict, "NeuronInfoLite"]) -> "NeuronInfoLite":
active = decoded.get("active")
axon_info = decoded.get("axon_info", {})
coldkey = decode_account_id(decoded.get("coldkey"))
coldkey = decoded.get("coldkey")
consensus = decoded.get("consensus")
dividends = decoded.get("dividends")
emission = decoded.get("emission")
hotkey = decode_account_id(decoded.get("hotkey"))
hotkey = decoded.get("hotkey")
incentive = decoded.get("incentive")
last_update = decoded.get("last_update")
netuid = decoded.get("netuid")
Expand Down Expand Up @@ -569,12 +523,9 @@ class DelegateInfo(InfoBase):

@classmethod
def _fix_decoded(cls, decoded: "DelegateInfo") -> "DelegateInfo":
hotkey = decode_account_id(decoded.get("hotkey_ss58"))
owner = decode_account_id(decoded.get("owner_ss58"))
nominators = [
(decode_account_id(x), Balance.from_rao(y))
for x, y in decoded.get("nominators")
]
hotkey = decoded.get("hotkey_ss58")
owner = decoded.get("owner_ss58")
nominators = [(x, Balance.from_rao(y)) for x, y in decoded.get("nominators")]
total_stake = sum((x[1] for x in nominators)) if nominators else Balance(0)
return cls(
hotkey_ss58=hotkey,
Expand Down Expand Up @@ -675,7 +626,7 @@ def _fix_decoded(cls, decoded: "SubnetInfo") -> "SubnetInfo":
},
emission_value=decoded.get("emission_value"),
burn=Balance.from_rao(decoded.get("burn")),
owner_ss58=decode_account_id(decoded.get("owner")),
owner_ss58=decoded.get("owner"),
)


Expand All @@ -695,14 +646,14 @@ class SubnetIdentity(InfoBase):
@classmethod
def _fix_decoded(cls, decoded: dict) -> "SubnetIdentity":
return cls(
subnet_name=bytes(decoded["subnet_name"]).decode(),
github_repo=bytes(decoded["github_repo"]).decode(),
subnet_contact=bytes(decoded["subnet_contact"]).decode(),
subnet_url=bytes(decoded["subnet_url"]).decode(),
discord=bytes(decoded["discord"]).decode(),
description=bytes(decoded["description"]).decode(),
logo_url=bytes(decoded["logo_url"]).decode(),
additional=bytes(decoded["additional"]).decode(),
subnet_name=decoded["subnet_name"],
github_repo=decoded["github_repo"],
subnet_contact=decoded["subnet_contact"],
subnet_url=decoded["subnet_url"],
discord=decoded["discord"],
description=decoded["description"],
logo_url=decoded["logo_url"],
additional=decoded["additional"],
)


Expand Down Expand Up @@ -742,8 +693,8 @@ def _fix_decoded(cls, decoded: Any) -> "DynamicInfo":
subnet_name = bytes([int(b) for b in decoded.get("subnet_name")]).decode()
is_dynamic = True if netuid > 0 else False # Patching for netuid 0

owner_hotkey = decode_account_id(decoded.get("owner_hotkey"))
owner_coldkey = decode_account_id(decoded.get("owner_coldkey"))
owner_hotkey = decoded.get("owner_hotkey")
owner_coldkey = decoded.get("owner_coldkey")

emission = Balance.from_rao(decoded.get("emission")).set_unit(0)
alpha_in = Balance.from_rao(decoded.get("alpha_in")).set_unit(netuid)
Expand Down Expand Up @@ -919,11 +870,10 @@ def _fix_decoded(
cls, coldkey: str, decoded: tuple
) -> "ColdkeySwapAnnouncementInfo":
execution_block, new_coldkey_hash = decoded
hash_str = "0x" + bytes(new_coldkey_hash[0]).hex()
return cls(
coldkey=coldkey,
execution_block=int(execution_block),
new_coldkey_hash=hash_str,
new_coldkey_hash=new_coldkey_hash,
)


Expand Down Expand Up @@ -953,8 +903,8 @@ def _fix_decoded(cls, decoded: Any) -> "SubnetState":
netuid = decoded.get("netuid")
return cls(
netuid=netuid,
hotkeys=[decode_account_id(val) for val in decoded.get("hotkeys")],
coldkeys=[decode_account_id(val) for val in decoded.get("coldkeys")],
hotkeys=decoded.get("hotkeys"),
coldkeys=decoded.get("coldkeys"),
active=decoded.get("active"),
validator_permit=decoded.get("validator_permit"),
pruning_score=[u16tf(val) for val in decoded.get("pruning_score")],
Expand Down Expand Up @@ -1122,10 +1072,6 @@ def _fix_decoded(cls, decoded: dict) -> "MetagraphInfo":
# Name and symbol
decoded.update({"name": bytes(decoded.get("name")).decode()})
decoded.update({"symbol": bytes(decoded.get("symbol")).decode()})
for key in ["identities", "identity"]:
raw_data = decoded.get(key)
processed = process_nested(raw_data, _chr_str)
decoded.update({key: processed})

return cls(
# Subnet index
Expand Down Expand Up @@ -1194,8 +1140,8 @@ def _fix_decoded(cls, decoded: dict) -> "MetagraphInfo":
alpha_low=u16tf(decoded["alpha_low"]),
bonds_moving_avg=u64tf(decoded["bonds_moving_avg"]),
# Metagraph info.
hotkeys=[decode_account_id(ck) for ck in decoded.get("hotkeys", [])],
coldkeys=[decode_account_id(hk) for hk in decoded.get("coldkeys", [])],
hotkeys=decoded.get("hotkeys", []),
coldkeys=decoded.get("coldkeys", []),
identities=decoded["identities"],
axons=decoded.get("axons", []),
active=decoded["active"],
Expand All @@ -1214,11 +1160,11 @@ def _fix_decoded(cls, decoded: dict) -> "MetagraphInfo":
total_stake=[_tbwu(ts, _netuid) for ts in decoded["total_stake"]],
# Dividend break down
tao_dividends_per_hotkey=[
(decode_account_id(alpha[0]), _tbwu(alpha[1]))
(alpha[0], _tbwu(alpha[1]))
for alpha in decoded["tao_dividends_per_hotkey"]
],
alpha_dividends_per_hotkey=[
(decode_account_id(adphk[0]), _tbwu(adphk[1], _netuid))
(adphk[0], _tbwu(adphk[1], _netuid))
for adphk in decoded["alpha_dividends_per_hotkey"]
],
)
Expand Down Expand Up @@ -1258,21 +1204,9 @@ class CrowdloanData(InfoBase):

@classmethod
def _fix_decoded(cls, decoded: dict[str, Any]) -> "CrowdloanData":
creator = (
decode_account_id(creator_raw)
if (creator_raw := decoded.get("creator"))
else None
)
funds_account = (
decode_account_id(funds_raw)
if (funds_raw := decoded.get("funds_account"))
else None
)
target_address = (
decode_account_id(target_raw)
if (target_raw := decoded.get("target_address"))
else None
)
creator = decoded.get("creator")
funds_account = decoded.get("funds_account")
target_address = decoded.get("target_address")
return cls(
creator=creator,
funds_account=funds_account,
Expand Down
Loading
Loading