communex.cli.balance

  1import re
  2from typing import Optional
  3
  4import typer
  5from typer import Context
  6
  7from communex._common import IPFS_REGEX, BalanceUnit, format_balance
  8from communex.balance import to_nano
  9from communex.cli._common import (
 10    make_custom_context,
 11    print_table_from_plain_dict,
 12)
 13from communex.errors import ChainTransactionError
 14from communex.faucet.powv2 import solve_for_difficulty_fast
 15
 16balance_app = typer.Typer(no_args_is_help=True)
 17
 18
 19@balance_app.command()
 20def free_balance(
 21    ctx: Context,
 22    key: str,
 23    unit: BalanceUnit = BalanceUnit.joule,
 24    password: Optional[str] = None,
 25):
 26    """
 27    Gets free balance of a key.
 28    """
 29    context = make_custom_context(ctx)
 30    client = context.com_client()
 31
 32    key_address = context.resolve_key_ss58(key, password)
 33
 34    with context.progress_status(
 35        f"Getting free balance of key {key_address}..."
 36    ):
 37        balance = client.get_balance(key_address)
 38
 39    context.output(format_balance(balance, unit))
 40
 41
 42@balance_app.command()
 43def staked_balance(
 44    ctx: Context,
 45    key: str,
 46    unit: BalanceUnit = BalanceUnit.joule,
 47    password: Optional[str] = None,
 48):
 49    """
 50    Gets the balance staked on the key itself.
 51    """
 52    context = make_custom_context(ctx)
 53    client = context.com_client()
 54
 55    key_address = context.resolve_key_ss58(key, password)
 56
 57    with context.progress_status(
 58        f"Getting staked balance of key {key_address}..."
 59    ):
 60        result = sum(client.get_staketo(key=key_address).values())
 61
 62    context.output(format_balance(result, unit))
 63
 64
 65@balance_app.command()
 66def show(
 67    ctx: Context,
 68    key: str,
 69    unit: BalanceUnit = BalanceUnit.joule,
 70    password: Optional[str] = None,
 71):
 72    """
 73    Gets entire balance of a key (free balance + staked balance).
 74    """
 75    context = make_custom_context(ctx)
 76    client = context.com_client()
 77
 78    key_address = context.resolve_key_ss58(key, password)
 79
 80    with context.progress_status(f"Getting value of key {key_address}..."):
 81        staked_balance = sum(client.get_staketo(key=key_address).values())
 82        free_balance = client.get_balance(key_address)
 83        balance_sum = free_balance + staked_balance
 84
 85    print_table_from_plain_dict(
 86        {
 87            "Free": format_balance(free_balance, unit),
 88            "Staked": format_balance(staked_balance, unit),
 89            "Total": format_balance(balance_sum, unit),
 90        },
 91        ["Result", "Amount"],
 92        context.console,
 93    )
 94
 95
 96@balance_app.command()
 97def get_staked(
 98    ctx: Context,
 99    key: str,
100    unit: BalanceUnit = BalanceUnit.joule,
101    password: Optional[str] = None,
102):
103    """
104    Gets total stake of a key it delegated across other keys.
105    """
106    context = make_custom_context(ctx)
107    client = context.com_client()
108
109    key_address = context.resolve_key_ss58(key, password)
110
111    with context.progress_status(f"Getting stake of {key_address}..."):
112        result = sum(client.get_staketo(key=key_address).values())
113
114    context.output(format_balance(result, unit))
115
116
117@balance_app.command()
118def transfer(ctx: Context, key: str, amount: float, dest: str):
119    """
120    Transfer amount to destination using key
121    """
122    context = make_custom_context(ctx)
123    client = context.com_client()
124
125    nano_amount = to_nano(amount)
126
127    resolved_key = context.load_key(key, None)
128    resolved_dest = context.resolve_key_ss58(dest, None)
129
130    if not context.confirm(
131        f"Are you sure you want to transfer {amount} tokens to {dest}?"
132    ):
133        raise typer.Abort()
134
135    with context.progress_status(f"Transferring {amount} tokens to {dest}..."):
136        response = client.transfer(
137            key=resolved_key, amount=nano_amount, dest=resolved_dest
138        )
139
140    if response.is_success:
141        context.info(f"Transferred {amount} tokens to {dest}")
142    else:
143        raise ChainTransactionError(response.error_message)  # type: ignore
144
145
146@balance_app.command()
147def transfer_stake(
148    ctx: Context, key: str, amount: float, from_key: str, dest: str
149):
150    """
151    Transfers stake of key from point A to point B
152    """
153    context = make_custom_context(ctx)
154    client = context.com_client()
155
156    nano_amount = to_nano(amount)
157    keypair = context.load_key(key, None)
158    resolved_from = context.resolve_key_ss58(from_key)
159    resolved_dest = context.resolve_key_ss58(dest)
160
161    with context.progress_status(
162        f"Transferring {amount} tokens from {from_key} to {dest}' ..."
163    ):
164        response = client.transfer_stake(
165            key=keypair,
166            amount=nano_amount,
167            from_module_key=resolved_from,
168            dest_module_address=resolved_dest,
169        )
170
171    if response.is_success:
172        context.info(f"Transferred {amount} tokens from {from_key} to {dest}")
173    else:
174        raise ChainTransactionError(response.error_message)  # type: ignore
175
176
177@balance_app.command()
178def stake(
179    ctx: Context,
180    key: str,
181    amount: float,
182    dest: str,
183):
184    """
185    Stake amount to destination using key
186    """
187    context = make_custom_context(ctx)
188    client = context.com_client()
189
190    nano_amount = to_nano(amount)
191    keypair = context.load_key(key, None)
192    resolved_dest = context.resolve_key_ss58(dest, None)
193
194    delegating_message = (
195        "By default you delegate DAO "
196        "voting power to the validator you stake to. "
197        "In case you want to change this, call: "
198        "`comx key power-delegation <key> --disable`."
199    )
200    context.info("INFO: ", style="bold green", end="")  # type: ignore
201    context.info(delegating_message)  # type: ignore
202
203    with context.progress_status(f"Staking {amount} tokens to {dest}..."):
204        response = client.stake(
205            key=keypair, amount=nano_amount, dest=resolved_dest
206        )
207
208    if response.is_success:
209        context.info(f"Staked {amount} tokens to {dest}")
210    else:
211        raise ChainTransactionError(response.error_message)  # type: ignore
212
213
214@balance_app.command()
215def unstake(ctx: Context, key: str, amount: float, dest: str):
216    """
217    Unstake amount from destination using key
218    """
219    context = make_custom_context(ctx)
220    client = context.com_client()
221
222    nano_amount = to_nano(amount)
223    keypair = context.load_key(key, None)
224    resolved_dest = context.resolve_key_ss58(dest, None)
225
226    with context.progress_status(f"Unstaking {amount} tokens from {dest}'..."):
227        response = client.unstake(
228            key=keypair, amount=nano_amount, dest=resolved_dest
229        )  # TODO: is it right?
230
231    if response.is_success:
232        context.info(f"Unstaked {amount} tokens from {dest}")
233    else:
234        raise ChainTransactionError(response.error_message)  # type: ignore
235
236
237@balance_app.command()
238def run_faucet(
239    ctx: Context,
240    key: str,
241    num_processes: Optional[int] = None,
242    num_executions: int = 1,
243):
244    context = make_custom_context(ctx)
245    use_testnet = ctx.obj.use_testnet
246
247    if not use_testnet:
248        context.error("Faucet only enabled on testnet")
249        raise typer.Exit(code=1)
250
251    resolved_key = context.load_key(key, None)
252
253    client = context.com_client()
254    for _ in range(num_executions):
255        with context.progress_status("Solving PoW..."):
256            solution = solve_for_difficulty_fast(
257                client,
258                resolved_key,
259                client.url,
260                num_processes=num_processes,
261            )
262        with context.progress_status("Sending solution to blockchain"):
263            params = {
264                "block_number": solution.block_number,
265                "nonce": solution.nonce,
266                "work": solution.seal,
267                "key": resolved_key.ss58_address,
268            }
269            client.compose_call(
270                "faucet",
271                params=params,
272                unsigned=True,
273                module="FaucetModule",
274                key=resolved_key.ss58_address,  # type: ignore
275            )
276
277
278@balance_app.command()
279def transfer_dao_funds(
280    ctx: Context,
281    signer_key: str,
282    amount: float,
283    cid_hash: str,
284    dest: str,
285):
286    context = make_custom_context(ctx)
287
288    if not re.match(IPFS_REGEX, cid_hash):
289        context.error(f"CID provided is invalid: {cid_hash}")
290        raise typer.Exit(code=1)
291
292    ipfs_prefix = "ipfs://"
293    cid = ipfs_prefix + cid_hash
294
295    nano_amount = to_nano(amount)
296    keypair = context.load_key(signer_key, None)
297    dest = context.resolve_key_ss58(dest, None)
298
299    client = context.com_client()
300    client.add_transfer_dao_treasury_proposal(keypair, cid, nano_amount, dest)