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
  9from communex.misc import get_map_modules
 10from communex.types import ModuleInfoWithOptionalBalance
 11
 12misc_app = typer.Typer(no_args_is_help=True)
 13
 14
 15def circulating_tokens(c_client: CommuneClient) -> int:
 16    """
 17    Gets total circulating supply
 18    """
 19
 20    with c_client.get_conn(init=True) as substrate:
 21        block_hash = substrate.get_block_hash()
 22
 23    total_balance = c_client.get_total_free_issuance(block_hash=block_hash)
 24    total_stake = c_client.get_total_stake(block_hash=block_hash)
 25    return total_stake + total_balance
 26
 27
 28@misc_app.command()
 29def circulating_supply(ctx: Context, unit: BalanceUnit = BalanceUnit.joule):
 30    """
 31    Gets the value of all keys on the network, stake + balances
 32    """
 33    context = make_custom_context(ctx)
 34    client = context.com_client()
 35
 36    with context.progress_status(
 37        "Getting circulating supply, across all subnets..."
 38    ):
 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 = (
 68        (daily_token_rewards * (1 - fee_to_float) * 365)
 69        / total_staked_tokens
 70        * 100
 71    )
 72
 73    context.output(f"Fee {fee} | APR {_apr:.2f}%")
 74
 75
 76@misc_app.command(name="stats")
 77def stats(ctx: Context, balances: bool = False, netuid: int = 0):
 78    context = make_custom_context(ctx)
 79    client = context.com_client()
 80
 81    with context.progress_status(
 82        f"Getting Modules on a subnet with netuid {netuid}..."
 83    ):
 84        modules = get_map_modules(
 85            client, netuid=netuid, include_balances=balances
 86        )
 87    modules_to_list = [value for _, value in modules.items()]
 88    local_keys = local_key_addresses(password_provider=context.password_manager)
 89    local_modules = [
 90        *filter(
 91            lambda module: module["key"] in local_keys.values(), modules_to_list
 92        )
 93    ]
 94    local_miners: list[ModuleInfoWithOptionalBalance] = []
 95    local_validators: list[ModuleInfoWithOptionalBalance] = []
 96    local_inactive: list[ModuleInfoWithOptionalBalance] = []
 97    for module in local_modules:
 98        if module["incentive"] == module["dividends"] == 0:
 99            local_inactive.append(module)
100        elif module["incentive"] > module["dividends"]:
101            local_miners.append(module)
102        else:
103            local_validators.append(module)
104
105    print_module_info(
106        client, local_inactive, context.console, netuid, "inactive"
107    )
108    print_module_info(client, local_miners, context.console, netuid, "miners")
109    print_module_info(
110        client, local_validators, context.console, netuid, "validators"
111    )
112
113
114@misc_app.command(name="treasury-address")
115def get_treasury_address(ctx: Context):
116    context = make_custom_context(ctx)
117    client = context.com_client()
118
119    with context.progress_status("Getting DAO treasury address..."):
120        dao_address = client.get_dao_treasury_address()
121    context.output(dao_address)
122
123
124@misc_app.command()
125def delegate_rootnet_control(ctx: Context, key: str, target: str):
126    """
127    Delegates control of the rootnet to a key
128    """
129    context = make_custom_context(ctx)
130    client = context.com_client()
131    resolved_key = context.load_key(key, None)
132    ss58_target = context.resolve_key_ss58(target, None)
133
134    with context.progress_status("Delegating control of the rootnet..."):
135        client.delegate_rootnet_control(resolved_key, ss58_target)
136    context.info("Control delegated.")