Use async-substrate-interface for runtime decoding

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

@@ -8,6 +8,7 @@
 
 from scalecodec.utils.ss58 import ss58_encode
 
+from bittensor.core.chain_data.info_base import InfoBase
 from bittensor.core.chain_data.utils import (
     ChainDataType,
     from_scale_encoding,
@@ -18,7 +19,7 @@
 
 
 @dataclass
-class DynamicInfo:
+class DynamicInfo(InfoBase):
     netuid: int
     owner_hotkey: str
     owner_coldkey: str
@@ -43,26 +44,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"])

bittensor/core/chain_data/info_base.py

@@ -0,0 +1,17 @@
+from dataclasses import dataclass
+from typing import Any, TypeVar
+
+T = TypeVar("T", bound="InfoBase")
+
+
+@dataclass
+class InfoBase:
+    """Base dataclass for info objects."""
+
+    @classmethod
+    def from_dict(cls, decoded: dict) -> T:
+        return cls(**decoded)
+
+    @classmethod
+    def list_from_dicts(cls, any_list: list[Any]) -> list[T]:
+        return [cls.from_dict(any_) for any_ in any_list]

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,
         )
 

bittensor/core/chain_data/metagraph_info.py

@@ -3,6 +3,7 @@
 
 from bittensor.core.chain_data.axon_info import AxonInfo
 from bittensor.core.chain_data.chain_identity import ChainIdentity
+from bittensor.core.chain_data.info_base import InfoBase
 from bittensor.core.chain_data.subnet_identity import SubnetIdentity
 from bittensor.core.chain_data.utils import (
     ChainDataType,
@@ -20,7 +21,7 @@ def _tbwu(val: int, netuid: Optional[int] = 0) -> Balance:
 
 
 @dataclass
-class MetagraphInfo:
+class MetagraphInfo(InfoBase):
     # Subnet index
     netuid: int
 
@@ -120,34 +121,7 @@ class MetagraphInfo:
     ]  # List of dividend payout in alpha via subnet.
 
     @classmethod
-    def from_vec_u8(cls, vec_u8: bytes) -> Optional["MetagraphInfo"]:
-        """Returns a Metagraph object from encoded MetagraphInfo vector."""
-        if len(vec_u8) == 0:
-            return None
-        decoded = from_scale_encoding(vec_u8, ChainDataType.MetagraphInfo)
-        if decoded is None:
-            return None
-
-        return MetagraphInfo.fix_decoded_values(decoded)
-
-    @classmethod
-    def list_from_vec_u8(cls, vec_u8: bytes) -> list["MetagraphInfo"]:
-        """Returns a list of Metagraph objects from a list of encoded MetagraphInfo vectors."""
-        decoded = from_scale_encoding(
-            vec_u8, ChainDataType.MetagraphInfo, is_vec=True, is_option=True
-        )
-        if decoded is None:
-            return []
-
-        decoded = [
-            MetagraphInfo.fix_decoded_values(meta)
-            for meta in decoded
-            if meta is not None
-        ]
-        return decoded
-
-    @classmethod
-    def fix_decoded_values(cls, decoded: dict) -> "MetagraphInfo":
+    def from_dict(cls, decoded: dict) -> "MetagraphInfo":
         """Returns a Metagraph object from a decoded MetagraphInfo dictionary."""
         # Subnet index
         _netuid = decoded["netuid"]

bittensor/core/chain_data/neuron_certificate.py

@@ -1,20 +1,13 @@
 from dataclasses import dataclass
-from typing import List
 
-from bittensor.core.chain_data.utils import from_scale_encoding, ChainDataType
 from bittensor.utils import Certificate
 
 
 # Dataclasses for chain data.
 @dataclass
-class NeuronCertificate:
+class NeuronCertificate:  # TODO Info?
     """
     Dataclass for neuron certificate.
     """
 
     certificate: Certificate
-
-    @classmethod
-    def from_vec_u8(cls, vec_u8: List[int]) -> "NeuronCertificate":
-        """Returns a NeuronCertificate object from a ``vec_u8``."""
-        return from_scale_encoding(vec_u8, ChainDataType.NeuronCertificate)