Bringing back lost methods for setting weights (#2412)

* add methods to fix set_weights extrinsic

* replace `[color]...[/color]`

Author: basfroman

bittensor/core/async_subtensor.py

@@ -117,7 +117,7 @@ def __init__(self, network: str = DEFAULT_NETWORK):
             self.network = network
             if network == "local":
                 logging.warning(
-                    "[yellow]Warning[/yellow]: Verify your local subtensor is running on port 9944."
+                    "Warning: Verify your local subtensor is running on port 9944."
                 )
         else:
             is_valid, _ = validate_chain_endpoint(network)
@@ -1271,19 +1271,19 @@ async def get_delegate_identities(
 
     async def is_hotkey_registered(self, netuid: int, hotkey_ss58: str) -> bool:
         """Checks to see if the hotkey is registered on a given netuid"""
-        _result = await self.substrate.query(
+        result = await self.substrate.query(
             module="SubtensorModule",
             storage_function="Uids",
             params=[netuid, hotkey_ss58],
         )
-        if _result is not None:
+        if result is not None:
             return True
         else:
             return False
 
     async def get_uid_for_hotkey_on_subnet(
         self, hotkey_ss58: str, netuid: int, block_hash: Optional[str] = None
-    ):
+    ) -> Optional[int]:
         """
         Retrieves the unique identifier (UID) for a neuron's hotkey on a specific subnet.
 
@@ -1297,12 +1297,42 @@ async def get_uid_for_hotkey_on_subnet(
 
         The UID is a critical identifier within the network, linking the neuron's hotkey to its operational and governance activities on a particular subnet.
         """
-        return self.substrate.query(
+        result = await self.substrate.query(
             module="SubtensorModule",
             storage_function="Uids",
             params=[netuid, hotkey_ss58],
             block_hash=block_hash,
         )
+        return result
+
+    async def weights_rate_limit(self, netuid: int) -> Optional[int]:
+        """
+        Returns network WeightsSetRateLimit hyperparameter.
+
+        Args:
+            netuid (int): The unique identifier of the subnetwork.
+
+        Returns:
+            Optional[int]: The value of the WeightsSetRateLimit hyperparameter, or ``None`` if the subnetwork does not exist or the parameter is not found.
+        """
+        call = await self.get_hyperparameter(
+            param_name="WeightsSetRateLimit", netuid=netuid
+        )
+        return None if call is None else int(call)
+
+    async def blocks_since_last_update(self, netuid: int, uid: int) -> Optional[int]:
+        """
+        Returns the number of blocks since the last update for a specific UID in the subnetwork.
+
+        Args:
+            netuid (int): The unique identifier of the subnetwork.
+            uid (int): The unique identifier of the neuron.
+
+        Returns:
+            Optional[int]: The number of blocks since the last update, or ``None`` if the subnetwork or UID does not exist.
+        """
+        call = await self.get_hyperparameter(param_name="LastUpdate", netuid=netuid)
+        return None if call is None else await self.get_current_block() - int(call[uid])
 
     # extrinsics
 
@@ -1445,12 +1475,15 @@ async def set_weights(
 
         This function is crucial in shaping the network's collective intelligence, where each neuron's learning and contribution are influenced by the weights it sets towards others【81†source】.
         """
-        uid = self.get_uid_for_hotkey_on_subnet(wallet.hotkey.ss58_address, netuid)
+        uid = await self.get_uid_for_hotkey_on_subnet(
+            wallet.hotkey.ss58_address, netuid
+        )
         retries = 0
         success = False
         message = "No attempt made. Perhaps it is too soon to set weights!"
         while (
-            self.blocks_since_last_update(netuid, uid) > self.weights_rate_limit(netuid)  # type: ignore
+            await self.blocks_since_last_update(netuid, uid)
+            > await self.weights_rate_limit(netuid)
             and retries < max_retries
         ):
             try:

bittensor/core/extrinsics/async_transfer.py

@@ -111,7 +111,7 @@ async def do_transfer() -> tuple[bool, str, str]:
     # Validate destination address.
     if not is_valid_bittensor_address_or_public_key(destination):
         logging.error(
-            f":cross_mark: <red>Invalid destination SS58 address</red>:[bold white]\n  {destination}[/bold white]"
+            f":cross_mark: <red>Invalid destination SS58 address</red>: {destination}"
         )
         return False
     logging.info(f"Initiating transfer on network: {subtensor.network}")

bittensor/utils/__init__.py

@@ -369,9 +369,9 @@ def validate_chain_endpoint(endpoint_url: str) -> tuple[bool, str]:
     parsed = urlparse(endpoint_url)
     if parsed.scheme not in ("ws", "wss"):
         return False, (
-            f"Invalid URL or network name provided: [bright_cyan]({endpoint_url})[/bright_cyan].\n"
-            "Allowed network names are [bright_cyan]finney, test, local[/bright_cyan]. "
-            "Valid chain endpoints should use the scheme [bright_cyan]`ws` or `wss`[/bright_cyan].\n"
+            f"Invalid URL or network name provided: ({endpoint_url}).\n"
+            "Allowed network names are finney, test, local. "
+            "Valid chain endpoints should use the scheme `ws` or `wss`.\n"
         )
     if not parsed.netloc:
         return False, "Invalid URL passed as the endpoint"

bittensor/utils/balance.py

@@ -72,13 +72,13 @@ def __str__(self):
 
     def __rich__(self):
         int_tao, fract_tao = format(float(self.tao), "f").split(".")
-        return f"[green]{self.unit}[/green][green]{int_tao}[/green][green].[/green][dim green]{fract_tao}[/dim green]"
+        return f"<green>{self.unit}{int_tao}.{fract_tao}</green>"
 
     def __str_rao__(self):
         return f"{self.rao_unit}{int(self.rao)}"
 
     def __rich_rao__(self):
-        return f"[green]{self.rao_unit}{int(self.rao)}[/green]"
+        return f"<green>{self.rao_unit}{int(self.rao)}</green>"
 
     def __repr__(self):
         return self.__str__()