added to Subtensor: burned_register, get_subnet_burn_cost, recycle and related extrinsics

bittensor/core/extrinsics/registration.py

@@ -18,6 +18,7 @@
 import time
 from typing import Union, Optional, TYPE_CHECKING
 
+from bittensor_wallet.errors import KeyFileError
 from retry import retry
 from rich.prompt import Confirm
 
@@ -285,3 +286,166 @@ def register_extrinsic(
             # Failed to register after max attempts.
             bt_console.print("[red]No more attempts.[/red]")
             return False
+
+
+@ensure_connected
+def _do_burned_register(
+    self,
+    netuid: int,
+    wallet: "Wallet",
+    wait_for_inclusion: bool = False,
+    wait_for_finalization: bool = True,
+) -> tuple[bool, Optional[str]]:
+    """
+    Performs a burned register extrinsic call to the Subtensor chain.
+
+    This method sends a registration transaction to the Subtensor blockchain using the burned register mechanism. It
+    retries the call up to three times with exponential backoff in case of failures.
+
+    Args:
+        self (bittensor.core.subtensor.Subtensor): Subtensor instance.
+        netuid (int): The network unique identifier to register on.
+        wallet (bittensor_wallet.Wallet): The wallet to be registered.
+        wait_for_inclusion (bool): Whether to wait for the transaction to be included in a block. Default is False.
+        wait_for_finalization (bool): Whether to wait for the transaction to be finalized. Default is True.
+
+    Returns:
+        Tuple[bool, Optional[str]]: A tuple containing a boolean indicating success or failure, and an optional error message.
+    """
+
+    @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="burned_register",
+            call_params={
+                "netuid": netuid,
+                "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, None
+
+        # 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 burned_register_extrinsic(
+    subtensor: "Subtensor",
+    wallet: "Wallet",
+    netuid: int,
+    wait_for_inclusion: bool = False,
+    wait_for_finalization: bool = True,
+    prompt: bool = False,
+) -> bool:
+    """Registers the wallet to chain by recycling TAO.
+
+    Args:
+        subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance.
+        wallet (bittensor.wallet): Bittensor wallet object.
+        netuid (int): The ``netuid`` of the subnet to register on.
+        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.
+        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.
+        prompt (bool): If ``true``, the call waits for confirmation from the user before proceeding.
+
+    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``.
+    """
+    if not subtensor.subnet_exists(netuid):
+        bt_console.print(
+            ":cross_mark: [red]Failed[/red]: error: [bold white]subnet:{}[/bold white] does not exist.".format(
+                netuid
+            )
+        )
+        return False
+
+    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
+    with bt_console.status(
+        f":satellite: Checking Account on [bold]subnet:{netuid}[/bold]..."
+    ):
+        neuron = subtensor.get_neuron_for_pubkey_and_subnet(
+            wallet.hotkey.ss58_address, netuid=netuid
+        )
+
+        old_balance = subtensor.get_balance(wallet.coldkeypub.ss58_address)
+
+        recycle_amount = subtensor.recycle(netuid=netuid)
+        if not neuron.is_null:
+            bt_console.print(
+                ":white_heavy_check_mark: [green]Already Registered[/green]:\n"
+                "uid: [bold white]{}[/bold white]\n"
+                "netuid: [bold white]{}[/bold white]\n"
+                "hotkey: [bold white]{}[/bold white]\n"
+                "coldkey: [bold white]{}[/bold white]".format(
+                    neuron.uid, neuron.netuid, neuron.hotkey, neuron.coldkey
+                )
+            )
+            return True
+
+    if prompt:
+        # Prompt user for confirmation.
+        if not Confirm.ask(f"Recycle {recycle_amount} to register on subnet:{netuid}?"):
+            return False
+
+    with bt_console.status(":satellite: Recycling TAO for Registration..."):
+        success, err_msg = _do_burned_register(
+            self=subtensor,
+            netuid=netuid,
+            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)
+            return False
+        # Successful registration, final check for neuron and pubkey
+        else:
+            bt_console.print(":satellite: Checking Balance...")
+            block = subtensor.get_current_block()
+            new_balance = subtensor.get_balance(
+                wallet.coldkeypub.ss58_address, block=block
+            )
+
+            bt_console.print(
+                "Balance:\n  [blue]{}[/blue] :arrow_right: [green]{}[/green]".format(
+                    old_balance, new_balance
+                )
+            )
+            is_registered = subtensor.is_hotkey_registered(
+                netuid=netuid, 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]"
+                )
+                return False

bittensor/core/subtensor.py

@@ -54,7 +54,10 @@
     do_serve_prometheus,
     prometheus_extrinsic,
 )
-from bittensor.core.extrinsics.registration import register_extrinsic
+from bittensor.core.extrinsics.registration import (
+    burned_register_extrinsic,
+    register_extrinsic,
+)
 from bittensor.core.extrinsics.serving import (
     do_serve_axon,
     serve_axon_extrinsic,
@@ -958,6 +961,36 @@ def register(
             log_verbose=log_verbose,
         )
 
+    def burned_register(
+        self,
+        wallet: "Wallet",
+        netuid: int,
+        wait_for_inclusion: bool = False,
+        wait_for_finalization: bool = True,
+        prompt: bool = False,
+    ) -> bool:
+        """
+        Registers a neuron on the Bittensor network by recycling TAO. This method of registration involves recycling TAO tokens, allowing them to be re-mined by performing work on the network.
+
+        Args:
+            wallet (bittensor_wallet.Wallet): The wallet associated with the neuron to be registered.
+            netuid (int): The unique identifier of the subnet.
+            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 `True`.
+            prompt (bool, optional): If ``True``, prompts for user confirmation before proceeding. Defaults to `False`.
+
+        Returns:
+            bool: ``True`` if the registration is successful, False otherwise.
+        """
+        return burned_register_extrinsic(
+            subtensor=self,
+            wallet=wallet,
+            netuid=netuid,
+            wait_for_inclusion=wait_for_inclusion,
+            wait_for_finalization=wait_for_finalization,
+            prompt=prompt,
+        )
+
     def serve_axon(
         self,
         netuid: int,
@@ -1412,6 +1445,30 @@ def bonds(
 
         return b_map
 
+    def get_subnet_burn_cost(self, block: Optional[int] = None) -> Optional[str]:
+        """
+        Retrieves the burn cost for registering a new subnet within the Bittensor network. This cost represents the amount of Tao that needs to be locked or burned to establish a new subnet.
+
+        Args:
+            block (Optional[int]): The blockchain block number for the query.
+
+        Returns:
+            int: The burn cost for subnet registration.
+
+        The subnet burn cost is an important economic parameter, reflecting the network's mechanisms for controlling the proliferation of subnets and ensuring their commitment to the network's long-term viability.
+        """
+        lock_cost = self.query_runtime_api(
+            runtime_api="SubnetRegistrationRuntimeApi",
+            method="get_network_registration_cost",
+            params=[],
+            block=block,
+        )
+
+        if lock_cost is None:
+            return None
+
+        return lock_cost
+
     # Metagraph uses this method
     def neurons(self, netuid: int, block: Optional[int] = None) -> list["NeuronInfo"]:
         """
@@ -1812,6 +1869,22 @@ def difficulty(self, netuid: int, block: Optional[int] = None) -> Optional[int]:
             return None
         return int(call)
 
+    def recycle(self, netuid: int, block: Optional[int] = None) -> Optional["Balance"]:
+        """
+        Retrieves the 'Burn' hyperparameter for a specified subnet. The 'Burn' parameter represents the amount of Tao that is effectively recycled within the Bittensor network.
+
+        Args:
+            netuid (int): The unique identifier of the subnet.
+            block (Optional[int]): The blockchain block number for the query.
+
+        Returns:
+            Optional[Balance]: The value of the 'Burn' hyperparameter if the subnet exists, None otherwise.
+
+        Understanding the 'Burn' rate is essential for analyzing the network registration usage, particularly how it is correlated with user activity and the overall cost of participation in a given subnet.
+        """
+        call = self._get_hyperparameter(param_name="Burn", netuid=netuid, block=block)
+        return None if call is None else Balance.from_rao(int(call))
+
     # Subnet 27 uses this method
     _do_serve_prometheus = do_serve_prometheus
     # Subnet 27 uses this method name

tests/unit_tests/extrinsics/test_registration.py

@@ -18,9 +18,7 @@
 import pytest
 from bittensor_wallet import Wallet
 
-from bittensor.core.extrinsics.registration import (
-    register_extrinsic,
-)
+from bittensor.core.extrinsics import registration
 from bittensor.core.subtensor import Subtensor
 from bittensor.utils.registration import POWSolution
 
@@ -95,7 +93,7 @@ def test_register_extrinsic_without_pow(
         "rich.prompt.Confirm.ask", return_value=prompt_response
     ), mocker.patch("torch.cuda.is_available", return_value=cuda_available):
         # Act
-        result = register_extrinsic(
+        result = registration.register_extrinsic(
             subtensor=mock_subtensor,
             wallet=mock_wallet,
             netuid=123,
@@ -160,7 +158,7 @@ def test_register_extrinsic_with_pow(
                 return_value=hotkey_registered
             )
 
-        result = register_extrinsic(
+        result = registration.register_extrinsic(
             subtensor=mock_subtensor,
             wallet=mock_wallet,
             netuid=123,
@@ -179,3 +177,57 @@ def test_register_extrinsic_with_pow(
 
         # Assert
         assert result == expected_result, f"Test failed for test_id: {test_id}."
+
+
+@pytest.mark.parametrize(
+    "subnet_exists, neuron_is_null, recycle_success, prompt, prompt_response, is_registered, expected_result, test_id",
+    [
+        # Happy paths
+        (True, False, None, False, None, None, True, "neuron-not-null"),
+        (True, True, True, True, True, True, True, "happy-path-wallet-registered"),
+        # Error paths
+        (False, True, False, False, None, None, False, "subnet-non-existence"),
+        (True, True, True, True, False, None, False, "prompt-declined"),
+        (True, True, False, True, True, False, False, "error-path-recycling-failed"),
+        (True, True, True, True, True, False, False, "error-path-not-registered"),
+    ],
+)
+def test_burned_register_extrinsic(
+    mock_subtensor,
+    mock_wallet,
+    subnet_exists,
+    neuron_is_null,
+    recycle_success,
+    prompt,
+    prompt_response,
+    is_registered,
+    expected_result,
+    test_id,
+    mocker,
+):
+    # Arrange
+    with mocker.patch.object(
+        mock_subtensor, "subnet_exists", return_value=subnet_exists
+    ), mocker.patch.object(
+        mock_subtensor,
+        "get_neuron_for_pubkey_and_subnet",
+        return_value=mocker.MagicMock(is_null=neuron_is_null),
+    ), mocker.patch(
+        "bittensor.core.extrinsics.registration._do_burned_register",
+        return_value=(recycle_success, "Mock error message"),
+    ), mocker.patch.object(
+        mock_subtensor, "is_hotkey_registered", return_value=is_registered
+    ):
+        mock_confirm = mocker.MagicMock(return_value=prompt_response)
+        registration.Confirm.ask = mock_confirm
+        # Act
+        result = registration.burned_register_extrinsic(
+            subtensor=mock_subtensor, wallet=mock_wallet, netuid=123, prompt=prompt
+        )
+        # Assert
+        assert result == expected_result, f"Test failed for test_id: {test_id}"
+
+        if prompt:
+            mock_confirm.assert_called_once()
+        else:
+            mock_confirm.assert_not_called()