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