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    )