diff --git a/bittensor_cli/cli.py b/bittensor_cli/cli.py index 47c005c3..a55d7ebe 100755 --- a/bittensor_cli/cli.py +++ b/bittensor_cli/cli.py @@ -2913,7 +2913,12 @@ def wallet_inspect( ), wallet_name: str = Options.wallet_name, wallet_path: str = Options.wallet_path, - wallet_hotkey: str = Options.wallet_hotkey, + ss58_address: Optional[str] = typer.Option( + None, + "--ss58-address", + "--ss58", + help="SS58 address of the coldkey to inspect. Allows inspecting any coldkey without a local wallet file.", + ), network: Optional[list[str]] = Options.network, netuids: str = Options.netuids, quiet: bool = Options.quiet, @@ -2921,9 +2926,11 @@ def wallet_inspect( json_output: bool = Options.json_output, ): """ - Displays the details of the user's wallet pairs (coldkey, hotkey) on the Bittensor network. + Displays the details of the user's wallet (coldkey) on the Bittensor network. + + The output is presented as two separate tables: - The output is presented as a table with the below columns: + [bold]Coldkey Overview[/bold]: - [blue bold]Coldkey[/blue bold]: The coldkey associated with the user's wallet. @@ -2931,14 +2938,22 @@ def wallet_inspect( - [blue bold]Delegate[/blue bold]: The name of the delegate to which the coldkey has staked TAO. - - [blue bold]Stake[/blue bold]: The amount of stake held by both the coldkey and hotkey. + - [blue bold]Stake[/blue bold]: The amount of stake delegated. - - [blue bold]Emission[/blue bold]: The emission or rewards earned from staking. + - [blue bold]Emission[/blue bold]: The daily emission earned from delegation. - - [blue bold]Netuid[/blue bold]: The network unique identifier of the subnet where the hotkey is active (i.e., validating). + [bold]Hotkey Details[/bold]: + + - [blue bold]Coldkey[/blue bold]: The parent coldkey of the hotkey. + + - [blue bold]Netuid[/blue bold]: The network unique identifier of the subnet where the hotkey is active. - [blue bold]Hotkey[/blue bold]: The hotkey associated with the neuron on the network. + - [blue bold]Stake[/blue bold]: The amount of stake held by the hotkey. + + - [blue bold]Emission[/blue bold]: The emission or rewards earned from staking. + USAGE This command can be used to inspect a single wallet or all the wallets located at a specified path. It is useful for a comprehensive overview of a user's participation and performance in the Bittensor network. @@ -2949,10 +2964,10 @@ def wallet_inspect( [green]$[/green] btcli wallet inspect --all -n 1 -n 2 -n 3 + [green]$[/green] btcli wallet inspect --ss58-address 5FHneW46... + [bold]Note[/bold]: The `inspect` command is for displaying information only and does not perform any transactions or state changes on the blockchain. It is intended to be used with Bittensor CLI and not as a standalone function in user code. """ - print_error("This command is disabled on the 'rao' network.") - raise typer.Exit() self.verbosity_handler(quiet, verbose, json_output, False) if netuids: @@ -2962,14 +2977,26 @@ def wallet_inspect( "Netuids must be a comma-separated list of ints, e.g., `--netuids 1,2,3,4`.", ) + self.initialize_chain(network) + + if ss58_address: + return self._run_command( + wallets.inspect( + None, + self.subtensor, + netuids_filter=netuids, + all_wallets=False, + ss58_address=ss58_address, + ) + ) + # if all-wallets is entered, ask for path ask_for = [WO.NAME, WO.PATH] if not all_wallets else [WO.PATH] validate = WV.WALLET if not all_wallets else WV.NONE wallet = self.wallet_ask( - wallet_name, wallet_path, wallet_hotkey, ask_for=ask_for, validate=validate + wallet_name, wallet_path, None, ask_for=ask_for, validate=validate ) - self.initialize_chain(network) return self._run_command( wallets.inspect( wallet, diff --git a/bittensor_cli/src/commands/wallets.py b/bittensor_cli/src/commands/wallets.py index abd76d35..7e2d766e 100644 --- a/bittensor_cli/src/commands/wallets.py +++ b/bittensor_cli/src/commands/wallets.py @@ -1,6 +1,5 @@ import asyncio import hashlib -import itertools import json import os from collections import defaultdict @@ -1559,69 +1558,178 @@ async def transfer( return result +def _build_coldkey_table(network: str) -> Table: + """Build the coldkey overview table for wallet inspect output.""" + return Table( + Column("[bold white]Coldkey", style="dark_orange"), + Column("[bold white]Balance", style="dark_sea_green"), + Column("[bold white]Delegate", style="bright_cyan", overflow="fold"), + Column("[bold white]Stake", style="light_goldenrod2"), + Column("[bold white]Emission", style="rgb(42,161,152)"), + title=( + f"[underline dark_orange]Coldkey Overview" + f"[/underline dark_orange]\n" + f"[dark_orange]Network: {network}\n" + ), + show_edge=False, + expand=True, + box=box.MINIMAL, + border_style="bright_black", + ) + + +def _build_hotkey_table(network: str) -> Table: + """Build the hotkey details table for wallet inspect output.""" + return Table( + Column("[bold white]Coldkey", style="dark_orange"), + Column("[bold white]Netuid", style="dark_orange"), + Column("[bold white]Hotkey", style="bright_magenta", overflow="fold"), + Column("[bold white]Stake", style="light_goldenrod2"), + Column("[bold white]Emission", style="rgb(42,161,152)"), + title=( + f"[underline dark_orange]Hotkey Details" + f"[/underline dark_orange]\n" + f"[dark_orange]Network: {network}\n" + ), + show_edge=False, + expand=True, + box=box.MINIMAL, + border_style="bright_black", + ) + + +def _make_delegate_rows( + delegates: list[tuple[DelegateInfo, Balance]], + registered_delegate_info: dict, +) -> Generator[list[str], None, None]: + """Yield coldkey table rows for each delegate with a positive stake.""" + for delegate_info, staked in delegates: + if not staked.tao > 0: + continue + delegate_name = _resolve_delegate_name( + delegate_info.hotkey_ss58, registered_delegate_info + ) + daily_return = _calculate_daily_return(delegate_info, staked) + yield ["", "", str(delegate_name), str(staked), str(daily_return)] + + +def _resolve_delegate_name(hotkey_ss58: str, registered_delegate_info: dict) -> str: + """Look up the display name for a delegate, falling back to the SS58 address.""" + if hotkey_ss58 in registered_delegate_info: + return registered_delegate_info[hotkey_ss58].display + return hotkey_ss58 + + +def _calculate_daily_return(delegate_info: DelegateInfo, staked: Balance) -> float: + """Calculate the estimated daily return for a delegation.""" + if delegate_info.total_stake.tao != 0: + return delegate_info.total_daily_return.tao * ( + staked.tao / delegate_info.total_stake.tao + ) + return 0 + + +def _make_neuron_rows( + coldkey_ss58: str, + coldkey_name: str, + all_netuids: list[int], + neuron_state_dict: dict, + hotkeys: list = None, +) -> Generator[list[str], None, None]: + """Yield hotkey table rows for each neuron registered under the given coldkey.""" + hotkeys = hotkeys or [] + for netuid in all_netuids: + for neuron in neuron_state_dict[netuid]: + if neuron.coldkey == coldkey_ss58: + hotkey_label = _format_hotkey_label(neuron.hotkey, hotkeys) + stake = Balance.from_rao(neuron.stake.rao).set_unit(netuid) + emission = Balance.from_tao(neuron.emission).set_unit(netuid) + yield [ + coldkey_name, + str(netuid), + hotkey_label, + str(stake), + str(emission), + ] + + +def _format_hotkey_label(hotkey_ss58: str, hotkeys: list) -> str: + """Format a hotkey address with its wallet name prefix if available.""" + for wallet in hotkeys: + if get_hotkey_pub_ss58(wallet) == hotkey_ss58: + return f"{wallet.hotkey_str}-{hotkey_ss58}" + return hotkey_ss58 + + +def _populate_coldkey_table( + coldkey_table: Table, + coldkey_rows: list[list[str]], +) -> None: + """Add rows to the coldkey overview table with section separators.""" + for i, row in enumerate(coldkey_rows): + is_last_row = i + 1 == len(coldkey_rows) + coldkey_table.add_row(*row) + if is_last_row or (coldkey_rows[i + 1][0] != ""): + coldkey_table.add_row(end_section=True) + + +def _populate_hotkey_table( + hotkey_table: Table, + hotkey_rows: list[list[str]], +) -> None: + """Add rows to the hotkey details table with section separators.""" + for i, row in enumerate(hotkey_rows): + is_last_row = i + 1 == len(hotkey_rows) + hotkey_table.add_row(*row) + if is_last_row or (hotkey_rows[i + 1][0] != ""): + hotkey_table.add_row(end_section=True) + + async def inspect( - wallet: Wallet, + wallet: Optional[Wallet], subtensor: SubtensorInterface, netuids_filter: list[int], all_wallets: bool = False, + ss58_address: Optional[str] = None, ): - # TODO add json_output when this is re-enabled and updated for dTAO - def delegate_row_maker( - delegates_: list[tuple[DelegateInfo, Balance]], - ) -> Generator[list[str], None, None]: - for d_, staked in delegates_: - if not staked.tao > 0: - continue - if d_.hotkey_ss58 in registered_delegate_info: - delegate_name = registered_delegate_info[d_.hotkey_ss58].display - else: - delegate_name = d_.hotkey_ss58 - yield ( - [""] * 2 - + [ - str(delegate_name), - str(staked), - str( - d_.total_daily_return.tao * (staked.tao / d_.total_stake.tao) - if d_.total_stake.tao != 0 - else 0 - ), - ] - + [""] * 4 - ) - - def neuron_row_maker( - wallet_, all_netuids_, nsd - ) -> Generator[list[str], None, None]: - hotkeys = get_hotkey_wallets_for_wallet(wallet_) - for netuid in all_netuids_: - for n in nsd[netuid]: - if n.coldkey == wallet_.coldkeypub.ss58_address: - hotkey_name: str = "" - if hotkey_names := [ - w.hotkey_str - for w in hotkeys - if get_hotkey_pub_ss58(w) == n.hotkey - ]: - hotkey_name = f"{hotkey_names[0]}-" - yield [""] * 5 + [ - str(netuid), - f"{hotkey_name}{n.hotkey}", - str(n.stake), - str(Balance.from_tao(n.emission)), - ] - - if all_wallets: + if ss58_address: + # Direct SS58 address lookup — no wallet file needed + coldkey_addresses = [ss58_address] + coldkey_names = {ss58_address: ss58_address[:8] + "..."} + hotkeys_by_coldkey: dict[str, list] = {ss58_address: []} + elif all_wallets: print_verbose("Fetching data for all wallets") wallets = get_coldkey_wallets_for_path(wallet.path) - all_hotkeys = get_all_wallets_for_path( - wallet.path - ) # TODO verify this is correct - + wallets_with_ckp = [w for w in wallets if w.coldkeypub_file.exists_on_device()] + coldkey_addresses = [w.coldkeypub.ss58_address for w in wallets_with_ckp] + coldkey_names = { + w.coldkeypub.ss58_address: w.name for w in wallets_with_ckp + } + hotkeys_by_coldkey = { + w.coldkeypub.ss58_address: get_hotkey_wallets_for_wallet(w) + for w in wallets_with_ckp + } + all_hotkeys = get_all_wallets_for_path(wallet.path) else: print_verbose(f"Fetching data for wallet: {wallet.name}") - wallets = [wallet] - all_hotkeys = get_hotkey_wallets_for_wallet(wallet) + wallets_with_ckp = [wallet] if wallet.coldkeypub_file.exists_on_device() else [] + coldkey_addresses = [w.coldkeypub.ss58_address for w in wallets_with_ckp] + coldkey_names = { + w.coldkeypub.ss58_address: w.name for w in wallets_with_ckp + } + hotkeys_by_coldkey = { + w.coldkeypub.ss58_address: get_hotkey_wallets_for_wallet(w) + for w in wallets_with_ckp + } + all_hotkeys = get_hotkey_wallets_for_wallet(wallet) if wallet else [] + + if not coldkey_addresses: + console.print("[yellow]No wallet data found.[/yellow]") + return + + # For ss58_address mode we skip the hotkey-based netuid filter + if ss58_address: + all_hotkeys = [] with console.status("Synchronising with chain...", spinner="aesthetic") as status: block_hash = await subtensor.substrate.get_chain_head() @@ -1634,7 +1742,7 @@ def neuron_row_maker( all_hotkeys, block_hash=block_hash, ) - # bittensor.logging.debug(f"Netuids to check: {all_netuids}") + with console.status("Pulling delegates info...", spinner="aesthetic"): registered_delegate_info = await subtensor.get_delegate_identities() if not registered_delegate_info: @@ -1642,33 +1750,13 @@ def neuron_row_maker( ":warning:[yellow]Could not get delegate info from chain.[/yellow]" ) - table = Table( - Column("[bold white]Coldkey", style="dark_orange"), - Column("[bold white]Balance", style="dark_sea_green"), - Column("[bold white]Delegate", style="bright_cyan", overflow="fold"), - Column("[bold white]Stake", style="light_goldenrod2"), - Column("[bold white]Emission", style="rgb(42,161,152)"), - Column("[bold white]Netuid", style="dark_orange"), - Column("[bold white]Hotkey", style="bright_magenta", overflow="fold"), - Column("[bold white]Stake", style="light_goldenrod2"), - Column("[bold white]Emission", style="rgb(42,161,152)"), - title=f"[underline dark_orange]Wallets[/underline dark_orange]\n[dark_orange]Network: {subtensor.network}\n", - show_edge=False, - expand=True, - box=box.MINIMAL, - border_style="bright_black", - ) - rows = [] - wallets_with_ckp_file = [ - wallet for wallet in wallets if wallet.coldkeypub_file.exists_on_device() - ] + coldkey_table = _build_coldkey_table(subtensor.network) + hotkey_table = _build_hotkey_table(subtensor.network) + all_delegates: list[list[tuple[DelegateInfo, Balance]]] with console.status("Pulling balance data...", spinner="aesthetic"): balances, all_neurons, all_delegates = await asyncio.gather( - subtensor.get_balances( - *[w.coldkeypub.ss58_address for w in wallets_with_ckp_file], - block_hash=block_hash, - ), + subtensor.get_balances(*coldkey_addresses, block_hash=block_hash), asyncio.gather( *[ subtensor.neurons_lite(netuid=netuid, block_hash=block_hash) @@ -1677,8 +1765,8 @@ def neuron_row_maker( ), asyncio.gather( *[ - subtensor.get_delegated(w.coldkeypub.ss58_address) - for w in wallets_with_ckp_file + subtensor.get_delegated(addr) + for addr in coldkey_addresses ] ), ) @@ -1686,23 +1774,33 @@ def neuron_row_maker( for netuid, neuron in zip(all_netuids, all_neurons): neuron_state_dict[netuid] = neuron if neuron else [] - for wall, d in zip(wallets_with_ckp_file, all_delegates): - rows.append([wall.name, str(balances[wall.coldkeypub.ss58_address])] + [""] * 7) - for row in itertools.chain( - delegate_row_maker(d), - neuron_row_maker(wall, all_netuids, neuron_state_dict), + coldkey_rows = [] + hotkey_rows = [] + for addr, delegates in zip(coldkey_addresses, all_delegates): + name = coldkey_names[addr] + coldkey_rows.append( + [name, str(balances.get(addr, Balance(0))), "", "", ""] + ) + for row in _make_delegate_rows(delegates, registered_delegate_info): + coldkey_rows.append(row) + + hotkeys = hotkeys_by_coldkey.get(addr, []) + for row in _make_neuron_rows( + addr, name, all_netuids, neuron_state_dict, hotkeys ): - rows.append(row) + hotkey_rows.append(row) - for i, row in enumerate(rows): - is_last_row = i + 1 == len(rows) - table.add_row(*row) + if coldkey_rows: + _populate_coldkey_table(coldkey_table, coldkey_rows) + console.print(coldkey_table) - # If last row or new coldkey starting next - if is_last_row or (rows[i + 1][0] != ""): - table.add_row(end_section=True) + if hotkey_rows: + console.print() + _populate_hotkey_table(hotkey_table, hotkey_rows) + console.print(hotkey_table) - return console.print(table) + if not coldkey_rows and not hotkey_rows: + console.print("[yellow]No wallet data found.[/yellow]") async def faucet( diff --git a/tests/unit_tests/test_inspect.py b/tests/unit_tests/test_inspect.py new file mode 100644 index 00000000..0cb41fa8 --- /dev/null +++ b/tests/unit_tests/test_inspect.py @@ -0,0 +1,173 @@ +from unittest.mock import MagicMock, patch + +from rich.table import Table + +from bittensor_cli.src.bittensor.balances import Balance +from bittensor_cli.src.bittensor.chain_data import DelegateInfo +from bittensor_cli.src.commands.wallets import ( + _build_coldkey_table, + _build_hotkey_table, + _calculate_daily_return, + _format_hotkey_label, + _make_delegate_rows, + _populate_coldkey_table, + _populate_hotkey_table, + _resolve_delegate_name, +) + + +def _make_mock_delegate( + hotkey_ss58: str, + total_stake_tao: float, + total_daily_return_tao: float, +) -> DelegateInfo: + """Create a mock DelegateInfo with the specified stake and return values.""" + delegate = MagicMock(spec=DelegateInfo) + delegate.hotkey_ss58 = hotkey_ss58 + delegate.total_stake = Balance.from_tao(total_stake_tao) + delegate.total_daily_return = Balance.from_tao(total_daily_return_tao) + return delegate + + +def _make_mock_neuron(coldkey: str, hotkey: str, stake_tao: float, emission: float): + """Create a mock NeuronInfoLite with the specified fields.""" + neuron = MagicMock() + neuron.coldkey = coldkey + neuron.hotkey = hotkey + neuron.stake = Balance.from_tao(stake_tao) + neuron.emission = emission + return neuron + + +class TestBuildColdkeyTable: + def test_returns_table_with_correct_columns(self): + table = _build_coldkey_table("finney") + assert isinstance(table, Table) + column_names = [col.header for col in table.columns] + assert "[bold white]Coldkey" in column_names + assert "[bold white]Balance" in column_names + assert "[bold white]Delegate" in column_names + assert "[bold white]Stake" in column_names + assert "[bold white]Emission" in column_names + assert len(table.columns) == 5 + + def test_title_contains_network(self): + table = _build_coldkey_table("test") + assert "test" in table.title + + +class TestBuildHotkeyTable: + def test_returns_table_with_correct_columns(self): + table = _build_hotkey_table("finney") + assert isinstance(table, Table) + column_names = [col.header for col in table.columns] + assert "[bold white]Coldkey" in column_names + assert "[bold white]Netuid" in column_names + assert "[bold white]Hotkey" in column_names + assert "[bold white]Stake" in column_names + assert "[bold white]Emission" in column_names + assert len(table.columns) == 5 + + def test_title_contains_network(self): + table = _build_hotkey_table("test") + assert "test" in table.title + + +class TestResolveDelegateName: + def test_known_delegate_returns_display_name(self): + info = {"5abc": MagicMock(display="MyDelegate")} + assert _resolve_delegate_name("5abc", info) == "MyDelegate" + + def test_unknown_delegate_returns_ss58(self): + assert _resolve_delegate_name("5xyz", {}) == "5xyz" + + +class TestCalculateDailyReturn: + def test_positive_stake_returns_proportional_return(self): + delegate = _make_mock_delegate("5abc", 100.0, 10.0) + staked = Balance.from_tao(50.0) + result = _calculate_daily_return(delegate, staked) + assert result == 5.0 + + def test_zero_total_stake_returns_zero(self): + delegate = _make_mock_delegate("5abc", 0.0, 10.0) + staked = Balance.from_tao(50.0) + result = _calculate_daily_return(delegate, staked) + assert result == 0 + + +class TestMakeDelegateRows: + def test_yields_rows_for_positive_stakes(self): + delegate = _make_mock_delegate("5abc", 100.0, 10.0) + delegates = [(delegate, Balance.from_tao(50.0))] + info = {"5abc": MagicMock(display="MyDelegate")} + rows = list(_make_delegate_rows(delegates, info)) + assert len(rows) == 1 + assert rows[0][2] == "MyDelegate" + + def test_skips_zero_stake_delegates(self): + delegate = _make_mock_delegate("5abc", 100.0, 10.0) + delegates = [(delegate, Balance.from_tao(0.0))] + rows = list(_make_delegate_rows(delegates, {})) + assert len(rows) == 0 + + def test_multiple_delegates(self): + d1 = _make_mock_delegate("5aaa", 100.0, 10.0) + d2 = _make_mock_delegate("5bbb", 200.0, 20.0) + delegates = [ + (d1, Balance.from_tao(10.0)), + (d2, Balance.from_tao(20.0)), + ] + rows = list(_make_delegate_rows(delegates, {})) + assert len(rows) == 2 + + +class TestFormatHotkeyLabel: + def test_known_hotkey_includes_wallet_name(self): + mock_wallet = MagicMock() + mock_wallet.hotkey_str = "myhk" + with patch( + "bittensor_cli.src.commands.wallets.get_hotkey_pub_ss58", + return_value="5hotkey", + ): + result = _format_hotkey_label("5hotkey", [mock_wallet]) + assert result == "myhk-5hotkey" + + def test_unknown_hotkey_returns_ss58(self): + with patch( + "bittensor_cli.src.commands.wallets.get_hotkey_pub_ss58", + return_value="5other", + ): + result = _format_hotkey_label("5hotkey", [MagicMock()]) + assert result == "5hotkey" + + +class TestPopulateColdkeyTable: + def test_adds_rows_to_table(self): + table = _build_coldkey_table("finney") + rows = [ + ["alice", "100.0", "", "", ""], + ["", "", "Delegate1", "50.0", "1.0"], + ] + _populate_coldkey_table(table, rows) + assert table.row_count == 3 + + def test_empty_rows_noop(self): + table = _build_coldkey_table("finney") + _populate_coldkey_table(table, []) + assert table.row_count == 0 + + +class TestPopulateHotkeyTable: + def test_adds_rows_to_table(self): + table = _build_hotkey_table("finney") + rows = [ + ["alice", "1", "5hotkey", "10.0", "0.5"], + ] + _populate_hotkey_table(table, rows) + assert table.row_count == 2 + + def test_empty_rows_noop(self): + table = _build_hotkey_table("finney") + _populate_hotkey_table(table, []) + assert table.row_count == 0