Merge pull request #2114 from opentensor/feat/arbitrage-coldkeys-re-add

Coldkey arbitration period info updates

bittensor/commands/schedule_coldkey_swap.py

@@ -27,6 +27,32 @@
 console = bittensor.__console__
 
 
+def fetch_arbitration_stats(subtensor, wallet):
+    """
+    Performs a check of the current arbitration data (if any), and displays it through the bittensor console.
+    """
+    arbitration_check = len(subtensor.check_in_arbitration(wallet.coldkey.ss58_address))
+    if arbitration_check == 0:
+        bittensor.__console__.print(
+            "[green]There has been no previous key swap initiated for your coldkey.[/green]"
+        )
+    if arbitration_check == 1:
+        arbitration_remaining = subtensor.get_remaining_arbitration_period(
+            wallet.coldkey.ss58_address
+        )
+        hours, minutes, seconds = convert_blocks_to_time(arbitration_remaining)
+        bittensor.__console__.print(
+            "[yellow]There has been 1 swap request made for this coldkey already."
+            " By adding another swap request, the key will enter arbitration."
+            f" Your key swap is scheduled for {hours} hours, {minutes} minutes, {seconds} seconds"
+            " from now.[/yellow]"
+        )
+    if arbitration_check > 1:
+        bittensor.__console__.print(
+            f"[red]This coldkey is currently in arbitration with a total swaps of {arbitration_check}.[/red]"
+        )
+
+
 class ScheduleColdKeySwapCommand:
     """
     Executes the ``schedule_coldkey_swap`` command to schedule a coldkey swap on the Bittensor network.
@@ -86,7 +112,7 @@ def _run(cli: "bittensor.cli", subtensor: "bittensor.subtensor"):
             "[yellow]If you call this on the same key multiple times, the key will enter arbitration.[/yellow]"
         )
 
-        ScheduleColdKeySwapCommand.fetch_arbitration_stats(subtensor, wallet)
+        fetch_arbitration_stats(subtensor, wallet)
 
         # Get the values for the command
         if not cli.config.is_set("new_coldkey"):
@@ -221,25 +247,6 @@ def add_args(command_parser: argparse.ArgumentParser):
         bittensor.wallet.add_args(schedule_coldkey_swap_parser)
         bittensor.subtensor.add_args(schedule_coldkey_swap_parser)
 
-    @staticmethod
-    def fetch_arbitration_stats(subtensor, wallet):
-        arbitration_check = len(
-            subtensor.check_in_arbitration(wallet.coldkey.ss58_address)
-        )
-        if arbitration_check == 0:
-            bittensor.__console__.print(
-                "[green]There has been no previous key swap initiated for your coldkey.[/green]"
-            )
-        if arbitration_check == 1:
-            bittensor.__console__.print(
-                "[yellow]There has been 1 swap request made for this coldkey already."
-                " By adding another swap request, the key will enter arbitration.[/yellow]"
-            )
-        if arbitration_check > 1:
-            bittensor.__console__.print(
-                f"[red]This coldkey is currently in arbitration with a total swaps of {arbitration_check}.[/red]"
-            )
-
 
 class CheckColdKeySwapCommand:
     """
@@ -289,31 +296,7 @@ def _run(cli: "bittensor.cli", subtensor: "bittensor.subtensor"):
         config = cli.config.copy()
         wallet = bittensor.wallet(config=config)
 
-        CheckColdKeySwapCommand.fetch_arbitration_stats(subtensor, wallet)
-
-    @staticmethod
-    def fetch_arbitration_stats(subtensor, wallet):
-        arbitration_check = len(
-            subtensor.check_in_arbitration(wallet.coldkey.ss58_address)
-        )
-        if arbitration_check == 0:
-            bittensor.__console__.print(
-                "[green]There has been no previous key swap initiated for your coldkey.[/green]"
-            )
-        if arbitration_check == 1:
-            arbitration_remaining = subtensor.get_remaining_arbitration_period(
-                wallet.coldkey.ss58_address
-            )
-            hours, minutes, seconds = convert_blocks_to_time(arbitration_remaining)
-            bittensor.__console__.print(
-                "[yellow]There has been 1 swap request made for this coldkey already."
-                " By adding another swap request, the key will enter arbitration."
-                f" Your key swap is scheduled for {hours} hours, {minutes} minutes, {seconds} seconds"
-            )
-        if arbitration_check > 1:
-            bittensor.__console__.print(
-                f"[red]This coldkey is currently in arbitration with a total swaps of {arbitration_check}.[/red]"
-            )
+        fetch_arbitration_stats(subtensor, wallet)
 
     @classmethod
     def check_config(cls, config: "bittensor.config"):

bittensor/subtensor.py

@@ -44,7 +44,6 @@
 from bittensor.utils import torch, weight_utils, format_error_message
 from bittensor.utils.registration import (
     POWSolution,
-    create_pow,
     torch,
 )
 
@@ -4613,38 +4612,6 @@ def make_substrate_call_with_retry(encoded_coldkey_: List[int]):
     # Scheduled Coldkey Swap Information
     ######################################
 
-    def get_scheduled_coldkey_swap(
-        self, coldkey_ss58: str, block: Optional[int] = None
-    ) -> Optional[ScheduledColdkeySwapInfo]:
-        """
-        Retrieves the scheduled coldkey swap information for a given coldkey.
-
-        Args:
-            coldkey_ss58 (str): The SS58 address of the coldkey.
-            block (Optional[int], optional): The block number to query. If None, uses the latest block.
-
-        Returns:
-            Optional[ScheduledColdkeySwapInfo]: The scheduled coldkey swap information, or None if not found.
-        """
-        encoded_coldkey = ss58_to_vec_u8(coldkey_ss58)
-
-        hex_bytes_result = self.query_runtime_api(
-            runtime_api="ColdkeySwapRuntimeApi",
-            method="get_scheduled_coldkey_swap",
-            params=[encoded_coldkey],
-            block=block,
-        )
-
-        if hex_bytes_result is None:
-            return None
-
-        if hex_bytes_result.startswith("0x"):
-            bytes_result = bytes.fromhex(hex_bytes_result[2:])
-        else:
-            bytes_result = bytes.fromhex(hex_bytes_result)
-
-        return ScheduledColdkeySwapInfo.from_vec_u8(bytes_result)
-
     def get_coldkey_swap_destinations(
         self, coldkey_ss58: str, block: Optional[int] = None
     ) -> Optional[List[str]]:

tests/e2e_tests/subcommands/wallet/test_check_and_schedule_cold_key_swap.py

@@ -52,6 +52,7 @@ def test_check_and_schedule_coldkey_swap(local_chain, capsys):
     # Assert one swap has already been scheduled
     output = capsys.readouterr().out
     assert "There has been 1 swap request made for this coldkey already." in output
+    assert "Your key swap is scheduled for " in output
 
     # Generate new wallet to transfer to
     dave_keypair, dave_exec_command, dave_wallet_path = setup_wallet("//Dave")

tests/unit_tests/utils/test_formatting.py

@@ -0,0 +1,10 @@
+from bittensor.utils.formatting import convert_blocks_to_time
+
+
+def test_convert_blocks_to_time():
+    assert convert_blocks_to_time(6301) == (21, 0, 12)
+    assert convert_blocks_to_time(300) == (1, 0, 0)
+    assert convert_blocks_to_time(10) == (0, 2, 0)
+    assert convert_blocks_to_time(1) == (0, 0, 12)
+    assert convert_blocks_to_time(186, block_time=3) == (0, 9, 18)
+    assert convert_blocks_to_time(0) == (0, 0, 0)