240 lines
10 KiB
Python
240 lines
10 KiB
Python
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 None
|
|
print("Press q to quit, or c to continue")
|
|
while True:
|
|
entered_key = sys.stdin.read(1)
|
|
if entered_key == "q":
|
|
return None
|
|
elif entered_key == "c":
|
|
break
|
|
|
|
|
|
def check_unusual_transaction(amount: Decimal, fee: Decimal):
|
|
return fee >= amount
|
|
|
|
|
|
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"]
|
|
override = args["override"]
|
|
|
|
if not override and check_unusual_transaction(amount, fee):
|
|
print(
|
|
f"A transaction of amount {amount} and fee {fee} is unusual.\n"
|
|
f"Pass in --override if you are sure you mean to do this."
|
|
)
|
|
return
|
|
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 None
|
|
|
|
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"Sync status: {'Synced' if (await wallet_client.get_synced()) else 'Not synced'}")
|
|
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" -Total Balance: " f"{balances['confirmed_wallet_balance']/units['colouredcoin']}")
|
|
print(f" -Pending Total Balance: {balances['unconfirmed_wallet_balance']/units['colouredcoin']}")
|
|
print(f" -Spendable Balance: {balances['spendable_balance']/units['colouredcoin']}")
|
|
else:
|
|
print(f"Wallet ID {wallet_id} type {typ}")
|
|
print(
|
|
f" -Total Balance: {balances['confirmed_wallet_balance']/units['chia']} {address_prefix} "
|
|
f"({balances['confirmed_wallet_balance']} mojo)"
|
|
)
|
|
print(
|
|
f" -Pending Total Balance: {balances['unconfirmed_wallet_balance']/units['chia']} {address_prefix} "
|
|
f"({balances['unconfirmed_wallet_balance']} mojo)"
|
|
)
|
|
print(
|
|
f" -Spendable: {balances['spendable_balance']/units['chia']} {address_prefix} "
|
|
f"({balances['spendable_balance']} mojo)"
|
|
)
|
|
|
|
|
|
async def get_wallet(wallet_client: WalletRpcClient, fingerprint: int = None) -> Optional[Tuple[WalletRpcClient, int]]:
|
|
if fingerprint is not None:
|
|
fingerprints = [fingerprint]
|
|
else:
|
|
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 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 None
|
|
wallet_client, fingerprint = wallet_client_f
|
|
await function(extra_params, wallet_client, fingerprint)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
except Exception as e:
|
|
if isinstance(e, aiohttp.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()
|