Poem "Risen from the Past". Act 3. (#2363)

* add `get_delegate_by_hotkey`, update `DelegateInfo` in chain data

* add `root_register_extrinsic`, `set_root_weights_extrinsic` and related stuff

* add `Subtensor.get_all_subnets_info` method and related stuff

* add `Subtensor.get_delegate_take` method and tests

* ruff

* remove unused import

Author: basfroman

bittensor/core/chain_data/delegate_info.py

@@ -1,10 +1,9 @@
-from dataclasses import dataclass
-from typing import Optional, Any
+import bt_decode
 
-from scalecodec.utils.ss58 import ss58_encode
+from dataclasses import dataclass
+from typing import Optional
 
-from bittensor.core.chain_data.utils import from_scale_encoding, ChainDataType
-from bittensor.core.settings import SS58_FORMAT
+from bittensor.core.chain_data.utils import decode_account_id
 from bittensor.utils import u16_normalized_float
 from bittensor.utils.balance import Balance
 
@@ -24,7 +23,6 @@ class DelegateInfo:
         validator_permits (list[int]): List of subnets that the delegate is allowed to validate on.
         return_per_1000 (int): Return per 1000 TAO, for the delegate over a day.
         total_daily_return (int): Total daily return of the delegate.
-
     """
 
     hotkey_ss58: str  # Hotkey of delegate
@@ -37,69 +35,78 @@ class DelegateInfo:
     validator_permits: list[
         int
     ]  # List of subnets that the delegate is allowed to validate on
-    registrations: tuple[int]  # List of subnets that the delegate is registered on
+    registrations: list[int]  # list of subnets that the delegate is registered on
     return_per_1000: Balance  # Return per 1000 tao of the delegate over a day
     total_daily_return: Balance  # Total daily return of the delegate
 
     @classmethod
-    def fix_decoded_values(cls, decoded: Any) -> "DelegateInfo":
-        """Fixes the decoded values."""
-
-        return cls(
-            hotkey_ss58=ss58_encode(decoded["delegate_ss58"], SS58_FORMAT),
-            owner_ss58=ss58_encode(decoded["owner_ss58"], SS58_FORMAT),
-            take=u16_normalized_float(decoded["take"]),
-            nominators=[
-                (
-                    ss58_encode(nom[0], SS58_FORMAT),
-                    Balance.from_rao(nom[1]),
-                )
-                for nom in decoded["nominators"]
-            ],
-            total_stake=Balance.from_rao(
-                sum([nom[1] for nom in decoded["nominators"]])
-            ),
-            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"]),
+    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)
+        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,
+            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),
         )
 
     @classmethod
-    def from_vec_u8(cls, vec_u8: list[int]) -> Optional["DelegateInfo"]:
-        """Returns a DelegateInfo object from a ``vec_u8``."""
-        if len(vec_u8) == 0:
-            return None
-
-        decoded = from_scale_encoding(vec_u8, ChainDataType.DelegateInfo)
-        if decoded is None:
-            return None
-
-        return DelegateInfo.fix_decoded_values(decoded)
-
-    @classmethod
-    def list_from_vec_u8(cls, vec_u8: list[int]) -> list["DelegateInfo"]:
-        """Returns a list of DelegateInfo objects from a ``vec_u8``."""
-        decoded = from_scale_encoding(vec_u8, ChainDataType.DelegateInfo, is_vec=True)
-
-        if decoded is None:
-            return []
-
-        return [DelegateInfo.fix_decoded_values(d) for d in decoded]
+    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: list[int]
-    ) -> list[tuple["DelegateInfo", "Balance"]]:
-        """Returns a list of Tuples of DelegateInfo objects, and Balance, from a ``vec_u8``.
-
-        This is the list of delegates that the user has delegated to, and the amount of stake delegated.
-        """
-        decoded = from_scale_encoding(vec_u8, ChainDataType.DelegatedInfo, is_vec=True)
-        if decoded is None:
-            return []
-
-        return [
-            (DelegateInfo.fix_decoded_values(d), Balance.from_rao(s))
-            for d, s in decoded
-        ]
+        cls, vec_u8: bytes
+    ) -> 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),
+            )
+            results.append((delegate, Balance.from_rao(b)))
+        return results

bittensor/core/chain_data/subnet_info.py

@@ -1,13 +1,10 @@
 from dataclasses import dataclass
-from typing import Any, Optional, Union
 
-from scalecodec.utils.ss58 import ss58_encode
+import bt_decode
 
-from bittensor.core.chain_data.utils import from_scale_encoding, ChainDataType
-from bittensor.core.settings import SS58_FORMAT
+from bittensor.core.chain_data.utils import decode_account_id
 from bittensor.utils import u16_normalized_float
 from bittensor.utils.balance import Balance
-from bittensor.utils.registration import torch, use_torch
 
 
 @dataclass
@@ -28,76 +25,39 @@ class SubnetInfo:
     blocks_since_epoch: int
     tempo: int
     modality: int
-    # netuid -> topk percentile prunning score requirement (u16:MAX normalized.)
     connection_requirements: dict[str, float]
     emission_value: float
     burn: Balance
     owner_ss58: str
 
     @classmethod
-    def from_vec_u8(cls, vec_u8: list[int]) -> Optional["SubnetInfo"]:
-        """Returns a SubnetInfo object from a ``vec_u8``."""
-        if len(vec_u8) == 0:
-            return None
-
-        decoded = from_scale_encoding(vec_u8, ChainDataType.SubnetInfo)
-        if decoded is None:
-            return None
-
-        return SubnetInfo.fix_decoded_values(decoded)
-
-    @classmethod
-    def list_from_vec_u8(cls, vec_u8: list[int]) -> list["SubnetInfo"]:
-        """Returns a list of SubnetInfo objects from a ``vec_u8``."""
-        decoded = from_scale_encoding(
-            vec_u8, ChainDataType.SubnetInfo, is_vec=True, is_option=True
-        )
-
-        if decoded is None:
-            return []
-
-        return [SubnetInfo.fix_decoded_values(d) for d in decoded]
-
-    @classmethod
-    def fix_decoded_values(cls, decoded: dict) -> "SubnetInfo":
-        """Returns a SubnetInfo object from a decoded SubnetInfo dictionary."""
-        return SubnetInfo(
-            netuid=decoded["netuid"],
-            rho=decoded["rho"],
-            kappa=decoded["kappa"],
-            difficulty=decoded["difficulty"],
-            immunity_period=decoded["immunity_period"],
-            max_allowed_validators=decoded["max_allowed_validators"],
-            min_allowed_weights=decoded["min_allowed_weights"],
-            max_weight_limit=decoded["max_weights_limit"],
-            scaling_law_power=decoded["scaling_law_power"],
-            subnetwork_n=decoded["subnetwork_n"],
-            max_n=decoded["max_allowed_uids"],
-            blocks_since_epoch=decoded["blocks_since_last_step"],
-            tempo=decoded["tempo"],
-            modality=decoded["network_modality"],
-            connection_requirements={
-                str(int(netuid)): u16_normalized_float(int(req))
-                for netuid, req in decoded["network_connect"]
-            },
-            emission_value=decoded["emission_values"],
-            burn=Balance.from_rao(decoded["burn"]),
-            owner_ss58=ss58_encode(decoded["owner"], SS58_FORMAT),
-        )
-
-    def to_parameter_dict(self) -> Union[dict[str, Any], "torch.nn.ParameterDict"]:
-        """Returns a torch tensor or dict of the subnet info."""
-        if use_torch():
-            return torch.nn.ParameterDict(self.__dict__)
-        else:
-            return self.__dict__
-
-    @classmethod
-    def from_parameter_dict(
-        cls, parameter_dict: Union[dict[str, Any], "torch.nn.ParameterDict"]
-    ) -> "SubnetInfo":
-        """Creates a SubnetInfo instance from a parameter dictionary."""
-        if use_torch():
-            return cls(**dict(parameter_dict))
-        else:
-            return cls(**parameter_dict)
+    def list_from_vec_u8(cls, vec_u8: bytes) -> list["SubnetInfo"]:
+        decoded = bt_decode.SubnetInfo.decode_vec_option(vec_u8)
+        result = []
+        for d in decoded:
+            result.append(
+                SubnetInfo(
+                    netuid=d.netuid,
+                    rho=d.rho,
+                    kappa=d.kappa,
+                    difficulty=d.difficulty,
+                    immunity_period=d.immunity_period,
+                    max_allowed_validators=d.max_allowed_validators,
+                    min_allowed_weights=d.min_allowed_weights,
+                    max_weight_limit=d.max_weights_limit,
+                    scaling_law_power=d.scaling_law_power,
+                    subnetwork_n=d.subnetwork_n,
+                    max_n=d.max_allowed_uids,
+                    blocks_since_epoch=d.blocks_since_last_step,
+                    tempo=d.tempo,
+                    modality=d.network_modality,
+                    connection_requirements={
+                        str(int(netuid)): u16_normalized_float(int(req))
+                        for (netuid, req) in d.network_connect
+                    },
+                    emission_value=d.emission_values,
+                    burn=Balance.from_rao(d.burn),
+                    owner_ss58=decode_account_id(d.owner),
+                )
+            )
+        return result

bittensor/core/chain_data/utils.py

@@ -260,7 +260,7 @@ def from_scale_encoding_using_type_string(
 }
 
 
-def decode_account_id(account_id_bytes: list) -> str:
+def decode_account_id(account_id_bytes: Union[bytes, str]) -> str:
     """
     Decodes an AccountId from bytes to a Base64 string using SS58 encoding.