communex.cli.balance

  1import re
  2from typing import Optional, cast
  3
  4import typer
  5from typer import Context
  6
  7from communex._common import IPFS_REGEX, BalanceUnit, format_balance
  8from communex.balance import from_nano, to_nano
  9from communex.cli._common import (
 10    make_custom_context,
 11    print_table_from_plain_dict,
 12)
 13from communex.compat.key import local_key_addresses
 14from communex.errors import ChainTransactionError
 15from communex.faucet.powv2 import solve_for_difficulty_fast
 16
 17balance_app = typer.Typer(no_args_is_help=True)
 18
 19
 20@balance_app.command()
 21def free_balance(
 22    ctx: Context,
 23    key: str,
 24    unit: BalanceUnit = BalanceUnit.joule,
 25    password: Optional[str] = None,
 26):
 27    """
 28    Gets free balance of a key.
 29    """
 30    context = make_custom_context(ctx)
 31    client = context.com_client()
 32
 33    key_address = context.resolve_key_ss58(key, password)
 34
 35    with context.progress_status(
 36        f"Getting free balance of key {key_address}..."
 37    ):
 38        balance = client.get_balance(key_address)
 39
 40    context.output(format_balance(balance, unit))
 41
 42
 43@balance_app.command()
 44def staked_balance(
 45    ctx: Context,
 46    key: str,
 47    unit: BalanceUnit = BalanceUnit.joule,
 48    password: Optional[str] = None,
 49):
 50    """
 51    Gets the balance staked on the key itself.
 52    """
 53    context = make_custom_context(ctx)
 54    client = context.com_client()
 55
 56    key_address = context.resolve_key_ss58(key, password)
 57
 58    with context.progress_status(
 59        f"Getting staked balance of key {key_address}..."
 60    ):
 61        result = sum(client.get_staketo(key=key_address).values())
 62
 63    context.output(format_balance(result, unit))
 64
 65
 66@balance_app.command()
 67def show(
 68    ctx: Context,
 69    key: str,
 70    unit: BalanceUnit = BalanceUnit.joule,
 71    password: Optional[str] = None,
 72):
 73    """
 74    Gets entire balance of a key (free balance + staked balance).
 75    """
 76    context = make_custom_context(ctx)
 77    client = context.com_client()
 78
 79    key_address = context.resolve_key_ss58(key, password)
 80
 81    with context.progress_status(f"Getting value of key {key_address}..."):
 82        staked_balance = sum(client.get_staketo(key=key_address).values())
 83        free_balance = client.get_balance(key_address)
 84        balance_sum = free_balance + staked_balance
 85
 86    print_table_from_plain_dict(
 87        {
 88            "Free": format_balance(free_balance, unit),
 89            "Staked": format_balance(staked_balance, unit),
 90            "Total": format_balance(balance_sum, unit),
 91        },
 92        ["Result", "Amount"],
 93        context.console,
 94    )
 95
 96
 97@balance_app.command()
 98def get_staked(
 99    ctx: Context,
100    key: str,
101    unit: BalanceUnit = BalanceUnit.joule,
102    password: Optional[str] = None,
103):
104    """
105    Gets total stake of a key it delegated across other keys.
106    """
107    context = make_custom_context(ctx)
108    client = context.com_client()
109
110    key_address = context.resolve_key_ss58(key, password)
111
112    with context.progress_status(f"Getting stake of {key_address}..."):
113        result = sum(client.get_staketo(key=key_address).values())
114
115    context.output(format_balance(result, unit))
116
117
118@balance_app.command()
119def transfer(ctx: Context, key: str, amount: float, dest: str):
120    """
121    Transfer amount to destination using key
122    """
123    context = make_custom_context(ctx)
124    client = context.com_client()
125
126    nano_amount = to_nano(amount)
127
128    resolved_key = context.load_key(key, None)
129    resolved_dest = context.resolve_key_ss58(dest, None)
130
131    if not context.confirm(
132        f"Are you sure you want to transfer {amount} tokens to {dest}?"
133    ):
134        raise typer.Abort()
135
136    with context.progress_status(f"Transferring {amount} tokens to {dest}..."):
137        response = client.transfer(
138            key=resolved_key, amount=nano_amount, dest=resolved_dest
139        )
140
141    if response.is_success:
142        context.info(f"Transferred {amount} tokens to {dest}")
143    else:
144        raise ChainTransactionError(response.error_message)  # type: ignore
145
146
147@balance_app.command()
148def transfer_stake(
149    ctx: Context, key: str, amount: float, from_key: str, dest: str
150):
151    """
152    Transfers stake of key from point A to point B
153    """
154    context = make_custom_context(ctx)
155    client = context.com_client()
156
157    nano_amount = to_nano(amount)
158    keypair = context.load_key(key, None)
159    resolved_from = context.resolve_key_ss58(from_key)
160    resolved_dest = context.resolve_key_ss58(dest)
161
162    with context.progress_status(
163        f"Transferring {amount} tokens from {from_key} to {dest}' ..."
164    ):
165        response = client.transfer_stake(
166            key=keypair,
167            amount=nano_amount,
168            from_module_key=resolved_from,
169            dest_module_address=resolved_dest,
170        )
171
172    if response.is_success:
173        context.info(f"Transferred {amount} tokens from {from_key} to {dest}")
174    else:
175        raise ChainTransactionError(response.error_message)  # type: ignore
176
177
178@balance_app.command()
179def stake(
180    ctx: Context,
181    key: str,
182    amount: float,
183    dest: str,
184):
185    """
186    Stake amount to destination using key
187    """
188    context = make_custom_context(ctx)
189    client = context.com_client()
190
191    nano_amount = to_nano(amount)
192    keypair = context.load_key(key, None)
193    resolved_dest = context.resolve_key_ss58(dest, None)
194
195    delegating_message = (
196        "By default you delegate DAO "
197        "voting power to the validator you stake to. "
198        "In case you want to change this, call: "
199        "`comx key power-delegation <key> --disable`."
200    )
201    context.info("INFO: ", style="bold green", end="")  # type: ignore
202    context.info(delegating_message)  # type: ignore
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)
301
302
303@balance_app.command()
304def bridge(
305    ctx: Context,
306    key: str,
307    amount: float,
308):
309    context = make_custom_context(ctx)
310    client = context.com_client()
311
312    nano_amount = to_nano(amount)
313    keypair = context.load_key(key, None)
314
315    with context.progress_status(f"Bridging {amount} tokens..."):
316        try:
317            client.bridge(keypair, nano_amount)
318        except Exception as e:
319            context.error(f"Failed to bridge {amount} tokens: {e}")
320        else:
321            context.info(f"Bridged {amount}$j successfully")
322
323
324@balance_app.command()
325def bridge_withdraw(
326    ctx: Context,
327    key: str,
328    amount: float,
329):
330    context = make_custom_context(ctx)
331    client = context.com_client()
332
333    nano_amount = to_nano(amount)
334    keypair = context.load_key(key, None)
335
336    with context.progress_status(f"Withdrawing {amount} tokens..."):
337        try:
338            client.bridge_withdraw(keypair, nano_amount)
339        except Exception as e:
340            context.error(f"Failed to withdraw {amount} tokens: {e}")
341        else:
342            context.info(f"Withdrew {amount}$j successfully")
343
344
345@balance_app.command()
346def bridged_balance(
347    ctx: Context,
348    key: Optional[str] = None,
349):
350    context = make_custom_context(ctx)
351    client = context.com_client()
352    with context.progress_status("Getting bridged balance..."):
353        bridge_map = client.query_map(
354            "Bridged", params=[], extract_value=False
355        )["Bridged"]
356        bridge_map = cast(dict[str, int], bridge_map)
357    if key is None:
358        local_keys = local_key_addresses(context.password_manager)
359        local_bridge_map = {
360            address: from_nano(ammount)
361            for address, ammount in bridge_map.items()
362            if address in local_keys.values()
363        }
364    else:
365        key = cast(str, context.resolve_key_ss58(key, None))
366        bridged_amount = from_nano(bridge_map.get(key, 0))
367        local_bridge_map = {key: bridged_amount}
368
369    print_table_from_plain_dict(
370        local_bridge_map, ["Address", "Amount"], context.console
371    )