communex.cli.misc

  1import typer
  2from typer import Context
  3
  4from communex._common import BalanceUnit, format_balance
  5from communex.balance import from_nano
  6from communex.cli._common import make_custom_context, print_module_info
  7from communex.client import CommuneClient
  8from communex.compat.key import (local_key_addresses,
  9                                 resolve_key_ss58_encrypted,
 10                                 try_classic_load_key)
 11from communex.misc import get_map_modules
 12from communex.types import ModuleInfoWithOptionalBalance
 13
 14misc_app = typer.Typer(no_args_is_help=True)
 15
 16
 17def circulating_tokens(c_client: CommuneClient) -> int:
 18    """
 19    Gets total circulating supply
 20    """
 21
 22    with c_client.get_conn(init=True) as substrate:
 23        block_hash = substrate.get_block_hash()
 24
 25    total_balance = c_client.get_total_free_issuance(block_hash=block_hash)
 26    total_stake = c_client.get_total_stake(block_hash=block_hash)
 27    return total_stake + total_balance
 28
 29
 30@misc_app.command()
 31def circulating_supply(ctx: Context, unit: BalanceUnit = BalanceUnit.joule):
 32    """
 33    Gets the value of all keys on the network, stake + balances
 34    """
 35    context = make_custom_context(ctx)
 36    client = context.com_client()
 37
 38    with context.progress_status("Getting circulating supply, across all subnets..."):
 39        supply = circulating_tokens(client)
 40
 41    context.output(format_balance(supply, unit))
 42
 43
 44@misc_app.command()
 45def apr(ctx: Context, fee: int = 0):
 46    """
 47    Gets the current staking APR on validators.
 48    The miner reinvest rate & fee are specified in percentages.
 49    """
 50    context = make_custom_context(ctx)
 51    client = context.com_client()
 52
 53    # adjusting the fee to the correct format
 54    # the default validator fee on the commune network is 20%
 55    fee_to_float = fee / 100
 56
 57    # network parameters
 58    block_time = 8  # seconds
 59    seconds_in_a_day = 86400
 60    blocks_in_a_day = seconds_in_a_day / block_time
 61
 62    with context.progress_status("Getting staking APR..."):
 63        unit_emission = client.get_unit_emission()
 64        total_staked_tokens = client.query("TotalStake")
 65    # 50% of the total emission goes to stakers
 66    daily_token_rewards = blocks_in_a_day * from_nano(unit_emission) / 2
 67    _apr = (daily_token_rewards * (1 - fee_to_float)
 68            * 365) / total_staked_tokens * 100
 69
 70    context.output(f"Fee {fee} | APR {_apr:.2f}%")
 71
 72
 73@misc_app.command(name="stats")
 74def stats(ctx: Context, balances: bool = False, netuid: int = 0):
 75    context = make_custom_context(ctx)
 76    client = context.com_client()
 77
 78    with context.progress_status(f"Getting Modules on a subnet with netuid {netuid}..."):
 79        modules = get_map_modules(
 80            client, netuid=netuid, include_balances=balances)
 81    modules_to_list = [value for _, value in modules.items()]
 82    local_keys = local_key_addresses()
 83    local_modules = [
 84        *filter(lambda module: module["key"] in local_keys.values(), modules_to_list)]
 85    local_miners: list[ModuleInfoWithOptionalBalance] = []
 86    local_validators: list[ModuleInfoWithOptionalBalance] = []
 87    local_inactive: list[ModuleInfoWithOptionalBalance] = []
 88    for module in local_modules:
 89        if module["incentive"] == module["dividends"] == 0:
 90            local_inactive.append(module)
 91        elif module["incentive"] > module["dividends"]:
 92            local_miners.append(module)
 93        else:
 94            local_validators.append(module)
 95
 96    print_module_info(client, local_inactive,
 97                      context.console, netuid, "inactive")
 98    print_module_info(client, local_miners, context.console, netuid, "miners")
 99    print_module_info(client, local_validators,
100                      context.console, netuid, "validators")
101
102
103@misc_app.command(name="treasury-address")
104def get_treasury_address(ctx: Context):
105    context = make_custom_context(ctx)
106    client = context.com_client()
107
108    with context.progress_status("Getting DAO treasury address..."):
109        dao_address = client.get_dao_treasury_address()
110    context.output(dao_address)
111
112
113@misc_app.command()
114def delegate_rootnet_control(ctx: Context, key: str, target: str):
115    """
116    Delegates control of the rootnet to a key
117    """
118    context = make_custom_context(ctx)
119    client = context.com_client()
120    resolved_key = try_classic_load_key(key, context)
121    ss58_target = resolve_key_ss58_encrypted(target, context)
122
123    with context.progress_status("Delegating control of the rootnet..."):
124        client.delegate_rootnet_control(
125            resolved_key, ss58_target
126        )
127    context.info("Control delegated.")