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