communex.cli.network
1import re 2from typing import Optional, cast 3 4import typer 5from rich.progress import track 6from typer import Context 7 8import communex.balance as c_balance 9from communex.cli._common import ( 10 CustomCtx, make_custom_context, 11 print_table_from_plain_dict, tranform_network_params 12) 13from communex.client import CommuneClient 14from communex.compat.key import local_key_addresses, try_classic_load_key 15from communex.misc import (IPFS_REGEX, get_global_params, 16 local_keys_to_stakedbalance) 17from communex.types import NetworkParams 18from communex.util import convert_cid_on_proposal 19 20network_app = typer.Typer(no_args_is_help=True) 21 22 23@network_app.command() 24def last_block(ctx: Context, hash: bool = False): 25 """ 26 Gets the last block 27 """ 28 context = make_custom_context(ctx) 29 client = context.com_client() 30 31 info = "number" if not hash else "hash" 32 33 block = client.get_block() 34 block_info = None 35 if block: 36 block_info = block["header"][info] 37 38 context.output(str(block_info)) 39 40 41@network_app.command() 42def params(ctx: Context): 43 """ 44 Gets global params 45 """ 46 context = make_custom_context(ctx) 47 client = context.com_client() 48 49 with context.progress_status("Getting global network params ..."): 50 global_params = get_global_params(client) 51 printable_params = tranform_network_params(global_params) 52 print_table_from_plain_dict( 53 printable_params, ["Global params", "Value"], context.console 54 ) 55 56 57@network_app.command() 58def list_proposals(ctx: Context, query_cid: bool = typer.Option(True)): 59 """ 60 Gets proposals 61 """ 62 context = make_custom_context(ctx) 63 client = context.com_client() 64 65 with context.progress_status("Getting proposals..."): 66 try: 67 proposals = client.query_map_proposals() 68 if query_cid: 69 proposals = convert_cid_on_proposal(proposals) 70 except IndexError: 71 context.info("No proposals found.") 72 return 73 74 for proposal_id, batch_proposal in proposals.items(): 75 status = batch_proposal["status"] 76 if isinstance(status, dict): 77 batch_proposal["status"] = [*status.keys()][0] 78 print_table_from_plain_dict( 79 batch_proposal, [ 80 f"Proposal id: {proposal_id}", "Params"], context.console 81 ) 82 83 84@network_app.command() 85def propose_globally( 86 ctx: Context, 87 key: str, 88 cid: str, 89 max_name_length: int = typer.Option(None), 90 min_name_length: int = typer.Option(None), 91 max_allowed_subnets: int = typer.Option(None), 92 max_allowed_modules: int = typer.Option(None), 93 max_registrations_per_block: int = typer.Option(None), 94 max_allowed_weights: int = typer.Option(None), 95 max_burn: int = typer.Option(None), 96 min_burn: int = typer.Option(None), 97 floor_delegation_fee: int = typer.Option(None), 98 floor_founder_share: int = typer.Option(None), 99 min_weight_stake: int = typer.Option(None), 100 curator: str = typer.Option(None), 101 proposal_cost: int = typer.Option(None), 102 proposal_expiration: int = typer.Option(None), 103 general_subnet_application_cost: int = typer.Option(None), 104 kappa: int = typer.Option(None), 105 rho: int = typer.Option(None), 106 subnet_immunity_period: int = typer.Option(None), 107): 108 provided_params = locals().copy() 109 provided_params.pop("ctx") 110 provided_params.pop("key") 111 112 provided_params = { 113 key: value for key, value in provided_params.items() if value is not None 114 } 115 """ 116 Adds a global proposal to the network. 117 """ 118 context = make_custom_context(ctx) 119 resolved_key = try_classic_load_key(key, context) 120 client = context.com_client() 121 122 provided_params = cast(NetworkParams, provided_params) 123 global_params = get_global_params(client) 124 global_params_config = global_params["governance_config"] 125 global_params["proposal_cost"] = global_params_config["proposal_cost"] # type: ignore 126 global_params["proposal_expiration"] = global_params_config["proposal_expiration"] # type: ignore 127 global_params.pop("governance_config") # type: ignore 128 global_params.update(provided_params) 129 130 if not re.match(IPFS_REGEX, cid): 131 context.error(f"CID provided is invalid: {cid}") 132 exit(1) 133 with context.progress_status("Adding a proposal..."): 134 client.add_global_proposal(resolved_key, global_params, cid) 135 context.info("Proposal added.") 136 137 138def get_valid_voting_keys( 139 ctx: CustomCtx, 140 client: CommuneClient, 141 threshold: int = 25000000000, # 25 $COMAI 142) -> dict[str, int]: 143 local_keys = local_key_addresses(ctx=ctx, universal_password=None) 144 keys_stake = local_keys_to_stakedbalance(client, local_keys) 145 keys_stake = {key: stake for key, 146 stake in keys_stake.items() if stake >= threshold} 147 return keys_stake 148 149 150@network_app.command() 151def vote_proposal( 152 ctx: Context, 153 proposal_id: int, 154 key: Optional[str] = None, 155 agree: bool = typer.Option(True, "--disagree"), 156): 157 """ 158 Casts a vote on a specified proposal. Without specifying a key, all keys on disk will be used. 159 """ 160 context = make_custom_context(ctx) 161 client = context.com_client() 162 163 if key is None: 164 context.info("Voting with all keys on disk...") 165 delegators = client.get_voting_power_delegators() 166 keys_stake = get_valid_voting_keys(context, client) 167 keys_stake = { 168 key: stake for key, 169 stake in keys_stake.items() if key not in delegators 170 } 171 else: 172 keys_stake = {key: None} 173 174 for voting_key in track(keys_stake.keys(), description="Voting..."): 175 176 resolved_key = try_classic_load_key(voting_key, context) 177 try: 178 client.vote_on_proposal(resolved_key, proposal_id, agree) 179 except Exception as e: 180 print(f"Error while voting with key {key}: ", e) 181 print("Skipping...") 182 continue 183 184 185@network_app.command() 186def unvote_proposal(ctx: Context, key: str, proposal_id: int): 187 """ 188 Retracts a previously cast vote on a specified proposal. 189 """ 190 context = make_custom_context(ctx) 191 client = context.com_client() 192 193 resolved_key = try_classic_load_key(key, context) 194 with context.progress_status(f"Unvoting on a proposal {proposal_id}..."): 195 client.unvote_on_proposal(resolved_key, proposal_id) 196 197 198@network_app.command() 199def add_custom_proposal(ctx: Context, key: str, cid: str): 200 """ 201 Adds a custom proposal. 202 """ 203 context = make_custom_context(ctx) 204 if not re.match(IPFS_REGEX, cid): 205 context.error(f"CID provided is invalid: {cid}") 206 exit(1) 207 else: 208 ipfs_prefix = "ipfs://" 209 cid = ipfs_prefix + cid 210 client = context.com_client() 211 # append the ipfs hash 212 ipfs_prefix = "ipfs://" 213 cid = ipfs_prefix + cid 214 215 resolved_key = try_classic_load_key(key, context) 216 217 with context.progress_status("Adding a proposal..."): 218 client.add_custom_proposal(resolved_key, cid) 219 220 221@network_app.command() 222def set_root_weights(ctx: Context, key: str): 223 """ 224 Command for rootnet validators to set the weights on subnets. 225 """ 226 227 context = make_custom_context(ctx) 228 client = context.com_client() 229 rootnet_id = 0 230 231 # Ask set new weights ? 232 with context.progress_status("Getting subnets to vote on..."): 233 # dict[netuid, subnet_names] 234 subnet_names = client.query_map_subnet_names() 235 236 choices = [f"{uid}: {name}" for uid, name in subnet_names.items()] 237 238 # Prompt user to select subnets 239 selected_subnets = typer.prompt( 240 "Select subnets to set weights for (space-separated list of UIDs)", 241 prompt_suffix="\n" + "\n".join(choices) + "\nEnter UIDs: " 242 ) 243 244 # Parse the input string into a list of integers 245 uids = [int(uid.strip()) for uid in selected_subnets.split()] 246 247 weights: list[int] = [] 248 for uid in uids: 249 weight = typer.prompt( 250 f"Enter weight for subnet {uid} ({subnet_names[uid]})", 251 type=float 252 ) 253 weights.append(weight) 254 255 typer.echo("Selected subnets and weights:") 256 for uid, weight in zip(uids, weights): 257 typer.echo(f"Subnet {uid} ({subnet_names[uid]}): {weight}") 258 259 resolved_key = try_classic_load_key(key, context) 260 261 client.vote(netuid=rootnet_id, uids=uids, weights=weights, key=resolved_key) 262 263 264@network_app.command() 265def registration_burn( 266 ctx: Context, 267 netuid: int, 268): 269 """ 270 Appraises the cost of registering a module on the Commune network. 271 """ 272 273 context = make_custom_context(ctx) 274 client = context.com_client() 275 276 burn = client.get_burn(netuid) 277 registration_cost = c_balance.from_nano(burn) 278 context.info( 279 f"The cost to register on a netuid: {netuid} is: {registration_cost} $COMAI" 280 )
network_app =
<typer.main.Typer object>
@network_app.command()
def
last_block(ctx: typer.models.Context, hash: bool = False):
24@network_app.command() 25def last_block(ctx: Context, hash: bool = False): 26 """ 27 Gets the last block 28 """ 29 context = make_custom_context(ctx) 30 client = context.com_client() 31 32 info = "number" if not hash else "hash" 33 34 block = client.get_block() 35 block_info = None 36 if block: 37 block_info = block["header"][info] 38 39 context.output(str(block_info))
Gets the last block
@network_app.command()
def
params(ctx: typer.models.Context):
42@network_app.command() 43def params(ctx: Context): 44 """ 45 Gets global params 46 """ 47 context = make_custom_context(ctx) 48 client = context.com_client() 49 50 with context.progress_status("Getting global network params ..."): 51 global_params = get_global_params(client) 52 printable_params = tranform_network_params(global_params) 53 print_table_from_plain_dict( 54 printable_params, ["Global params", "Value"], context.console 55 )
Gets global params
@network_app.command()
def
list_proposals( ctx: typer.models.Context, query_cid: bool = <typer.models.OptionInfo object>):
58@network_app.command() 59def list_proposals(ctx: Context, query_cid: bool = typer.Option(True)): 60 """ 61 Gets proposals 62 """ 63 context = make_custom_context(ctx) 64 client = context.com_client() 65 66 with context.progress_status("Getting proposals..."): 67 try: 68 proposals = client.query_map_proposals() 69 if query_cid: 70 proposals = convert_cid_on_proposal(proposals) 71 except IndexError: 72 context.info("No proposals found.") 73 return 74 75 for proposal_id, batch_proposal in proposals.items(): 76 status = batch_proposal["status"] 77 if isinstance(status, dict): 78 batch_proposal["status"] = [*status.keys()][0] 79 print_table_from_plain_dict( 80 batch_proposal, [ 81 f"Proposal id: {proposal_id}", "Params"], context.console 82 )
Gets proposals
@network_app.command()
def
propose_globally( ctx: typer.models.Context, key: str, cid: str, max_name_length: int = <typer.models.OptionInfo object>, min_name_length: int = <typer.models.OptionInfo object>, max_allowed_subnets: int = <typer.models.OptionInfo object>, max_allowed_modules: int = <typer.models.OptionInfo object>, max_registrations_per_block: int = <typer.models.OptionInfo object>, max_allowed_weights: int = <typer.models.OptionInfo object>, max_burn: int = <typer.models.OptionInfo object>, min_burn: int = <typer.models.OptionInfo object>, floor_delegation_fee: int = <typer.models.OptionInfo object>, floor_founder_share: int = <typer.models.OptionInfo object>, min_weight_stake: int = <typer.models.OptionInfo object>, curator: str = <typer.models.OptionInfo object>, proposal_cost: int = <typer.models.OptionInfo object>, proposal_expiration: int = <typer.models.OptionInfo object>, general_subnet_application_cost: int = <typer.models.OptionInfo object>, kappa: int = <typer.models.OptionInfo object>, rho: int = <typer.models.OptionInfo object>, subnet_immunity_period: int = <typer.models.OptionInfo object>):
85@network_app.command() 86def propose_globally( 87 ctx: Context, 88 key: str, 89 cid: str, 90 max_name_length: int = typer.Option(None), 91 min_name_length: int = typer.Option(None), 92 max_allowed_subnets: int = typer.Option(None), 93 max_allowed_modules: int = typer.Option(None), 94 max_registrations_per_block: int = typer.Option(None), 95 max_allowed_weights: int = typer.Option(None), 96 max_burn: int = typer.Option(None), 97 min_burn: int = typer.Option(None), 98 floor_delegation_fee: int = typer.Option(None), 99 floor_founder_share: int = typer.Option(None), 100 min_weight_stake: int = typer.Option(None), 101 curator: str = typer.Option(None), 102 proposal_cost: int = typer.Option(None), 103 proposal_expiration: int = typer.Option(None), 104 general_subnet_application_cost: int = typer.Option(None), 105 kappa: int = typer.Option(None), 106 rho: int = typer.Option(None), 107 subnet_immunity_period: int = typer.Option(None), 108): 109 provided_params = locals().copy() 110 provided_params.pop("ctx") 111 provided_params.pop("key") 112 113 provided_params = { 114 key: value for key, value in provided_params.items() if value is not None 115 } 116 """ 117 Adds a global proposal to the network. 118 """ 119 context = make_custom_context(ctx) 120 resolved_key = try_classic_load_key(key, context) 121 client = context.com_client() 122 123 provided_params = cast(NetworkParams, provided_params) 124 global_params = get_global_params(client) 125 global_params_config = global_params["governance_config"] 126 global_params["proposal_cost"] = global_params_config["proposal_cost"] # type: ignore 127 global_params["proposal_expiration"] = global_params_config["proposal_expiration"] # type: ignore 128 global_params.pop("governance_config") # type: ignore 129 global_params.update(provided_params) 130 131 if not re.match(IPFS_REGEX, cid): 132 context.error(f"CID provided is invalid: {cid}") 133 exit(1) 134 with context.progress_status("Adding a proposal..."): 135 client.add_global_proposal(resolved_key, global_params, cid) 136 context.info("Proposal added.")
def
get_valid_voting_keys( ctx: communex.cli._common.CustomCtx, client: communex.client.CommuneClient, threshold: int = 25000000000) -> dict[str, int]:
139def get_valid_voting_keys( 140 ctx: CustomCtx, 141 client: CommuneClient, 142 threshold: int = 25000000000, # 25 $COMAI 143) -> dict[str, int]: 144 local_keys = local_key_addresses(ctx=ctx, universal_password=None) 145 keys_stake = local_keys_to_stakedbalance(client, local_keys) 146 keys_stake = {key: stake for key, 147 stake in keys_stake.items() if stake >= threshold} 148 return keys_stake
@network_app.command()
def
vote_proposal( ctx: typer.models.Context, proposal_id: int, key: Optional[str] = None, agree: bool = <typer.models.OptionInfo object>):
151@network_app.command() 152def vote_proposal( 153 ctx: Context, 154 proposal_id: int, 155 key: Optional[str] = None, 156 agree: bool = typer.Option(True, "--disagree"), 157): 158 """ 159 Casts a vote on a specified proposal. Without specifying a key, all keys on disk will be used. 160 """ 161 context = make_custom_context(ctx) 162 client = context.com_client() 163 164 if key is None: 165 context.info("Voting with all keys on disk...") 166 delegators = client.get_voting_power_delegators() 167 keys_stake = get_valid_voting_keys(context, client) 168 keys_stake = { 169 key: stake for key, 170 stake in keys_stake.items() if key not in delegators 171 } 172 else: 173 keys_stake = {key: None} 174 175 for voting_key in track(keys_stake.keys(), description="Voting..."): 176 177 resolved_key = try_classic_load_key(voting_key, context) 178 try: 179 client.vote_on_proposal(resolved_key, proposal_id, agree) 180 except Exception as e: 181 print(f"Error while voting with key {key}: ", e) 182 print("Skipping...") 183 continue
Casts a vote on a specified proposal. Without specifying a key, all keys on disk will be used.
@network_app.command()
def
unvote_proposal(ctx: typer.models.Context, key: str, proposal_id: int):
186@network_app.command() 187def unvote_proposal(ctx: Context, key: str, proposal_id: int): 188 """ 189 Retracts a previously cast vote on a specified proposal. 190 """ 191 context = make_custom_context(ctx) 192 client = context.com_client() 193 194 resolved_key = try_classic_load_key(key, context) 195 with context.progress_status(f"Unvoting on a proposal {proposal_id}..."): 196 client.unvote_on_proposal(resolved_key, proposal_id)
Retracts a previously cast vote on a specified proposal.
@network_app.command()
def
add_custom_proposal(ctx: typer.models.Context, key: str, cid: str):
199@network_app.command() 200def add_custom_proposal(ctx: Context, key: str, cid: str): 201 """ 202 Adds a custom proposal. 203 """ 204 context = make_custom_context(ctx) 205 if not re.match(IPFS_REGEX, cid): 206 context.error(f"CID provided is invalid: {cid}") 207 exit(1) 208 else: 209 ipfs_prefix = "ipfs://" 210 cid = ipfs_prefix + cid 211 client = context.com_client() 212 # append the ipfs hash 213 ipfs_prefix = "ipfs://" 214 cid = ipfs_prefix + cid 215 216 resolved_key = try_classic_load_key(key, context) 217 218 with context.progress_status("Adding a proposal..."): 219 client.add_custom_proposal(resolved_key, cid)
Adds a custom proposal.
@network_app.command()
def
set_root_weights(ctx: typer.models.Context, key: str):
222@network_app.command() 223def set_root_weights(ctx: Context, key: str): 224 """ 225 Command for rootnet validators to set the weights on subnets. 226 """ 227 228 context = make_custom_context(ctx) 229 client = context.com_client() 230 rootnet_id = 0 231 232 # Ask set new weights ? 233 with context.progress_status("Getting subnets to vote on..."): 234 # dict[netuid, subnet_names] 235 subnet_names = client.query_map_subnet_names() 236 237 choices = [f"{uid}: {name}" for uid, name in subnet_names.items()] 238 239 # Prompt user to select subnets 240 selected_subnets = typer.prompt( 241 "Select subnets to set weights for (space-separated list of UIDs)", 242 prompt_suffix="\n" + "\n".join(choices) + "\nEnter UIDs: " 243 ) 244 245 # Parse the input string into a list of integers 246 uids = [int(uid.strip()) for uid in selected_subnets.split()] 247 248 weights: list[int] = [] 249 for uid in uids: 250 weight = typer.prompt( 251 f"Enter weight for subnet {uid} ({subnet_names[uid]})", 252 type=float 253 ) 254 weights.append(weight) 255 256 typer.echo("Selected subnets and weights:") 257 for uid, weight in zip(uids, weights): 258 typer.echo(f"Subnet {uid} ({subnet_names[uid]}): {weight}") 259 260 resolved_key = try_classic_load_key(key, context) 261 262 client.vote(netuid=rootnet_id, uids=uids, weights=weights, key=resolved_key)
Command for rootnet validators to set the weights on subnets.
@network_app.command()
def
registration_burn(ctx: typer.models.Context, netuid: int):
265@network_app.command() 266def registration_burn( 267 ctx: Context, 268 netuid: int, 269): 270 """ 271 Appraises the cost of registering a module on the Commune network. 272 """ 273 274 context = make_custom_context(ctx) 275 client = context.com_client() 276 277 burn = client.get_burn(netuid) 278 registration_cost = c_balance.from_nano(burn) 279 context.info( 280 f"The cost to register on a netuid: {netuid} is: {registration_cost} $COMAI" 281 )
Appraises the cost of registering a module on the Commune network.