ARCHIVE/xrpl-dev-portal
1
0
Fork 0
mirror of https://github.com/XRPLF/xrpl-dev-portal.git synced 2025-11-27 15:15:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move Build a Desktop Wallet code samples

Browse Source
This commit is contained in:
mDuo13
2023-09-22 19:22:32 -07:00
parent cd3ac90249
commit 63efabe0c2
61 changed files with 143 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

292
content/_code-samples/build-a-desktop-wallet/py/3_account.py Normal file
View File

@@ -0,0 +1,292 @@
# "Build a Wallet" tutorial, step 3: Take account input & show account info
# This step demonstrates how to parse user input into account information and
# look up that information on the XRP Ledger.
# License: MIT. https://github.com/XRPLF/xrpl-dev-portal/blob/master/LICENSE
import xrpl
import wx
import asyncio
from threading import Thread
from decimal import Decimal
class XRPLMonitorThread(Thread):
"""
A worker thread to watch for new ledger events and pass the info back to
the main frame to be shown in the UI. Using a thread lets us maintain the
responsiveness of the UI while doing work in the background.
"""
def __init__(self, url, gui):
Thread.__init__(self, daemon=True)
# Note: For thread safety, this thread should treat self.gui as
# read-only; to modify the GUI, use wx.CallAfter(...)
self.gui = gui
self.url = url
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
self.loop.set_debug(True)
def run(self):
"""
This thread runs a never-ending event-loop that monitors messages coming
from the XRPL, sending them to the GUI thread when necessary, and also
handles making requests to the XRPL when the GUI prompts them.
"""
self.loop.run_forever()
async def watch_xrpl_account(self, address, wallet=None):
"""
This is the task that opens the connection to the XRPL, then handles
incoming subscription messages by dispatching them to the appropriate
part of the GUI.
"""
self.account = address
self.wallet = wallet
async with xrpl.asyncio.clients.AsyncWebsocketClient(self.url) as self.client:
await self.on_connected()
async for message in self.client:
mtype = message.get("type")
if mtype == "ledgerClosed":
wx.CallAfter(self.gui.update_ledger, message)
elif mtype == "transaction":
response = await self.client.request(xrpl.models.requests.AccountInfo(
account=self.account,
ledger_index=message["ledger_index"]
))
wx.CallAfter(self.gui.update_account, response.result["account_data"])
async def on_connected(self):
"""
Set up initial subscriptions and populate the GUI with data from the
ledger on startup. Requires that self.client be connected first.
"""
# Set up 2 subscriptions: all new ledgers, and any new transactions that
# affect the chosen account.
response = await self.client.request(xrpl.models.requests.Subscribe(
streams=["ledger"],
accounts=[self.account]
))
# The immediate response contains details for the last validated ledger.
# We can use this to fill in that area of the GUI without waiting for a
# new ledger to close.
wx.CallAfter(self.gui.update_ledger, response.result)
# Get starting values for account info.
response = await self.client.request(xrpl.models.requests.AccountInfo(
account=self.account,
ledger_index="validated"
))
if not response.is_successful():
print("Got error from server:", response)
# This most often happens if the account in question doesn't exist
# on the network we're connected to. Better handling would be to use
# wx.CallAfter to display an error dialog in the GUI and possibly
# let the user try inputting a different account.
exit(1)
wx.CallAfter(self.gui.update_account, response.result["account_data"])
class AutoGridBagSizer(wx.GridBagSizer):
"""
Helper class for adding a bunch of items uniformly to a GridBagSizer.
"""
def __init__(self, parent):
wx.GridBagSizer.__init__(self, vgap=5, hgap=5)
self.parent = parent
def BulkAdd(self, ctrls):
"""
Given a two-dimensional iterable `ctrls`, add all the items in a grid
top-to-bottom, left-to-right, with each inner iterable being a row. Set
the total number of columns based on the longest iterable.
"""
flags = wx.EXPAND|wx.ALL|wx.RESERVE_SPACE_EVEN_IF_HIDDEN|wx.ALIGN_CENTER_VERTICAL
for x, row in enumerate(ctrls):
for y, ctrl in enumerate(row):
self.Add(ctrl, (x,y), flag=flags, border=5)
self.parent.SetSizer(self)
class TWaXLFrame(wx.Frame):
"""
Tutorial Wallet for the XRP Ledger (TWaXL)
user interface, main frame.
"""
def __init__(self, url, test_network=True):
wx.Frame.__init__(self, None, title="TWaXL", size=wx.Size(800,400))
self.test_network = test_network
# The ledger's current reserve settings. To be filled in later.
self.reserve_base = None
self.reserve_inc = None
self.build_ui()
# Pop up to ask user for their account ---------------------------------
address, wallet = self.prompt_for_account()
self.classic_address = address
# Start background thread for updates from the ledger ------------------
self.worker = XRPLMonitorThread(url, self)
self.worker.start()
self.run_bg_job(self.worker.watch_xrpl_account(address, wallet))
def build_ui(self):
"""
Called during __init__ to set up all the GUI components.
"""
main_panel = wx.Panel(self)
self.acct_info_area = wx.StaticBox(main_panel, label="Account Info")
lbl_address = wx.StaticText(self.acct_info_area, label="Classic Address:")
self.st_classic_address = wx.StaticText(self.acct_info_area, label="TBD")
lbl_xaddress = wx.StaticText(self.acct_info_area, label="X-Address:")
self.st_x_address = wx.StaticText(self.acct_info_area, label="TBD")
lbl_xrp_bal = wx.StaticText(self.acct_info_area, label="XRP Balance:")
self.st_xrp_balance = wx.StaticText(self.acct_info_area, label="TBD")
lbl_reserve = wx.StaticText(self.acct_info_area, label="XRP Reserved:")
self.st_reserve = wx.StaticText(self.acct_info_area, label="TBD")
aia_sizer = AutoGridBagSizer(self.acct_info_area)
aia_sizer.BulkAdd( ((lbl_address, self.st_classic_address),
(lbl_xaddress, self.st_x_address),
(lbl_xrp_bal, self.st_xrp_balance),
(lbl_reserve, self.st_reserve)) )
self.ledger_info = wx.StaticText(main_panel, label="Not connected")
main_sizer = wx.BoxSizer(wx.VERTICAL)
main_sizer.Add(self.acct_info_area, 1, flag=wx.EXPAND|wx.ALL, border=5)
main_sizer.Add(self.ledger_info, 1, flag=wx.EXPAND|wx.ALL, border=5)
main_panel.SetSizer(main_sizer)
def run_bg_job(self, job):
"""
Schedules a job to run asynchronously in the XRPL worker thread.
The job should be a Future (for example, from calling an async function)
"""
task = asyncio.run_coroutine_threadsafe(job, self.worker.loop)
def toggle_dialog_style(self, event):
"""
Automatically switches to a password-style dialog if it looks like the
user is entering a secret key, and display ***** instead of s12345...
"""
dlg = event.GetEventObject()
v = dlg.GetValue().strip()
if v[:1] == "s":
dlg.SetWindowStyle(wx.TE_PASSWORD)
else:
dlg.SetWindowStyle(wx.TE_LEFT)
def prompt_for_account(self):
"""
Prompt the user for an account to use, in a base58-encoded format:
- master key seed: Grants read-write access.
(assumes the master key pair is not disabled)
- classic address. Grants read-only access.
- X-address. Grants read-only access.
Exits with error code 1 if the user cancels the dialog, if the input
doesn't match any of the formats, or if the user inputs an X-address
intended for use on a different network type (test/non-test).
Populates the classic address and X-address labels in the UI.
Returns (classic_address, wallet) where wallet is None in read-only mode
"""
account_dialog = wx.TextEntryDialog(self,
"Please enter an account address (for read-only)"
" or your secret (for read-write access)",
caption="Enter account",
value="rPT1Sjq2YGrBMTttX4GZHjKu9dyfzbpAYe")
account_dialog.Bind(wx.EVT_TEXT, self.toggle_dialog_style)
if account_dialog.ShowModal() != wx.ID_OK:
# If the user presses Cancel on the account entry, exit the app.
exit(1)
value = account_dialog.GetValue().strip()
account_dialog.Destroy()
classic_address = ""
wallet = None
x_address = ""
if xrpl.core.addresscodec.is_valid_xaddress(value):
x_address = value
classic_address, dest_tag, test_network = xrpl.core.addresscodec.xaddress_to_classic_address(value)
if test_network != self.test_network:
on_net = "a test network" if self.test_network else "Mainnet"
print(f"X-address {value} is meant for a different network type"
f"than this client is connected to."
f"(Client is on: {on_net})")
exit(1)
elif xrpl.core.addresscodec.is_valid_classic_address(value):
classic_address = value
x_address = xrpl.core.addresscodec.classic_address_to_xaddress(
value, tag=None, is_test_network=self.test_network)
else:
try:
# Check if it's a valid seed
seed_bytes, alg = xrpl.core.addresscodec.decode_seed(value)
wallet = xrpl.wallet.Wallet.from_seed(seed=value)
x_address = wallet.get_xaddress(is_test=self.test_network)
classic_address = wallet.address
except Exception as e:
print(e)
exit(1)
# Update the UI with the address values
self.st_classic_address.SetLabel(classic_address)
self.st_x_address.SetLabel(x_address)
return classic_address, wallet
def update_ledger(self, message):
"""
Process a ledger subscription message to update the UI with
information about the latest validated ledger.
"""
close_time_iso = xrpl.utils.ripple_time_to_datetime(message["ledger_time"]).isoformat()
self.ledger_info.SetLabel(f"Latest validated ledger:\n"
f"Ledger Index: {message['ledger_index']}\n"
f"Ledger Hash: {message['ledger_hash']}\n"
f"Close time: {close_time_iso}")
# Save reserve settings so we can calculate account reserve
self.reserve_base = xrpl.utils.drops_to_xrp(str(message["reserve_base"]))
self.reserve_inc = xrpl.utils.drops_to_xrp(str(message["reserve_inc"]))
def calculate_reserve_xrp(self, owner_count):
"""
Calculates how much XRP the user needs to reserve based on the account's
OwnerCount and the reserve values in the latest ledger.
"""
if self.reserve_base == None or self.reserve_inc == None:
return None
oc_decimal = Decimal(owner_count)
reserve_xrp = self.reserve_base + (self.reserve_inc * oc_decimal)
return reserve_xrp
def update_account(self, acct):
"""
Update the account info UI based on an account_info response.
"""
xrp_balance = str(xrpl.utils.drops_to_xrp(acct["Balance"]))
self.st_xrp_balance.SetLabel(xrp_balance)
# Display account reserve.
reserve_xrp = self.calculate_reserve_xrp(acct.get("OwnerCount", 0))
if reserve_xrp != None:
self.st_reserve.SetLabel(str(reserve_xrp))
if __name__ == "__main__":
WS_URL = "wss://s.altnet.rippletest.net:51233" # Testnet
app = wx.App()
frame = TWaXLFrame(WS_URL, test_network=True)
frame.Show()
app.MainLoop()