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