Merge pull request #2446 from opentensor/feat/roman/bittensor-console

BittensorConsole class

Author: roman-opentensor

bittensor/core/async_subtensor.py

@@ -132,10 +132,10 @@ def __init__(self, network: str = DEFAULT_NETWORK):
                     self.network = "custom"
             else:
                 logging.info(
-                    f"Network not specified or not valid. Using default chain endpoint: <blue>{NETWORK_MAP[DEFAULTS.subtensor.network]}</blue>."
+                    f"Network not specified or not valid. Using default chain endpoint: [blue]{NETWORK_MAP[DEFAULTS.subtensor.network]}[/blue]."
                 )
                 logging.info(
-                    "You can set this for commands with the <blue>--network</blue> flag, or by setting this in the config."
+                    "You can set this for commands with the [blue]--network[/blue] flag, or by setting this in the config."
                 )
                 self.chain_endpoint = NETWORK_MAP[DEFAULTS.subtensor.network]
                 self.network = DEFAULTS.subtensor.network
@@ -152,19 +152,19 @@ def __str__(self):
 
     async def __aenter__(self):
         logging.info(
-            f"<magenta>Connecting to Substrate:</magenta> <blue>{self}</blue><magenta>...</magenta>"
+            f"[magenta]Connecting to Substrate:[/magenta] [blue]{self}[/blue][magenta]...[/magenta]"
         )
         try:
             async with self.substrate:
                 return self
         except TimeoutException:
             logging.error(
-                f"<red>Error</red>: Timeout occurred connecting to substrate. Verify your chain and network settings: {self}"
+                f"[red]Error[/red]: Timeout occurred connecting to substrate. Verify your chain and network settings: {self}"
             )
             raise ConnectionError
         except (ConnectionRefusedError, ssl.SSLError) as error:
             logging.error(
-                f"<red>Error</red>: Connection refused when connecting to substrate. "
+                f"[red]Error[/red]: Connection refused when connecting to substrate. "
                 f"Verify your chain and network settings: {self}. Error: {error}"
             )
             raise ConnectionError
@@ -533,7 +533,7 @@ async def get_transfer_fee(
                 )
             except Exception as e:
                 logging.error(
-                    f":cross_mark: <red>Failed to get payment info: </red>{e}"
+                    f":cross_mark: [red]Failed to get payment info: [/red]{e}"
                 )
                 payment_info = {"partialFee": int(2e7)}  # assume  0.02 Tao
 
@@ -1408,7 +1408,7 @@ async def register(
             `True` if registration was successful, otherwise `False`.
         """
         logging.info(
-            f"Registering on netuid <blue>0</blue> on network: <blue>{self.network}</blue>"
+            f"Registering on netuid [blue]0[/blue] on network: [blue]{self.network}[/blue]"
         )
 
         # Check current recycle amount
@@ -1431,7 +1431,7 @@ async def register(
         # Check balance is sufficient
         if balance < current_recycle:
             logging.error(
-                f"<red>Insufficient balance {balance} to register neuron. Current recycle is {current_recycle} TAO</red>."
+                f"[red]Insufficient balance {balance} to register neuron. Current recycle is {current_recycle} TAO[/red]."
             )
             return False
 
@@ -1509,7 +1509,7 @@ async def set_weights(
         ) > await self.weights_rate_limit(netuid):
             try:
                 logging.info(
-                    f"Setting weights for subnet #<blue>{netuid}</blue>. Attempt <blue>{retries + 1} of {max_retries}</blue>."
+                    f"Setting weights for subnet #[blue]{netuid}[/blue]. Attempt [blue]{retries + 1} of {max_retries}[/blue]."
                 )
                 success, message = await set_weights_extrinsic(
                     subtensor=self,
@@ -1547,7 +1547,7 @@ async def root_set_weights(
         """
         netuids_ = np.array(netuids, dtype=np.int64)
         weights_ = np.array(weights, dtype=np.float32)
-        logging.info(f"Setting weights in network: <blue>{self.network}</blue>")
+        logging.info(f"Setting weights in network: [blue]{self.network}[/blue]")
         # Run the set weights operation.
         return await set_root_weights_extrinsic(
             subtensor=self,

bittensor/core/extrinsics/async_registration.py

@@ -132,12 +132,12 @@ async def register_extrinsic(
     logging.debug("Checking subnet status")
     if not await subtensor.subnet_exists(netuid):
         logging.error(
-            f":cross_mark: <red>Failed error:</red> subnet <blue>{netuid}</blue> does not exist."
+            f":cross_mark: [red]Failed error:[/red] subnet [blue]{netuid}[/blue] does not exist."
         )
         return False
 
     logging.info(
-        f":satellite: <magenta>Checking Account on subnet</magenta> <blue>{netuid}</blue> <magenta>...</magenta>"
+        f":satellite: [magenta]Checking Account on subnet[/magenta] [blue]{netuid}[/blue] [magenta]...[/magenta]"
     )
     neuron = await subtensor.get_neuron_for_pubkey_and_subnet(
         hotkey_ss58=wallet.hotkey.ss58_address,
@@ -146,7 +146,7 @@ async def register_extrinsic(
 
     if not neuron.is_null:
         logging.debug(
-            f"Wallet <green>{wallet}</green> is already registered on subnet <blue>{neuron.netuid}</blue> with uid<blue>{neuron.uid}</blue>."
+            f"Wallet [green]{wallet}[/green] is already registered on subnet [blue]{neuron.netuid}[/blue] with uid[blue]{neuron.uid}[/blue]."
         )
         return True
 
@@ -163,7 +163,7 @@ async def register_extrinsic(
 
     while True:
         logging.info(
-            f":satellite: <magenta>Registering...</magenta> <blue>({attempts}/{max_allowed_attempts})</blue>"
+            f":satellite: [magenta]Registering...[/magenta] [blue]({attempts}/{max_allowed_attempts})[/blue]"
         )
         # Solve latest POW.
         if cuda:
@@ -201,13 +201,13 @@ async def register_extrinsic(
             )
             if is_registered:
                 logging.error(
-                    f":white_heavy_check_mark: <green>Already registered on netuid:</green> <blue>{netuid}</blue>"
+                    f":white_heavy_check_mark: [green]Already registered on netuid:[/green] [blue]{netuid}[/blue]"
                 )
                 return True
 
         # pow successful, proceed to submit pow to chain for registration
         else:
-            logging.info(":satellite: <magenta>Submitting POW...</magenta>")
+            logging.info(":satellite: [magenta]Submitting POW...[/magenta]")
             # check if pow result is still valid
             while not await pow_result.is_stale_async(subtensor=subtensor):
                 result: tuple[bool, Optional[str]] = await _do_pow_register(
@@ -226,11 +226,10 @@ async def register_extrinsic(
 
                     if "HotKeyAlreadyRegisteredInSubNet" in err_msg:
                         logging.info(
-                            f":white_heavy_check_mark: <green>Already Registered on subnet:</green> <blue>{netuid}</blue>."
+                            f":white_heavy_check_mark: [green]Already Registered on subnet:[/green] [blue]{netuid}[/blue]."
                         )
                         return True
-
-                    logging.error(f":cross_mark: <red>Failed</red>: {err_msg}")
+                    logging.error(f":cross_mark: [red]Failed[/red]: {err_msg}")
                     await asyncio.sleep(0.5)
 
                 # Successful registration, final check for neuron and pubkey
@@ -241,29 +240,28 @@ async def register_extrinsic(
                     )
                     if is_registered:
                         logging.success(
-                            ":white_heavy_check_mark: <green>Registered</green>"
+                            ":white_heavy_check_mark: [green]Registered[/green]"
                         )
                         return True
                     else:
                         # neuron not found, try again
                         logging.error(
-                            ":cross_mark: <red>Unknown error. Neuron not found.</red>"
+                            ":cross_mark: [red]Unknown error. Neuron not found.[/red]"
                         )
                         continue
             else:
                 # Exited loop because pow is no longer valid.
-                logging.error("<red>POW is stale.</red>")
+                logging.error("[red]POW is stale.[/red]")
                 # Try again.
                 # continue
 
         if attempts < max_allowed_attempts:
             # Failed registration, retry pow
             attempts += 1
             logging.error(
-                f":satellite: <magenta>Failed registration, retrying pow ...</magenta> <blue>({attempts}/{max_allowed_attempts})</blue>"
+                f":satellite: [magenta]Failed registration, retrying pow ...[/magenta] [blue]({attempts}/{max_allowed_attempts})[/blue]"
             )
-            continue
         else:
             # Failed to register after max attempts.
-            logging.error("<red>No more attempts.</red>")
+            logging.error("[red]No more attempts.[/red]")
             return False

bittensor/core/extrinsics/async_root.py

@@ -67,18 +67,18 @@ async def root_register_extrinsic(
         return False
 
     logging.debug(
-        f"Checking if hotkey (<blue>{wallet.hotkey_str}</blue>) is registered on root."
+        f"Checking if hotkey ([blue]{wallet.hotkey_str}[/blue]) is registered on root."
     )
     is_registered = await subtensor.is_hotkey_registered(
         netuid=netuid, hotkey_ss58=wallet.hotkey.ss58_address
     )
     if is_registered:
         logging.error(
-            ":white_heavy_check_mark: <green>Already registered on root network.</green>"
+            ":white_heavy_check_mark: [green]Already registered on root network.[/green]"
         )
         return True
 
-    logging.info(":satellite: <magenta>Registering to root network...</magenta>")
+    logging.info(":satellite: [magenta]Registering to root network...[/magenta]")
     call = await subtensor.substrate.compose_call(
         call_module="SubtensorModule",
         call_function="root_register",
@@ -92,7 +92,7 @@ async def root_register_extrinsic(
     )
 
     if not success:
-        logging.error(f":cross_mark: <red>Failed error:</red> {err_msg}")
+        logging.error(f":cross_mark: [red]Failed error:[/red] {err_msg}")
         time.sleep(0.5)
         return False
 
@@ -105,12 +105,12 @@ async def root_register_extrinsic(
         )
         if uid is not None:
             logging.info(
-                f":white_heavy_check_mark: <green>Registered with UID</green> <blue>{uid}</blue>."
+                f":white_heavy_check_mark: [green]Registered with UID[/green] [blue]{uid}[/blue]."
             )
             return True
         else:
             # neuron not found, try again
-            logging.error(":cross_mark: <red>Unknown error. Neuron not found.</red>")
+            logging.error(":cross_mark: [red]Unknown error. Neuron not found.[/red]")
             return False
 
 
@@ -234,11 +234,11 @@ async def set_root_weights_extrinsic(
     logging.info("Normalizing weights")
     formatted_weights = normalize_max_weight(x=weights, limit=max_weight_limit)
     logging.info(
-        f"Raw weights -> Normalized weights: <blue>{weights}</blue> -> <green>{formatted_weights}</green>"
+        f"Raw weights -> Normalized weights: [blue]{weights}[/blue] -> [green]{formatted_weights}[/green]"
     )
 
     try:
-        logging.info(":satellite: <magenta>Setting root weights...<magenta>")
+        logging.info(":satellite: [magenta]Setting root weights...[magenta]")
         weight_uids, weight_vals = convert_weights_and_uids_for_emit(netuids, weights)
 
         success, error_message = await _do_set_root_weights(
@@ -255,14 +255,14 @@ async def set_root_weights_extrinsic(
             return True
 
         if success is True:
-            logging.info(":white_heavy_check_mark: <green>Finalized</green>")
+            logging.info(":white_heavy_check_mark: [green]Finalized[/green]")
             return True
         else:
             fmt_err = error_message
-            logging.error(f":cross_mark: <red>Failed error:</red> {fmt_err}")
+            logging.error(f":cross_mark: [red]Failed error:[/red] {fmt_err}")
             return False
 
     except SubstrateRequestException as e:
         fmt_err = format_error_message(e, subtensor.substrate)
-        logging.error(f":cross_mark: <red>Failed error:</red> {fmt_err}")
+        logging.error(f":cross_mark: [red]Failed error:[/red] {fmt_err}")
         return False

bittensor/core/extrinsics/async_transfer.py

@@ -98,7 +98,7 @@ async def transfer_extrinsic(
     # Validate destination address.
     if not is_valid_bittensor_address_or_public_key(destination):
         logging.error(
-            f":cross_mark: <red>Invalid destination SS58 address</red>: {destination}"
+            f":cross_mark: [red]Invalid destination SS58 address[/red]: {destination}"
         )
         return False
     logging.info(f"Initiating transfer on network: {subtensor.network}")
@@ -109,7 +109,7 @@ async def transfer_extrinsic(
 
     # Check balance.
     logging.info(
-        f":satellite: <magenta>Checking balance and fees on chain </magenta> <blue>{subtensor.network}</blue>"
+        f":satellite: [magenta]Checking balance and fees on chain [/magenta] [blue]{subtensor.network}[/blue]"
     )
     # check existential deposit and fee
     logging.debug("Fetching existential and fee")
@@ -135,13 +135,13 @@ async def transfer_extrinsic(
             return False
 
     if account_balance < (amount + fee + existential_deposit):
-        logging.error(":cross_mark: <red>Not enough balance</red>")
-        logging.error(f"\t\tBalance:\t<blue>{account_balance}</blue>")
-        logging.error(f"\t\tAmount:\t<blue>{amount}</blue>")
-        logging.error(f"\t\tFor fee:\t<blue>{fee}</blue>")
+        logging.error(":cross_mark: [red]Not enough balance[/red]")
+        logging.error(f"\t\tBalance:\t[blue]{account_balance}[/blue]")
+        logging.error(f"\t\tAmount:\t[blue]{amount}[/blue]")
+        logging.error(f"\t\tFor fee:\t[blue]{fee}[/blue]")
         return False
 
-    logging.info(":satellite: <magenta>Transferring...</magenta")
+    logging.info(":satellite: [magenta]Transferring...</magenta")
     success, block_hash, err_msg = await _do_transfer(
         subtensor=subtensor,
         wallet=wallet,
@@ -152,8 +152,8 @@ async def transfer_extrinsic(
     )
 
     if success:
-        logging.success(":white_heavy_check_mark: [green]Finalized</green>")
-        logging.info(f"[green]Block Hash:</green> <blue>{block_hash}</blue>")
+        logging.success(":white_heavy_check_mark: [green]Finalized[/green]")
+        logging.info(f"[green]Block Hash:[/green] [blue]{block_hash}[/blue]")
 
         if subtensor.network == "finney":
             logging.debug("Fetching explorer URLs")
@@ -162,18 +162,18 @@ async def transfer_extrinsic(
             )
             if explorer_urls != {} and explorer_urls:
                 logging.info(
-                    f"[green]Opentensor Explorer Link: {explorer_urls.get('opentensor')}</green>"
+                    f"[green]Opentensor Explorer Link: {explorer_urls.get('opentensor')}[/green]"
                 )
                 logging.info(
-                    f"[green]Taostats Explorer Link: {explorer_urls.get('taostats')}</green>"
+                    f"[green]Taostats Explorer Link: {explorer_urls.get('taostats')}[/green]"
                 )
 
-        logging.info(":satellite: <magenta>Checking Balance...<magenta>")
+        logging.info(":satellite: [magenta]Checking Balance...[magenta]")
         new_balance = await subtensor.get_balance(wallet.coldkeypub.ss58_address)
         logging.info(
-            f"Balance: [blue]{account_balance}</blue> :arrow_right: [green]{new_balance[wallet.coldkeypub.ss58_address]}</green>"
+            f"Balance: [blue]{account_balance}[/blue] :arrow_right: [green]{new_balance[wallet.coldkeypub.ss58_address]}[/green]"
         )
         return True
     else:
-        logging.error(f":cross_mark: <red>Failed</red>: {err_msg}")
+        logging.error(f":cross_mark: [red]Failed[/red]: {err_msg}")
         return False

bittensor/core/extrinsics/async_weights.py

@@ -125,7 +125,7 @@ async def set_weights_extrinsic(
     )
 
     logging.info(
-        ":satellite: <magenta>Setting weights on </magenta><blue>{subtensor.network}</blue> <magenta>...</magenta>"
+        ":satellite: [magenta]Setting weights on [/magenta][blue]{subtensor.network}[/blue] [magenta]...[/magenta]"
     )
     try:
         success, error_message = await _do_set_weights(
@@ -144,14 +144,14 @@ async def set_weights_extrinsic(
 
         if success is True:
             message = "Successfully set weights and Finalized."
-            logging.success(f":white_heavy_check_mark: <green>{message}</green>")
+            logging.success(f":white_heavy_check_mark: [green]{message}[/green]")
             return True, message
         else:
-            logging.error(f"<red>Failed</red> set weights. Error: {error_message}")
+            logging.error(f"[red]Failed[/red] set weights. Error: {error_message}")
             return False, error_message
 
     except Exception as error:
-        logging.error(f":cross_mark: <red>Failed</red> set weights. Error: {error}")
+        logging.error(f":cross_mark: [red]Failed[/red] set weights. Error: {error}")
         return False, str(error)