Use async-substrate-interface for runtime decoding

bittensor/core/chain_data/__init__.py

@@ -19,7 +19,6 @@
 )
 from .neuron_info import NeuronInfo
 from .neuron_info_lite import NeuronInfoLite
-from .neuron_certificate import NeuronCertificate
 from .prometheus_info import PrometheusInfo
 from .proposal_vote_data import ProposalVoteData
 from .scheduled_coldkey_swap_info import ScheduledColdkeySwapInfo
@@ -29,7 +28,7 @@
 from .subnet_info import SubnetInfo
 from .subnet_state import SubnetState
 from .weight_commit_info import WeightCommitInfo
-from .utils import custom_rpc_type_registry, decode_account_id, process_stake_data
+from .utils import decode_account_id, process_stake_data
 
 ProposalCallData = GenericCall
 
@@ -46,7 +45,6 @@
     MetagraphInfoPool,
     NeuronInfo,
     NeuronInfoLite,
-    NeuronCertificate,
     PrometheusInfo,
     ProposalCallData,
     ProposalVoteData,
@@ -57,7 +55,6 @@
     SubnetInfo,
     SubnetState,
     WeightCommitInfo,
-    custom_rpc_type_registry,
     decode_account_id,
     process_stake_data,
 ]

bittensor/core/chain_data/axon_info.py

@@ -6,14 +6,16 @@
 from dataclasses import asdict, dataclass
 from typing import Any, Union
 
+import netaddr
 from async_substrate_interface.utils import json
+from bittensor.core.chain_data.info_base import InfoBase
 from bittensor.utils import networking
 from bittensor.utils.btlogging import logging
 from bittensor.utils.registration import torch, use_torch
 
 
 @dataclass
-class AxonInfo:
+class AxonInfo(InfoBase):
     """
     The `AxonInfo` class represents information about an axon endpoint in the bittensor network. This includes
     properties such as IP address, ports, and relevant keys.
@@ -79,6 +81,20 @@ def to_string(self) -> str:
             logging.error(f"Error converting AxonInfo to string: {e}")
             return AxonInfo(0, "", 0, 0, "", "").to_string()
 
+    @classmethod
+    def _from_dict(cls, data):
+        return AxonInfo(
+            version=data["version"],
+            ip=str(netaddr.IPAddress(data["ip"])),
+            port=data["port"],
+            ip_type=data["ip_type"],
+            placeholder1=data["placeholder1"],
+            placeholder2=data["placeholder2"],
+            protocol=data["protocol"],
+            hotkey=data["hotkey"],
+            coldkey=data["coldkey"],
+        )
+
     @classmethod
     def from_string(cls, json_string: str) -> "AxonInfo":
         """

bittensor/core/chain_data/delegate_info.py

@@ -1,15 +1,14 @@
-import bt_decode
-
 from dataclasses import dataclass
-from typing import Optional
+from typing import Any, Optional
 
+from bittensor.core.chain_data.info_base import InfoBase
 from bittensor.core.chain_data.utils import decode_account_id
 from bittensor.utils import u16_normalized_float
 from bittensor.utils.balance import Balance
 
 
 @dataclass
-class DelegateInfo:
+class DelegateInfo(InfoBase):
     """
     Dataclass for delegate information. For a lighter version of this class, see ``DelegateInfoLite``.
 
@@ -40,73 +39,33 @@ class DelegateInfo:
     total_daily_return: Balance  # Total daily return of the delegate
 
     @classmethod
-    def from_vec_u8(cls, vec_u8: bytes) -> Optional["DelegateInfo"]:
-        decoded = bt_decode.DelegateInfo.decode(vec_u8)
-        hotkey = decode_account_id(decoded.delegate_ss58)
-        owner = decode_account_id(decoded.owner_ss58)
+    def _from_dict(cls, decoded: dict) -> Optional["DelegateInfo"]:
         nominators = [
-            (decode_account_id(x), Balance.from_rao(y)) for x, y in decoded.nominators
+            (decode_account_id(x), Balance.from_rao(y))
+            for x, y in decoded["nominators"]
         ]
         total_stake = sum((x[1] for x in nominators)) if nominators else Balance(0)
+
         return DelegateInfo(
-            hotkey_ss58=hotkey,
-            total_stake=total_stake,
+            hotkey_ss58=decode_account_id(decoded["delegate_ss58"]),
             nominators=nominators,
-            owner_ss58=owner,
-            take=u16_normalized_float(decoded.take),
-            validator_permits=decoded.validator_permits,
-            registrations=decoded.registrations,
-            return_per_1000=Balance.from_rao(decoded.return_per_1000),
-            total_daily_return=Balance.from_rao(decoded.total_daily_return),
+            owner_ss58=decode_account_id(decoded["owner_ss58"]),
+            registrations=decoded["registrations"],
+            return_per_1000=Balance.from_rao(decoded["return_per_1000"]),
+            take=u16_normalized_float(decoded["take"]),
+            total_daily_return=Balance.from_rao(decoded["total_daily_return"]),
+            total_stake=total_stake,
+            validator_permits=decoded["validator_permits"],
         )
 
     @classmethod
-    def list_from_vec_u8(cls, vec_u8: bytes) -> list["DelegateInfo"]:
-        decoded = bt_decode.DelegateInfo.decode_vec(vec_u8)
-        results = []
-        for d in decoded:
-            hotkey = decode_account_id(d.delegate_ss58)
-            owner = decode_account_id(d.owner_ss58)
-            nominators = [
-                (decode_account_id(x), Balance.from_rao(y)) for x, y in d.nominators
-            ]
-            total_stake = sum((x[1] for x in nominators)) if nominators else Balance(0)
-            results.append(
-                DelegateInfo(
-                    hotkey_ss58=hotkey,
-                    total_stake=total_stake,
-                    nominators=nominators,
-                    owner_ss58=owner,
-                    take=u16_normalized_float(d.take),
-                    validator_permits=d.validator_permits,
-                    registrations=d.registrations,
-                    return_per_1000=Balance.from_rao(d.return_per_1000),
-                    total_daily_return=Balance.from_rao(d.total_daily_return),
-                )
-            )
-        return results
-
-    @classmethod
-    def delegated_list_from_vec_u8(
-        cls, vec_u8: bytes
+    def delegated_list_from_dicts(
+        cls, delegates: list[Any]
     ) -> list[tuple["DelegateInfo", Balance]]:
-        decoded = bt_decode.DelegateInfo.decode_delegated(vec_u8)
-        results = []
-        for d, b in decoded:
-            nominators = [
-                (decode_account_id(x), Balance.from_rao(y)) for x, y in d.nominators
-            ]
-            total_stake = sum((x[1] for x in nominators)) if nominators else Balance(0)
-            delegate = DelegateInfo(
-                hotkey_ss58=decode_account_id(d.delegate_ss58),
-                total_stake=total_stake,
-                nominators=nominators,
-                owner_ss58=decode_account_id(d.owner_ss58),
-                take=u16_normalized_float(d.take),
-                validator_permits=d.validator_permits,
-                registrations=d.registrations,
-                return_per_1000=Balance.from_rao(d.return_per_1000),
-                total_daily_return=Balance.from_rao(d.total_daily_return),
+        return [
+            (
+                DelegateInfo.from_dict(delegate),
+                Balance.from_rao(balance),
             )
-            results.append((delegate, Balance.from_rao(b)))
-        return results
+            for delegate, balance in delegates
+        ]

bittensor/core/chain_data/delegate_info_lite.py

@@ -1,8 +1,10 @@
 from dataclasses import dataclass
 
+from bittensor.core.chain_data.info_base import InfoBase
+
 
 @dataclass
-class DelegateInfoLite:
+class DelegateInfoLite(InfoBase):
     """
     Dataclass for `DelegateLiteInfo`. This is a lighter version of :func:``DelegateInfo``.
 

bittensor/core/chain_data/dynamic_info.py

@@ -6,19 +6,15 @@
 from dataclasses import dataclass
 from typing import Optional, Union
 
-from scalecodec.utils.ss58 import ss58_encode
+from bittensor.core.chain_data.info_base import InfoBase
+from bittensor.core.chain_data.utils import decode_account_id
 
-from bittensor.core.chain_data.utils import (
-    ChainDataType,
-    from_scale_encoding,
-    SS58_FORMAT,
-)
 from bittensor.core.chain_data.subnet_identity import SubnetIdentity
 from bittensor.utils.balance import Balance
 
 
 @dataclass
-class DynamicInfo:
+class DynamicInfo(InfoBase):
     netuid: int
     owner_hotkey: str
     owner_coldkey: str
@@ -43,26 +39,7 @@ class DynamicInfo:
     subnet_identity: Optional[SubnetIdentity]
 
     @classmethod
-    def from_vec_u8(cls, vec_u8: Union[list[int], bytes]) -> Optional["DynamicInfo"]:
-        if len(vec_u8) == 0:
-            return None
-        decoded = from_scale_encoding(vec_u8, ChainDataType.DynamicInfo)
-        if decoded is None:
-            return None
-        return DynamicInfo.fix_decoded_values(decoded)
-
-    @classmethod
-    def list_from_vec_u8(cls, vec_u8: Union[list[int], bytes]) -> list["DynamicInfo"]:
-        decoded = from_scale_encoding(
-            vec_u8, ChainDataType.DynamicInfo, is_vec=True, is_option=True
-        )
-        if decoded is None:
-            return []
-        decoded = [DynamicInfo.fix_decoded_values(d) for d in decoded]
-        return decoded
-
-    @classmethod
-    def fix_decoded_values(cls, decoded: dict) -> "DynamicInfo":
+    def _from_dict(cls, decoded: dict) -> "DynamicInfo":
         """Returns a DynamicInfo object from a decoded DynamicInfo dictionary."""
 
         netuid = int(decoded["netuid"])
@@ -73,8 +50,8 @@ def fix_decoded_values(cls, decoded: dict) -> "DynamicInfo":
             True if int(decoded["netuid"]) > 0 else False
         )  # Root is not dynamic
 
-        owner_hotkey = ss58_encode(decoded["owner_hotkey"], SS58_FORMAT)
-        owner_coldkey = ss58_encode(decoded["owner_coldkey"], SS58_FORMAT)
+        owner_hotkey = decode_account_id(decoded["owner_hotkey"])
+        owner_coldkey = decode_account_id(decoded["owner_coldkey"])
 
         emission = Balance.from_rao(decoded["emission"]).set_unit(0)
         alpha_in = Balance.from_rao(decoded["alpha_in"]).set_unit(netuid)
@@ -104,9 +81,11 @@ def fix_decoded_values(cls, decoded: dict) -> "DynamicInfo":
 
         if decoded.get("subnet_identity"):
             subnet_identity = SubnetIdentity(
-                subnet_name=decoded["subnet_identity"]["subnet_name"],
-                github_repo=decoded["subnet_identity"]["github_repo"],
-                subnet_contact=decoded["subnet_identity"]["subnet_contact"],
+                subnet_name=bytes(decoded["subnet_identity"]["subnet_name"]).decode(),
+                github_repo=bytes(decoded["subnet_identity"]["github_repo"]).decode(),
+                subnet_contact=bytes(
+                    decoded["subnet_identity"]["subnet_contact"]
+                ).decode(),
             )
         else:
             subnet_identity = None

bittensor/core/chain_data/info_base.py

@@ -0,0 +1,28 @@
+from dataclasses import dataclass
+from typing import Any, TypeVar
+
+from bittensor.core.errors import SubstrateRequestException
+
+T = TypeVar("T", bound="InfoBase")
+
+
+@dataclass
+class InfoBase:
+    """Base dataclass for info objects."""
+
+    @classmethod
+    def from_dict(cls, decoded: dict) -> T:
+        try:
+            return cls._from_dict(decoded)
+        except KeyError as e:
+            raise SubstrateRequestException(
+                f"The {cls} structure is missing {e} from the chain.",
+            )
+
+    @classmethod
+    def list_from_dicts(cls, any_list: list[Any]) -> list[T]:
+        return [cls.from_dict(any_) for any_ in any_list]
+
+    @classmethod
+    def _from_dict(cls, decoded: dict) -> T:
+        return cls(**decoded)

bittensor/core/chain_data/ip_info.py

@@ -1,7 +1,6 @@
 from dataclasses import dataclass
-from typing import Optional, Any, Union
+from typing import Any, Union
 
-from bittensor.core.chain_data.utils import from_scale_encoding, ChainDataType
 from bittensor.utils import networking as net
 from bittensor.utils.registration import torch, use_torch
 
@@ -31,33 +30,11 @@ def encode(self) -> dict[str, Any]:
         }
 
     @classmethod
-    def from_vec_u8(cls, vec_u8: list[int]) -> Optional["IPInfo"]:
-        """Returns a IPInfo object from a ``vec_u8``."""
-        if len(vec_u8) == 0:
-            return None
-
-        decoded = from_scale_encoding(vec_u8, ChainDataType.IPInfo)
-        if decoded is None:
-            return None
-
-        return IPInfo.fix_decoded_values(decoded)
-
-    @classmethod
-    def list_from_vec_u8(cls, vec_u8: list[int]) -> list["IPInfo"]:
-        """Returns a list of IPInfo objects from a ``vec_u8``."""
-        decoded = from_scale_encoding(vec_u8, ChainDataType.IPInfo, is_vec=True)
-
-        if decoded is None:
-            return []
-
-        return [IPInfo.fix_decoded_values(d) for d in decoded]
-
-    @classmethod
-    def fix_decoded_values(cls, decoded: dict) -> "IPInfo":
+    def _from_dict(cls, decoded: dict) -> "IPInfo":
         """Returns a SubnetInfo object from a decoded IPInfo dictionary."""
         return IPInfo(
-            ip=net.int_to_ip(decoded["ip"]),
             ip_type=decoded["ip_type_and_protocol"] >> 4,
+            ip=net.int_to_ip(decoded["ip"]),
             protocol=decoded["ip_type_and_protocol"] & 0xF,
         )