import asyncio import sys import time from datetime import datetime from decimal import Decimal from typing import Callable, List, Optional, Tuple import aiohttp from chia.cmds.units import units from chia.rpc.wallet_rpc_client import WalletRpcClient from chia.server.start_wallet import SERVICE_NAME from chia.util.bech32m import encode_puzzle_hash from chia.util.byte_types import hexstr_to_bytes from chia.util.config import load_config from chia.util.default_root import DEFAULT_ROOT_PATH from chia.util.ints import uint16, uint64 from chia.wallet.transaction_record import TransactionRecord from chia.wallet.util.wallet_types import WalletType def print_transaction(tx: TransactionRecord, verbose: bool, name) -> None: if verbose: print(tx) else: chia_amount = Decimal(int(tx.amount)) / units["chia"] to_address = encode_puzzle_hash(tx.to_puzzle_hash, name) print(f"Transaction {tx.name}") print(f"Status: {'Confirmed' if tx.confirmed else ('In mempool' if tx.is_in_mempool() else 'Pending')}") print(f"Amount: {chia_amount} {name}") print(f"To address: {to_address}") print("Created at:", datetime.fromtimestamp(tx.created_at_time).strftime("%Y-%m-%d %H:%M:%S")) print("") async def get_transaction(args: dict, wallet_client: WalletRpcClient, fingerprint: int) -> None: wallet_id = args["id"] transaction_id = hexstr_to_bytes(args["tx_id"]) config = load_config(DEFAULT_ROOT_PATH, "config.yaml", SERVICE_NAME) name = config["network_overrides"]["config"][config["selected_network"]]["address_prefix"] tx: TransactionRecord = await wallet_client.get_transaction(wallet_id, transaction_id=transaction_id) print_transaction(tx, verbose=(args["verbose"] > 0), name=name) async def get_transactions(args: dict, wallet_client: WalletRpcClient, fingerprint: int) -> None: wallet_id = args["id"] txs: List[TransactionRecord] = await wallet_client.get_transactions(wallet_id) config = load_config(DEFAULT_ROOT_PATH, "config.yaml", SERVICE_NAME) name = config["network_overrides"]["config"][config["selected_network"]]["address_prefix"] if len(txs) == 0: print("There are no transactions to this address") offset = args["offset"] num_per_screen = 5 for i in range(offset, len(txs), num_per_screen): for j in range(0, num_per_screen): if i + j >= len(txs): break print_transaction(txs[i + j], verbose=(args["verbose"] > 0), name=name) if i + num_per_screen >= len(txs): return print("Press q to quit, or c to continue") while True: entered_key = sys.stdin.read(1) if entered_key == "q": return elif entered_key == "c": break async def send(args: dict, wallet_client: WalletRpcClient, fingerprint: int) -> None: wallet_id = args["id"] amount = Decimal(args["amount"]) fee = Decimal(args["fee"]) address = args["address"] print("Submitting transaction...") final_amount = uint64(int(amount * units["chia"])) final_fee = uint64(int(fee * units["chia"])) res = await wallet_client.send_transaction(wallet_id, final_amount, address, final_fee) tx_id = res.name start = time.time() while time.time() - start < 10: await asyncio.sleep(0.1) tx = await wallet_client.get_transaction(wallet_id, tx_id) if len(tx.sent_to) > 0: print(f"Transaction submitted to nodes: {tx.sent_to}") print(f"Do chia wallet get_transaction -f {fingerprint} -tx 0x{tx_id} to get status") return print("Transaction not yet submitted to nodes") print(f"Do 'chia wallet get_transaction -f {fingerprint} -tx 0x{tx_id}' to get status") async def get_address(args: dict, wallet_client: WalletRpcClient, fingerprint: int) -> None: wallet_id = args["id"] res = await wallet_client.get_next_address(wallet_id, False) print(res) async def print_balances(args: dict, wallet_client: WalletRpcClient, fingerprint: int) -> None: summaries_response = await wallet_client.get_wallets() config = load_config(DEFAULT_ROOT_PATH, "config.yaml") address_prefix = config["network_overrides"]["config"][config["selected_network"]]["address_prefix"] print(f"Wallet height: {await wallet_client.get_height_info()}") print(f"Balances, fingerprint: {fingerprint}") for summary in summaries_response: wallet_id = summary["id"] balances = await wallet_client.get_wallet_balance(wallet_id) typ = WalletType(int(summary["type"])).name if typ != "STANDARD_WALLET": print(f"Wallet ID {wallet_id} type {typ} {summary['name']}") print(f" -Confirmed: " f"{balances['confirmed_wallet_balance']/units['colouredcoin']}") print(f" -Confirmed - Pending Outgoing: {balances['unconfirmed_wallet_balance']/units['colouredcoin']}") print(f" -Spendable: {balances['spendable_balance']/units['colouredcoin']}") print(f" -Pending change: {balances['pending_change']/units['colouredcoin']}") else: print(f"Wallet ID {wallet_id} type {typ}") print( f" -Confirmed: {balances['confirmed_wallet_balance']} mojo " f"({balances['confirmed_wallet_balance']/units['chia']} {address_prefix})" ) print( f" -Unconfirmed: {balances['unconfirmed_wallet_balance']} mojo " f"({balances['unconfirmed_wallet_balance']/units['chia']} {address_prefix})" ) print( f" -Spendable: {balances['spendable_balance']} mojo " f"({balances['spendable_balance']/units['chia']} {address_prefix})" ) print( f" -Pending change: {balances['pending_change']} mojo " f"({balances['pending_change']/units['chia']} {address_prefix})" ) async def get_wallet(wallet_client: WalletRpcClient, fingerprint: int = None) -> Optional[Tuple[WalletRpcClient, int]]: fingerprints = await wallet_client.get_public_keys() if len(fingerprints) == 0: print("No keys loaded. Run 'chia keys generate' or import a key") return None if fingerprint is not None: if fingerprint not in fingerprints: print(f"Fingerprint {fingerprint} does not exist") return None if len(fingerprints) == 1: fingerprint = fingerprints[0] if fingerprint is not None: log_in_response = await wallet_client.log_in(fingerprint) else: print("Choose wallet key:") for i, fp in enumerate(fingerprints): print(f"{i+1}) {fp}") val = None while val is None: val = input("Enter a number to pick or q to quit: ") if val == "q": return None if not val.isdigit(): val = None else: index = int(val) - 1 if index >= len(fingerprints): print("Invalid value") val = None continue else: fingerprint = fingerprints[index] assert fingerprint is not None log_in_response = await wallet_client.log_in(fingerprint) if log_in_response["success"] is False: if log_in_response["error"] == "not_initialized": use_cloud = True if "backup_path" in log_in_response: path = log_in_response["backup_path"] print(f"Backup file from backup.chia.net downloaded and written to: {path}") val = input("Do you want to use this file to restore from backup? (Y/N) ") if val.lower() == "y": log_in_response = await wallet_client.log_in_and_restore(fingerprint, path) else: use_cloud = False if "backup_path" not in log_in_response or use_cloud is False: if use_cloud is True: val = input( "No online backup file found, \n Press S to skip restore from backup" " \n Press F to use your own backup file: " ) else: val = input( "Cloud backup declined, \n Press S to skip restore from backup" " \n Press F to use your own backup file: " ) if val.lower() == "s": log_in_response = await wallet_client.log_in_and_skip(fingerprint) elif val.lower() == "f": val = input("Please provide the full path to your backup file: ") log_in_response = await wallet_client.log_in_and_restore(fingerprint, val) if "success" not in log_in_response or log_in_response["success"] is False: if "error" in log_in_response: error = log_in_response["error"] print(f"Error: {log_in_response[error]}") return None return wallet_client, fingerprint async def execute_with_wallet(wallet_rpc_port: int, fingerprint: int, extra_params: dict, function: Callable) -> None: try: config = load_config(DEFAULT_ROOT_PATH, "config.yaml") self_hostname = config["self_hostname"] if wallet_rpc_port is None: wallet_rpc_port = config["wallet"]["rpc_port"] wallet_client = await WalletRpcClient.create(self_hostname, uint16(wallet_rpc_port), DEFAULT_ROOT_PATH, config) wallet_client_f = await get_wallet(wallet_client, fingerprint=fingerprint) if wallet_client_f is None: wallet_client.close() await wallet_client.await_closed() return wallet_client, fingerprint = wallet_client_f await function(extra_params, wallet_client, fingerprint) except KeyboardInterrupt: pass except Exception as e: if isinstance(e, aiohttp.client_exceptions.ClientConnectorError): print(f"Connection error. Check if wallet is running at {wallet_rpc_port}") else: print(f"Exception from 'wallet' {e}") wallet_client.close() await wallet_client.await_closed()