diff --git a/bittensor/core/chain_data/delegate_info.py b/bittensor/core/chain_data/delegate_info.py
index d77f1e1412..a840d1bb15 100644
--- a/bittensor/core/chain_data/delegate_info.py
+++ b/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
diff --git a/bittensor/core/chain_data/subnet_info.py b/bittensor/core/chain_data/subnet_info.py
index f1ce151872..4169746a08 100644
--- a/bittensor/core/chain_data/subnet_info.py
+++ b/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
diff --git a/bittensor/core/chain_data/utils.py b/bittensor/core/chain_data/utils.py
index 0544ca85a2..9c21c9d22e 100644
--- a/bittensor/core/chain_data/utils.py
+++ b/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.
diff --git a/bittensor/core/extrinsics/root.py b/bittensor/core/extrinsics/root.py
new file mode 100644
index 0000000000..1fd7e7b26e
--- /dev/null
+++ b/bittensor/core/extrinsics/root.py
@@ -0,0 +1,310 @@
+import time
+from typing import Optional, Union, TYPE_CHECKING
+
+import numpy as np
+from bittensor_wallet.errors import KeyFileError
+from numpy.typing import NDArray
+from retry import retry
+from rich.prompt import Confirm
+
+from bittensor.core.settings import bt_console, version_as_int
+from bittensor.utils import format_error_message, weight_utils
+from bittensor.utils.btlogging import logging
+from bittensor.utils.networking import ensure_connected
+from bittensor.utils.registration import torch, legacy_torch_api_compat
+
+if TYPE_CHECKING:
+ from bittensor_wallet import Wallet
+ from bittensor.core.subtensor import Subtensor
+
+
+@ensure_connected
+def _do_root_register(
+ self: "Subtensor",
+ wallet: "Wallet",
+ wait_for_inclusion: bool = False,
+ wait_for_finalization: bool = True,
+) -> tuple[bool, Optional[str]]:
+ @retry(delay=1, tries=3, backoff=2, max_delay=4)
+ def make_substrate_call_with_retry():
+ # create extrinsic call
+ call = self.substrate.compose_call(
+ call_module="SubtensorModule",
+ call_function="root_register",
+ call_params={"hotkey": wallet.hotkey.ss58_address},
+ )
+ extrinsic = self.substrate.create_signed_extrinsic(
+ call=call, keypair=wallet.coldkey
+ )
+ response = self.substrate.submit_extrinsic(
+ extrinsic,
+ wait_for_inclusion=wait_for_inclusion,
+ wait_for_finalization=wait_for_finalization,
+ )
+
+ # We only wait here if we expect finalization.
+ if not wait_for_finalization and not wait_for_inclusion:
+ return True
+
+ # process if registration successful, try again if pow is still valid
+ response.process_events()
+ if not response.is_success:
+ return False, format_error_message(response.error_message)
+ # Successful registration
+ else:
+ return True, None
+
+ return make_substrate_call_with_retry()
+
+
+def root_register_extrinsic(
+ subtensor: "Subtensor",
+ wallet: "Wallet",
+ wait_for_inclusion: bool = False,
+ wait_for_finalization: bool = True,
+ prompt: bool = False,
+) -> bool:
+ """Registers the wallet to root network.
+
+ Args:
+ subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance.
+ wallet (bittensor_wallet.Wallet): Bittensor wallet object.
+ wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning ``true``, or returns ``false`` if the extrinsic fails to enter the block within the timeout. Default is ``False``.
+ wait_for_finalization (bool): If set, waits for the extrinsic to be finalized on the chain before returning ``true``, or returns ``false`` if the extrinsic fails to be finalized within the timeout. Default is ``True``.
+ prompt (bool): If ``true``, the call waits for confirmation from the user before proceeding. Default is ``False``.
+
+ Returns:
+ success (bool): Flag is ``true`` if extrinsic was finalized or uncluded in the block. If we did not wait for finalization / inclusion, the response is ``true``.
+ """
+
+ try:
+ wallet.unlock_coldkey()
+ except KeyFileError:
+ bt_console.print(
+ ":cross_mark: [red]Keyfile is corrupt, non-writable, non-readable or the password used to decrypt is invalid[/red]:[bold white]\n [/bold white]"
+ )
+ return False
+
+ is_registered = subtensor.is_hotkey_registered(
+ netuid=0, hotkey_ss58=wallet.hotkey.ss58_address
+ )
+ if is_registered:
+ bt_console.print(
+ ":white_heavy_check_mark: [green]Already registered on root network.[/green]"
+ )
+ return True
+
+ if prompt:
+ # Prompt user for confirmation.
+ if not Confirm.ask("Register to root network?"):
+ return False
+
+ with bt_console.status(":satellite: Registering to root network..."):
+ success, err_msg = _do_root_register(
+ wallet=wallet,
+ wait_for_inclusion=wait_for_inclusion,
+ wait_for_finalization=wait_for_finalization,
+ )
+
+ if not success:
+ bt_console.print(f":cross_mark: [red]Failed[/red]: {err_msg}")
+ time.sleep(0.5)
+
+ # Successful registration, final check for neuron and pubkey
+ else:
+ is_registered = subtensor.is_hotkey_registered(
+ netuid=0, hotkey_ss58=wallet.hotkey.ss58_address
+ )
+ if is_registered:
+ bt_console.print(":white_heavy_check_mark: [green]Registered[/green]")
+ return True
+ else:
+ # neuron not found, try again
+ bt_console.print(
+ ":cross_mark: [red]Unknown error. Neuron not found.[/red]"
+ )
+
+
+@ensure_connected
+def _do_set_root_weights(
+ self: "Subtensor",
+ wallet: "Wallet",
+ uids: list[int],
+ vals: list[int],
+ netuid: int = 0,
+ version_key: int = version_as_int,
+ wait_for_inclusion: bool = False,
+ wait_for_finalization: bool = False,
+) -> tuple[bool, Optional[str]]:
+ """
+ Internal method to send a transaction to the Bittensor blockchain, setting weights for specified neurons on root. This method constructs and submits the transaction, handling retries and blockchain communication.
+
+ Args:
+ self (bittensor.core.subtensor.Subtensor): Subtensor instance.
+ wallet (bittensor_wallet.Wallet): The wallet associated with the neuron setting the weights.
+ uids (List[int]): List of neuron UIDs for which weights are being set.
+ vals (List[int]): List of weight values corresponding to each UID.
+ netuid (int): Unique identifier for the network.
+ version_key (int, optional): Version key for compatibility with the network. Defaults is a current ``version_as_int``.
+ wait_for_inclusion (bool, optional): Waits for the transaction to be included in a block. Defaults is ``False``.
+ wait_for_finalization (bool, optional): Waits for the transaction to be finalized on the blockchain. Defaults is ``False``.
+
+ Returns:
+ Tuple[bool, Optional[str]]: A tuple containing a success flag and an optional error message.
+
+ This method is vital for the dynamic weighting mechanism in Bittensor, where neurons adjust their trust in other neurons based on observed performance and contributions on the root network.
+ """
+
+ @retry(delay=2, tries=3, backoff=2, max_delay=4)
+ def make_substrate_call_with_retry():
+ call = self.substrate.compose_call(
+ call_module="SubtensorModule",
+ call_function="set_root_weights",
+ call_params={
+ "dests": uids,
+ "weights": vals,
+ "netuid": netuid,
+ "version_key": version_key,
+ "hotkey": wallet.hotkey.ss58_address,
+ },
+ )
+ # Period dictates how long the extrinsic will stay as part of waiting pool
+ extrinsic = self.substrate.create_signed_extrinsic(
+ call=call,
+ keypair=wallet.coldkey,
+ era={"period": 5},
+ )
+ response = self.substrate.submit_extrinsic(
+ extrinsic,
+ wait_for_inclusion=wait_for_inclusion,
+ wait_for_finalization=wait_for_finalization,
+ )
+ # We only wait here if we expect finalization.
+ if not wait_for_finalization and not wait_for_inclusion:
+ return True, "Not waiting for finalziation or inclusion."
+
+ response.process_events()
+ if response.is_success:
+ return True, "Successfully set weights."
+ else:
+ return False, response.error_message
+
+ return make_substrate_call_with_retry()
+
+
+@legacy_torch_api_compat
+def set_root_weights_extrinsic(
+ subtensor: "Subtensor",
+ wallet: "Wallet",
+ netuids: Union[NDArray[np.int64], "torch.LongTensor", list[int]],
+ weights: Union[NDArray[np.float32], "torch.FloatTensor", list[float]],
+ version_key: int = 0,
+ wait_for_inclusion: bool = False,
+ wait_for_finalization: bool = False,
+ prompt: bool = False,
+) -> bool:
+ """Sets the given weights and values on chain for wallet hotkey account.
+
+ Args:
+ subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance.
+ wallet (bittensor_wallet.Wallet): Bittensor wallet object. Bittensor wallet object.
+ netuids (Union[NDArray[np.int64], torch.LongTensor, list[int]]): The ``netuid`` of the subnet to set weights for.
+ weights (Union[NDArray[np.float32], torch.FloatTensor, list[float]]): Weights to set. These must be ``float`` s and must correspond to the passed ``netuid`` s.
+ version_key (int): The version key of the validator. Default is ``0``.
+ wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning ``true``, or returns ``false`` if the extrinsic fails to enter the block within the timeout. Default is ``False``.
+ wait_for_finalization (bool): If set, waits for the extrinsic to be finalized on the chain before returning ``true``, or returns ``false`` if the extrinsic fails to be finalized within the timeout. Default is ``False``.
+ prompt (bool): If ``true``, the call waits for confirmation from the user before proceeding. Default is ``False``.
+
+ Returns:
+ success (bool): Flag is ``true`` if extrinsic was finalized or uncluded in the block. If we did not wait for finalization / inclusion, the response is ``true``.
+ """
+
+ try:
+ wallet.unlock_coldkey()
+ except KeyFileError:
+ bt_console.print(
+ ":cross_mark: [red]Keyfile is corrupt, non-writable, non-readable or the password used to decrypt is invalid[/red]:[bold white]\n [/bold white]"
+ )
+ return False
+
+ # First convert types.
+ if isinstance(netuids, list):
+ netuids = np.array(netuids, dtype=np.int64)
+ if isinstance(weights, list):
+ weights = np.array(weights, dtype=np.float32)
+
+ # Get weight restrictions.
+ min_allowed_weights = subtensor.min_allowed_weights(netuid=0)
+ max_weight_limit = subtensor.max_weight_limit(netuid=0)
+
+ # Get non zero values.
+ non_zero_weight_idx = np.argwhere(weights > 0).squeeze(axis=1)
+ non_zero_weight_uids = netuids[non_zero_weight_idx]
+ non_zero_weights = weights[non_zero_weight_idx]
+ if non_zero_weights.size < min_allowed_weights:
+ raise ValueError(
+ "The minimum number of weights required to set weights is {}, got {}".format(
+ min_allowed_weights, non_zero_weights.size
+ )
+ )
+
+ # Normalize the weights to max value.
+ formatted_weights = weight_utils.normalize_max_weight(
+ x=weights, limit=max_weight_limit
+ )
+ bt_console.print(
+ f"\nRaw Weights -> Normalized weights: \n\t{weights} -> \n\t{formatted_weights}\n"
+ )
+
+ # Ask before moving on.
+ if prompt:
+ if not Confirm.ask(
+ "Do you want to set the following root weights?:\n[bold white] weights: {}\n uids: {}[/bold white ]?".format(
+ formatted_weights, netuids
+ )
+ ):
+ return False
+
+ with bt_console.status(
+ ":satellite: Setting root weights on [white]{}[/white] ...".format(
+ subtensor.network
+ )
+ ):
+ try:
+ weight_uids, weight_vals = weight_utils.convert_weights_and_uids_for_emit(
+ netuids, weights
+ )
+ success, error_message = _do_set_root_weights(
+ wallet=wallet,
+ netuid=0,
+ uids=weight_uids,
+ vals=weight_vals,
+ version_key=version_key,
+ wait_for_finalization=wait_for_finalization,
+ wait_for_inclusion=wait_for_inclusion,
+ )
+
+ bt_console.print(success, error_message)
+
+ if not wait_for_finalization and not wait_for_inclusion:
+ return True
+
+ if success is True:
+ bt_console.print(":white_heavy_check_mark: [green]Finalized[/green]")
+ logging.success(
+ prefix="Set weights",
+ suffix="Finalized: " + str(success),
+ )
+ return True
+ else:
+ bt_console.print(f":cross_mark: [red]Failed[/red]: {error_message}")
+ logging.warning(
+ prefix="Set weights",
+ suffix="Failed: " + str(error_message),
+ )
+ return False
+
+ except Exception as e:
+ bt_console.print(":cross_mark: [red]Failed[/red]: error:{}".format(e))
+ logging.warning(prefix="Set weights", suffix="Failed: " + str(e))
+ return False
diff --git a/bittensor/core/subtensor.py b/bittensor/core/subtensor.py
index b57b3d85bd..ac6c46bc46 100644
--- a/bittensor/core/subtensor.py
+++ b/bittensor/core/subtensor.py
@@ -39,11 +39,13 @@
from bittensor.core import settings
from bittensor.core.axon import Axon
from bittensor.core.chain_data import (
+ custom_rpc_type_registry,
+ DelegateInfo,
NeuronInfo,
+ NeuronInfoLite,
PrometheusInfo,
SubnetHyperparameters,
- NeuronInfoLite,
- custom_rpc_type_registry,
+ SubnetInfo,
)
from bittensor.core.config import Config
from bittensor.core.extrinsics.commit_weights import (
@@ -58,6 +60,10 @@
burned_register_extrinsic,
register_extrinsic,
)
+from bittensor.core.extrinsics.root import (
+ root_register_extrinsic,
+ set_root_weights_extrinsic,
+)
from bittensor.core.extrinsics.serving import (
do_serve_axon,
serve_axon_extrinsic,
@@ -69,10 +75,10 @@
transfer_extrinsic,
)
from bittensor.core.metagraph import Metagraph
-from bittensor.utils import torch
-from bittensor.utils import u16_normalized_float, networking
+from bittensor.utils import networking, torch, ss58_to_vec_u8, u16_normalized_float
from bittensor.utils.balance import Balance
from bittensor.utils.btlogging import logging
+from bittensor.utils.registration import legacy_torch_api_compat
from bittensor.utils.weight_utils import generate_weight_hash
KEY_NONCE: dict[str, int] = {}
@@ -902,6 +908,45 @@ def set_weights(
return success, message
+ @legacy_torch_api_compat
+ def root_set_weights(
+ self,
+ wallet: "Wallet",
+ netuids: Union[NDArray[np.int64], "torch.LongTensor", list],
+ weights: Union[NDArray[np.float32], "torch.FloatTensor", list],
+ version_key: int = 0,
+ wait_for_inclusion: bool = False,
+ wait_for_finalization: bool = False,
+ prompt: bool = False,
+ ) -> bool:
+ """
+ Sets the weights for neurons on the root network. This action is crucial for defining the influence and interactions of neurons at the root level of the Bittensor network.
+
+ Args:
+ wallet (bittensor_wallet.Wallet): The wallet associated with the neuron setting the weights.
+ netuids (Union[NDArray[np.int64], torch.LongTensor, list]): The list of neuron UIDs for which weights are being set.
+ weights (Union[NDArray[np.float32], torch.FloatTensor, list]): The corresponding weights to be set for each UID.
+ version_key (int, optional): Version key for compatibility with the network. Default is ``0``.
+ wait_for_inclusion (bool, optional): Waits for the transaction to be included in a block. Defaults to ``False``.
+ wait_for_finalization (bool, optional): Waits for the transaction to be finalized on the blockchain. Defaults to ``False``.
+ prompt (bool, optional): If ``True``, prompts for user confirmation before proceeding. Defaults to ``False``.
+
+ Returns:
+ bool: ``True`` if the setting of root-level weights is successful, False otherwise.
+
+ This function plays a pivotal role in shaping the root network's collective intelligence and decision-making processes, reflecting the principles of decentralized governance and collaborative learning in Bittensor.
+ """
+ return set_root_weights_extrinsic(
+ subtensor=self,
+ wallet=wallet,
+ netuids=netuids,
+ weights=weights,
+ version_key=version_key,
+ wait_for_inclusion=wait_for_inclusion,
+ wait_for_finalization=wait_for_finalization,
+ prompt=prompt,
+ )
+
def register(
self,
wallet: "Wallet",
@@ -961,6 +1006,35 @@ def register(
log_verbose=log_verbose,
)
+ def root_register(
+ self,
+ wallet: "Wallet",
+ wait_for_inclusion: bool = False,
+ wait_for_finalization: bool = True,
+ prompt: bool = False,
+ ) -> bool:
+ """
+ Registers the neuron associated with the wallet on the root network. This process is integral for participating in the highest layer of decision-making and governance within the Bittensor network.
+
+ Args:
+ wallet (bittensor.wallet): The wallet associated with the neuron to be registered on the root network.
+ wait_for_inclusion (bool): Waits for the transaction to be included in a block. Defaults to `False`.
+ wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Defaults to `True`.
+ prompt (bool): If ``True``, prompts for user confirmation before proceeding. Defaults to `False`.
+
+ Returns:
+ bool: ``True`` if the registration on the root network is successful, False otherwise.
+
+ This function enables neurons to engage in the most critical and influential aspects of the network's governance, signifying a high level of commitment and responsibility in the Bittensor ecosystem.
+ """
+ return root_register_extrinsic(
+ subtensor=self,
+ wallet=wallet,
+ wait_for_inclusion=wait_for_inclusion,
+ wait_for_finalization=wait_for_finalization,
+ prompt=prompt,
+ )
+
def burned_register(
self,
wallet: "Wallet",
@@ -1419,6 +1493,36 @@ def subnet_exists(self, netuid: int, block: Optional[int] = None) -> bool:
_result = self.query_subtensor("NetworksAdded", block, [netuid])
return getattr(_result, "value", False)
+ @networking.ensure_connected
+ def get_all_subnets_info(self, block: Optional[int] = None) -> list[SubnetInfo]:
+ """
+ Retrieves detailed information about all subnets within the Bittensor network. This function provides comprehensive data on each subnet, including its characteristics and operational parameters.
+
+ Args:
+ block (Optional[int]): The blockchain block number for the query.
+
+ Returns:
+ list[SubnetInfo]: A list of SubnetInfo objects, each containing detailed information about a subnet.
+
+ Gaining insights into the subnets' details assists in understanding the network's composition, the roles of different subnets, and their unique features.
+ """
+
+ @retry(delay=1, tries=3, backoff=2, max_delay=4)
+ def make_substrate_call_with_retry():
+ block_hash = None if block is None else self.substrate.get_block_hash(block)
+
+ return self.substrate.rpc_request(
+ method="subnetInfo_getSubnetsInfo", # custom rpc method
+ params=[block_hash] if block_hash else [],
+ )
+
+ json_body = make_substrate_call_with_retry()
+
+ if not (result := json_body.get("result", None)):
+ return []
+
+ return SubnetInfo.list_from_vec_u8(result)
+
# Metagraph uses this method
def bonds(
self, netuid: int, block: Optional[int] = None
@@ -1885,6 +1989,64 @@ def recycle(self, netuid: int, block: Optional[int] = None) -> Optional["Balance
call = self._get_hyperparameter(param_name="Burn", netuid=netuid, block=block)
return None if call is None else Balance.from_rao(int(call))
+ def get_delegate_take(
+ self, hotkey_ss58: str, block: Optional[int] = None
+ ) -> Optional[float]:
+ """
+ Retrieves the delegate 'take' percentage for a neuron identified by its hotkey. The 'take' represents the percentage of rewards that the delegate claims from its nominators' stakes.
+
+ Args:
+ hotkey_ss58 (str): The ``SS58`` address of the neuron's hotkey.
+ block (Optional[int]): The blockchain block number for the query.
+
+ Returns:
+ Optional[float]: The delegate take percentage, None if not available.
+
+ The delegate take is a critical parameter in the network's incentive structure, influencing the distribution of rewards among neurons and their nominators.
+ """
+ _result = self.query_subtensor("Delegates", block, [hotkey_ss58])
+ return (
+ None
+ if getattr(_result, "value", None) is None
+ else u16_normalized_float(_result.value)
+ )
+
+ @networking.ensure_connected
+ def get_delegate_by_hotkey(
+ self, hotkey_ss58: str, block: Optional[int] = None
+ ) -> Optional[DelegateInfo]:
+ """
+ Retrieves detailed information about a delegate neuron based on its hotkey. This function provides a comprehensive view of the delegate's status, including its stakes, nominators, and reward distribution.
+
+ Args:
+ hotkey_ss58 (str): The ``SS58`` address of the delegate's hotkey.
+ block (Optional[int]): The blockchain block number for the query. Default is ``None``.
+
+ Returns:
+ Optional[DelegateInfo]: Detailed information about the delegate neuron, ``None`` if not found.
+
+ This function is essential for understanding the roles and influence of delegate neurons within the Bittensor network's consensus and governance structures.
+ """
+
+ @retry(delay=1, tries=3, backoff=2, max_delay=4)
+ def make_substrate_call_with_retry(encoded_hotkey_: list[int]):
+ block_hash = None if block is None else self.substrate.get_block_hash(block)
+
+ return self.substrate.rpc_request(
+ method="delegateInfo_getDelegate", # custom rpc method
+ params=(
+ [encoded_hotkey_, block_hash] if block_hash else [encoded_hotkey_]
+ ),
+ )
+
+ encoded_hotkey = ss58_to_vec_u8(hotkey_ss58)
+ json_body = make_substrate_call_with_retry(encoded_hotkey)
+
+ if not (result := json_body.get("result", None)):
+ return None
+
+ return DelegateInfo.from_vec_u8(result)
+
# Subnet 27 uses this method
_do_serve_prometheus = do_serve_prometheus
# Subnet 27 uses this method name
diff --git a/tests/unit_tests/extrinsics/test_root.py b/tests/unit_tests/extrinsics/test_root.py
new file mode 100644
index 0000000000..bd37be203f
--- /dev/null
+++ b/tests/unit_tests/extrinsics/test_root.py
@@ -0,0 +1,309 @@
+import pytest
+from bittensor.core.subtensor import Subtensor
+from bittensor.core.extrinsics import root
+
+
+@pytest.fixture
+def mock_subtensor(mocker):
+ mock = mocker.MagicMock(spec=Subtensor)
+ mock.network = "magic_mock"
+ return mock
+
+
+@pytest.fixture
+def mock_wallet(mocker):
+ mock = mocker.MagicMock()
+ mock.hotkey.ss58_address = "fake_hotkey_address"
+ return mock
+
+
+@pytest.mark.parametrize(
+ "wait_for_inclusion, wait_for_finalization, hotkey_registered, registration_success, prompt, user_response, expected_result",
+ [
+ (
+ False,
+ True,
+ [True, None],
+ True,
+ True,
+ True,
+ True,
+ ), # Already registered after attempt
+ (
+ False,
+ True,
+ [False, True],
+ True,
+ True,
+ True,
+ True,
+ ), # Registration succeeds with user confirmation
+ (False, True, [False, False], False, False, None, None), # Registration fails
+ (
+ False,
+ True,
+ [False, False],
+ True,
+ False,
+ None,
+ None,
+ ), # Registration succeeds but neuron not found
+ (
+ False,
+ True,
+ [False, False],
+ True,
+ True,
+ False,
+ False,
+ ), # User declines registration
+ ],
+ ids=[
+ "success-already-registered",
+ "success-registration-succeeds",
+ "failure-registration-failed",
+ "failure-neuron-not-found",
+ "failure-prompt-declined",
+ ],
+)
+def test_root_register_extrinsic(
+ mock_subtensor,
+ mock_wallet,
+ wait_for_inclusion,
+ wait_for_finalization,
+ hotkey_registered,
+ registration_success,
+ prompt,
+ user_response,
+ expected_result,
+ mocker,
+):
+ # Arrange
+ mock_subtensor.is_hotkey_registered.side_effect = hotkey_registered
+
+ with mocker.patch("rich.prompt.Confirm.ask", return_value=user_response):
+ # Preps
+ mock_register = mocker.Mock(
+ return_value=(registration_success, "Error registering")
+ )
+ root._do_root_register = mock_register
+
+ # Act
+ result = root.root_register_extrinsic(
+ subtensor=mock_subtensor,
+ wallet=mock_wallet,
+ wait_for_inclusion=wait_for_inclusion,
+ wait_for_finalization=wait_for_finalization,
+ prompt=prompt,
+ )
+ # Assert
+ assert result == expected_result
+
+ if not hotkey_registered[0] and user_response:
+ mock_register.assert_called_once()
+
+
+@pytest.mark.parametrize(
+ "wait_for_inclusion, wait_for_finalization, netuids, weights, prompt, user_response, expected_success",
+ [
+ (True, False, [1, 2], [0.5, 0.5], True, True, True), # Success - weights set
+ (
+ False,
+ False,
+ [1, 2],
+ [0.5, 0.5],
+ False,
+ None,
+ True,
+ ), # Success - weights set no wait
+ (
+ True,
+ False,
+ [1, 2],
+ [2000, 20],
+ True,
+ True,
+ True,
+ ), # Success - large value to be normalized
+ (
+ True,
+ False,
+ [1, 2],
+ [2000, 0],
+ True,
+ True,
+ True,
+ ), # Success - single large value
+ (
+ True,
+ False,
+ [1, 2],
+ [0.5, 0.5],
+ True,
+ False,
+ False,
+ ), # Failure - prompt declined
+ (
+ True,
+ False,
+ [1, 2],
+ [0.5, 0.5],
+ False,
+ None,
+ False,
+ ), # Failure - setting weights failed
+ (
+ True,
+ False,
+ [],
+ [],
+ None,
+ False,
+ False,
+ ), # Exception catched - ValueError 'min() arg is an empty sequence'
+ ],
+ ids=[
+ "success-weights-set",
+ "success-not-wait",
+ "success-large-value",
+ "success-single-value",
+ "failure-user-declines",
+ "failure-setting-weights",
+ "failure-value-error-exception",
+ ],
+)
+def test_set_root_weights_extrinsic(
+ mock_subtensor,
+ mock_wallet,
+ wait_for_inclusion,
+ wait_for_finalization,
+ netuids,
+ weights,
+ prompt,
+ user_response,
+ expected_success,
+ mocker,
+):
+ # Preps
+ root._do_set_root_weights = mocker.Mock(
+ return_value=(expected_success, "Mock error")
+ )
+ mock_subtensor.min_allowed_weights = mocker.Mock(return_value=0)
+ mock_subtensor.max_weight_limit = mocker.Mock(return_value=1)
+ mock_confirm = mocker.Mock(return_value=(expected_success, "Mock error"))
+ root.Confirm.ask = mock_confirm
+
+ # Call
+ result = root.set_root_weights_extrinsic(
+ subtensor=mock_subtensor,
+ wallet=mock_wallet,
+ netuids=netuids,
+ weights=weights,
+ version_key=0,
+ wait_for_inclusion=wait_for_inclusion,
+ wait_for_finalization=wait_for_finalization,
+ prompt=prompt,
+ )
+
+ # Asserts
+ assert result == expected_success
+ if prompt:
+ mock_confirm.assert_called_once()
+ else:
+ mock_confirm.assert_not_called()
+
+
+@pytest.mark.parametrize(
+ "wait_for_inclusion, wait_for_finalization, netuids, weights, prompt, user_response, expected_success",
+ [
+ (True, False, [1, 2], [0.5, 0.5], True, True, True), # Success - weights set
+ (
+ False,
+ False,
+ [1, 2],
+ [0.5, 0.5],
+ False,
+ None,
+ True,
+ ), # Success - weights set no wait
+ (
+ True,
+ False,
+ [1, 2],
+ [2000, 20],
+ True,
+ True,
+ True,
+ ), # Success - large value to be normalized
+ (
+ True,
+ False,
+ [1, 2],
+ [2000, 0],
+ True,
+ True,
+ True,
+ ), # Success - single large value
+ (
+ True,
+ False,
+ [1, 2],
+ [0.5, 0.5],
+ True,
+ False,
+ False,
+ ), # Failure - prompt declined
+ (
+ True,
+ False,
+ [1, 2],
+ [0.5, 0.5],
+ False,
+ None,
+ False,
+ ), # Failure - setting weights failed
+ (
+ True,
+ False,
+ [],
+ [],
+ None,
+ False,
+ False,
+ ), # Exception catched - ValueError 'min() arg is an empty sequence'
+ ],
+ ids=[
+ "success-weights-set",
+ "success-not-wait",
+ "success-large-value",
+ "success-single-value",
+ "failure-user-declines",
+ "failure-setting-weights",
+ "failure-value-error-exception",
+ ],
+)
+def test_set_root_weights_extrinsic_torch(
+ mock_subtensor,
+ mock_wallet,
+ wait_for_inclusion,
+ wait_for_finalization,
+ netuids,
+ weights,
+ prompt,
+ user_response,
+ expected_success,
+ force_legacy_torch_compatible_api,
+ mocker,
+):
+ test_set_root_weights_extrinsic(
+ mock_subtensor,
+ mock_wallet,
+ wait_for_inclusion,
+ wait_for_finalization,
+ netuids,
+ weights,
+ prompt,
+ user_response,
+ expected_success,
+ mocker,
+ )
diff --git a/tests/unit_tests/test_chain_data.py b/tests/unit_tests/test_chain_data.py
index 353f697d46..65232e3382 100644
--- a/tests/unit_tests/test_chain_data.py
+++ b/tests/unit_tests/test_chain_data.py
@@ -364,116 +364,3 @@ def create_neuron_info_decoded(
"prometheus_info": prometheus_info,
"axon_info": axon_info,
}
-
-
-@pytest.fixture
-def mock_from_scale_encoding(mocker):
- return mocker.patch("bittensor.core.chain_data.delegate_info.from_scale_encoding")
-
-
-@pytest.fixture
-def mock_fix_decoded_values(mocker):
- return mocker.patch(
- "bittensor.core.chain_data.DelegateInfo.fix_decoded_values",
- side_effect=lambda x: x,
- )
-
-
-@pytest.mark.parametrize(
- "test_id, vec_u8, expected",
- [
- (
- "happy-path-1",
- [1, 2, 3],
- [
- DelegateInfo(
- hotkey_ss58="hotkey",
- total_stake=1000,
- nominators=[
- "nominator1",
- "nominator2",
- ],
- owner_ss58="owner",
- take=10.1,
- validator_permits=[1, 2, 3],
- registrations=[4, 5, 6],
- return_per_1000=100,
- total_daily_return=1000,
- )
- ],
- ),
- (
- "happy-path-2",
- [4, 5, 6],
- [
- DelegateInfo(
- hotkey_ss58="hotkey",
- total_stake=1000,
- nominators=[
- "nominator1",
- "nominator2",
- ],
- owner_ss58="owner",
- take=2.1,
- validator_permits=[1, 2, 3],
- registrations=[4, 5, 6],
- return_per_1000=100,
- total_daily_return=1000,
- )
- ],
- ),
- ],
-)
-def test_list_from_vec_u8_happy_path(
- mock_from_scale_encoding, mock_fix_decoded_values, test_id, vec_u8, expected
-):
- # Arrange
- mock_from_scale_encoding.return_value = expected
-
- # Act
- result = DelegateInfo.list_from_vec_u8(vec_u8)
-
- # Assert
- mock_from_scale_encoding.assert_called_once_with(
- vec_u8, ChainDataType.DelegateInfo, is_vec=True
- )
- assert result == expected, f"Failed {test_id}"
-
-
-@pytest.mark.parametrize(
- "test_id, vec_u8, expected",
- [
- ("edge_empty_list", [], []),
- ],
-)
-def test_list_from_vec_u8_edge_cases(
- mock_from_scale_encoding, mock_fix_decoded_values, test_id, vec_u8, expected
-):
- # Arrange
- mock_from_scale_encoding.return_value = None
-
- # Act
- result = DelegateInfo.list_from_vec_u8(vec_u8)
-
- # Assert
- mock_from_scale_encoding.assert_called_once_with(
- vec_u8, ChainDataType.DelegateInfo, is_vec=True
- )
- assert result == expected, f"Failed {test_id}"
-
-
-@pytest.mark.parametrize(
- "vec_u8, expected_exception",
- [
- ("not_a_list", TypeError),
- ],
-)
-def test_list_from_vec_u8_error_cases(
- vec_u8,
- expected_exception,
-):
- # No Arrange section needed as input values are provided via test parameters
-
- # Act & Assert
- with pytest.raises(expected_exception):
- _ = DelegateInfo.list_from_vec_u8(vec_u8)
diff --git a/tests/unit_tests/test_subtensor.py b/tests/unit_tests/test_subtensor.py
index bc1ea360c6..6d8fb1ff5f 100644
--- a/tests/unit_tests/test_subtensor.py
+++ b/tests/unit_tests/test_subtensor.py
@@ -2181,3 +2181,126 @@ def test_recycle_none(subtensor, mocker):
)
assert result is None
+
+
+# `get_all_subnets_info` tests
+def test_get_all_subnets_info_success(mocker, subtensor):
+ """Test get_all_subnets_info returns correct data when subnet information is found."""
+ # Prep
+ block = 123
+ subnet_data = [1, 2, 3] # Mocked response data
+ mocker.patch.object(
+ subtensor.substrate, "get_block_hash", return_value="mock_block_hash"
+ )
+ mock_response = {"result": subnet_data}
+ mocker.patch.object(subtensor.substrate, "rpc_request", return_value=mock_response)
+ mocker.patch.object(
+ subtensor_module.SubnetInfo,
+ "list_from_vec_u8",
+ return_value="list_from_vec_u80",
+ )
+
+ # Call
+ result = subtensor.get_all_subnets_info(block)
+
+ # Asserts
+ subtensor.substrate.get_block_hash.assert_called_once_with(block)
+ subtensor.substrate.rpc_request.assert_called_once_with(
+ method="subnetInfo_getSubnetsInfo", params=["mock_block_hash"]
+ )
+ subtensor_module.SubnetInfo.list_from_vec_u8.assert_called_once_with(subnet_data)
+
+
+@pytest.mark.parametrize("result_", [[], None])
+def test_get_all_subnets_info_no_data(mocker, subtensor, result_):
+ """Test get_all_subnets_info returns empty list when no subnet information is found."""
+ # Prep
+ block = 123
+ mocker.patch.object(
+ subtensor.substrate, "get_block_hash", return_value="mock_block_hash"
+ )
+ mock_response = {"result": result_}
+ mocker.patch.object(subtensor.substrate, "rpc_request", return_value=mock_response)
+ mocker.patch.object(subtensor_module.SubnetInfo, "list_from_vec_u8")
+
+ # Call
+ result = subtensor.get_all_subnets_info(block)
+
+ # Asserts
+ assert result == []
+ subtensor.substrate.get_block_hash.assert_called_once_with(block)
+ subtensor.substrate.rpc_request.assert_called_once_with(
+ method="subnetInfo_getSubnetsInfo", params=["mock_block_hash"]
+ )
+ subtensor_module.SubnetInfo.list_from_vec_u8.assert_not_called()
+
+
+def test_get_all_subnets_info_retry(mocker, subtensor):
+ """Test get_all_subnets_info retries on failure."""
+ # Prep
+ block = 123
+ subnet_data = [1, 2, 3]
+ mocker.patch.object(
+ subtensor.substrate, "get_block_hash", return_value="mock_block_hash"
+ )
+ mock_response = {"result": subnet_data}
+ mock_rpc_request = mocker.patch.object(
+ subtensor.substrate,
+ "rpc_request",
+ side_effect=[Exception, Exception, mock_response],
+ )
+ mocker.patch.object(
+ subtensor_module.SubnetInfo, "list_from_vec_u8", return_value=["some_data"]
+ )
+
+ # Call
+ result = subtensor.get_all_subnets_info(block)
+
+ # Asserts
+ subtensor.substrate.get_block_hash.assert_called_with(block)
+ assert mock_rpc_request.call_count == 3
+ subtensor_module.SubnetInfo.list_from_vec_u8.assert_called_once_with(subnet_data)
+ assert result == ["some_data"]
+
+
+def test_get_delegate_take_success(subtensor, mocker):
+ """Verify `get_delegate_take` method successful path."""
+ # Preps
+ fake_hotkey_ss58 = "FAKE_SS58"
+ fake_block = 123
+
+ subtensor_module.u16_normalized_float = mocker.Mock()
+ subtensor.query_subtensor = mocker.Mock(return_value=mocker.Mock(value="value"))
+
+ # Call
+ result = subtensor.get_delegate_take(hotkey_ss58=fake_hotkey_ss58, block=fake_block)
+
+ # Asserts
+ subtensor.query_subtensor.assert_called_once_with(
+ "Delegates", fake_block, [fake_hotkey_ss58]
+ )
+ subtensor_module.u16_normalized_float.assert_called_once_with(
+ subtensor.query_subtensor.return_value.value
+ )
+ assert result == subtensor_module.u16_normalized_float.return_value
+
+
+def test_get_delegate_take_none(subtensor, mocker):
+ """Verify `get_delegate_take` method returns None."""
+ # Preps
+ fake_hotkey_ss58 = "FAKE_SS58"
+ fake_block = 123
+
+ subtensor.query_subtensor = mocker.Mock(return_value=mocker.Mock(value=None))
+ subtensor_module.u16_normalized_float = mocker.Mock()
+
+ # Call
+ result = subtensor.get_delegate_take(hotkey_ss58=fake_hotkey_ss58, block=fake_block)
+
+ # Asserts
+ subtensor.query_subtensor.assert_called_once_with(
+ "Delegates", fake_block, [fake_hotkey_ss58]
+ )
+
+ subtensor_module.u16_normalized_float.assert_not_called()
+ assert result is None