# "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(seed=value, sequence=0) x_address = wallet.get_xaddress(is_test=self.test_network) classic_address = wallet.classic_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()