diff --git a/content/_code-samples/build-a-wallet/py/7_owned_objects.py b/content/_code-samples/build-a-wallet/py/7_owned_objects.py new file mode 100644 index 0000000000..5e3f106957 --- /dev/null +++ b/content/_code-samples/build-a-wallet/py/7_owned_objects.py @@ -0,0 +1,889 @@ +# "Build a Wallet" tutorial, extra step: Tokens and Other objects +# Show (issued / fungible) tokens and other objects owned by an account. +# License: MIT. https://github.com/XRPLF/xrpl-dev-portal/blob/master/LICENSE + +import xrpl +import wx +import wx.dataview +import wx.adv +import asyncio +import re +from threading import Thread +from decimal import Decimal + +from verify_domain import verify_account_domain + +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": + wx.CallAfter(self.gui.add_tx_from_sub, message) + 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"]) + if self.wallet: + wx.CallAfter(self.gui.enable_readwrite) + # Get the first page of the account's transaction history. Depending on + # the server we're connected to, the account's full history may not be + # available. + response = await self.client.request(xrpl.models.requests.AccountTx( + account=self.account + )) + wx.CallAfter(self.gui.update_account_tx, response.result) + # Look up issued tokens + response = await self.client.request(xrpl.models.requests.AccountLines( + account=self.account, + ledger_index="validated" + )) + if not response.is_successful(): + print("Error getting account lines:", response) + else: + wx.CallAfter(self.gui.update_account_lines, + response.result["lines"]) + # Look up all types of objects attached to the account + response = await self.client.request(xrpl.models.requests.AccountObjects( + account=self.account, + ledger_index="validated" + )) + if not response.is_successful(): + print("Error getting account objects:", response) + else: + wx.CallAfter(self.gui.update_account_objects, + response.result["account_objects"]) + + async def check_destination(self, destination, dlg): + """ + Check a potential destination address's details, and pass them back to + a "Send XRP" dialog: + - Is the account funded? + If not, payments below the reserve base will fail + - Do they have DisallowXRP enabled? + If so, the user should be warned they don't want XRP, but can click + through. + - Do they have a verified Domain? + If so, we want to show the user the associated domain info. + + Requires that self.client be connected first. + """ + + # The data to send back to the GUI thread: None for checks that weren't + # performed, True/False for actual results except where noted. + account_status = { + "funded": None, + "disallow_xrp": None, + "domain_verified": None, + "domain_str": "" # the decoded domain, regardless of verification + } + + # Look up the account. If this fails, the account isn't funded. + try: + response = await xrpl.asyncio.account.get_account_info(destination, + self.client, ledger_index="validated") + account_status["funded"] = True + dest_acct = response.result["account_data"] + except xrpl.asyncio.clients.exceptions.XRPLRequestFailureException: + # Not funded, so the other checks don't apply. + account_status["funded"] = False + wx.CallAfter(dlg.update_dest_info, account_status) + return + + # Check DisallowXRP flag + lsfDisallowXRP = 0x00080000 + if dest_acct["Flags"] & lsfDisallowXRP: + account_status["disallow_xrp"] = True + else: + account_status["disallow_xrp"] = False + + # Check domain verification + domain, verified = verify_account_domain(dest_acct) + account_status["domain_verified"] = verified + account_status["domain_str"] = domain + + # Send data back to the main thread. + wx.CallAfter(dlg.update_dest_info, account_status) + + async def send_xrp(self, paydata): + """ + Prepare, sign, and send an XRP payment with the provided parameters. + Expects a dictionary with: + { + "dtag": Destination Tag, as a string, optional + "to": Destination address (classic or X-address) + "amt": Amount of decimal XRP to send, as a string + } + """ + dtag = paydata.get("dtag", "") + if dtag.strip() == "": + dtag = None + if dtag is not None: + try: + dtag = int(dtag) + if dtag < 0 or dtag > 2**32-1: + raise ValueError("Destination tag must be a 32-bit unsigned integer") + except ValueError as e: + print("Invalid destination tag:", e) + print("Canceled sending payment.") + return + + tx = xrpl.models.transactions.Payment( + account=self.account, + destination=paydata["to"], + amount=xrpl.utils.xrp_to_drops(paydata["amt"]), + destination_tag=dtag + ) + # Autofill provides a sequence number, but this may fail if you try to + # send too many transactions too fast. You can send transactions more + # rapidly if you track the sequence number more carefully. + tx_signed = await xrpl.asyncio.transaction.safe_sign_and_autofill_transaction( + tx, self.wallet, self.client) + await xrpl.asyncio.transaction.submit_transaction(tx_signed, self.client) + wx.CallAfter(self.gui.add_pending_tx, tx_signed) + + +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 SendXRPDialog(wx.Dialog): + """ + Pop-up dialog that prompts the user for the information necessary to send a + direct XRP-to-XRP payment on the XRPL. + """ + def __init__(self, parent, max_send=100000000.0): + wx.Dialog.__init__(self, parent, title="Send XRP") + sizer = AutoGridBagSizer(self) + self.parent = parent + + # Icons to indicate a validation error + bmp_err = wx.ArtProvider.GetBitmap(wx.ART_ERROR, wx.ART_CMN_DIALOG, size=(16,16)) + self.err_to = wx.StaticBitmap(self, bitmap=bmp_err) + self.err_dtag = wx.StaticBitmap(self, bitmap=bmp_err) + self.err_amt = wx.StaticBitmap(self, bitmap=bmp_err) + self.err_to.Hide() + self.err_dtag.Hide() + self.err_amt.Hide() + + # Icons for domain verification + bmp_check = wx.ArtProvider.GetBitmap(wx.ART_TICK_MARK, wx.ART_CMN_DIALOG, size=(16,16)) + self.domain_text = wx.StaticText(self, label="") + self.domain_verified = wx.StaticBitmap(self, bitmap=bmp_check) + self.domain_verified.Hide() + + if max_send <= 0: + max_send = 100000000.0 + self.err_amt.Show() + self.err_amt.SetToolTip("Not enough XRP to pay the reserve and transaction cost!") + + lbl_to = wx.StaticText(self, label="To (Address):") + lbl_dtag = wx.StaticText(self, label="Destination Tag:") + lbl_amt = wx.StaticText(self, label="Amount of XRP:") + self.txt_to = wx.TextCtrl(self) + self.txt_dtag = wx.TextCtrl(self) + self.txt_amt = wx.SpinCtrlDouble(self, value="20.0", min=0.000001, max=max_send) + self.txt_amt.SetDigits(6) + self.txt_amt.SetIncrement(1.0) + + # The "Send" button is functionally an "OK" button except for the text. + self.btn_send = wx.Button(self, wx.ID_OK, label="Send") + btn_cancel = wx.Button(self, wx.ID_CANCEL) + + sizer.BulkAdd(((lbl_to, self.txt_to, self.err_to), + (self.domain_verified, self.domain_text), + (lbl_dtag, self.txt_dtag, self.err_dtag), + (lbl_amt, self.txt_amt, self.err_amt), + (btn_cancel, self.btn_send)) ) + sizer.Fit(self) + + self.txt_dtag.Bind(wx.EVT_TEXT, self.on_dest_tag_edit) + self.txt_to.Bind(wx.EVT_TEXT, self.on_to_edit) + + def get_payment_data(self): + """ + Construct a dictionary with the relevant payment details to pass to the + worker thread for making a payment. Called after the user clicks "Send". + """ + return { + "to": self.txt_to.GetValue().strip(), + "dtag": self.txt_dtag.GetValue().strip(), + "amt": self.txt_amt.GetValue(), + } + + def on_to_edit(self, event): + """ + When the user edits the "To" field, check that the address is well- + formatted. If it's an X-address, fill in the destination tag and disable + it. Also, start a background check to confirm more details about the + address. + """ + v = self.txt_to.GetValue().strip() + # Reset warnings / domain verification + err_msg = "" + self.err_to.SetToolTip("") + self.err_to.Hide() + self.domain_text.SetLabel("") + self.domain_verified.Hide() + + if xrpl.core.addresscodec.is_valid_xaddress(v): + cl_addr, tag, is_test = xrpl.core.addresscodec.xaddress_to_classic_address(v) + if tag is None: # Not the same as tag = 0 + tag = "" + self.txt_dtag.ChangeValue(str(tag)) + self.txt_dtag.Disable() + + if cl_addr == self.parent.classic_address: + err_msg = "Can't send XRP to self." + elif is_test != self.parent.test_network: + err_msg = "This address is intended for a different network." + + elif not self.txt_dtag.IsEditable(): + self.txt_dtag.Clear() + self.txt_dtag.Enable() + + if not (xrpl.core.addresscodec.is_valid_classic_address(v) or + xrpl.core.addresscodec.is_valid_xaddress(v) ): + self.btn_send.Disable() + err_msg = "Not a valid address." + elif v == self.parent.classic_address: + self.btn_send.Disable() + err_msg = "Can't send XRP to self." + else: + self.parent.run_bg_job(self.parent.worker.check_destination(v, self)) + + if err_msg: + self.err_to.SetToolTip(err_msg) + self.err_to.Show() + else: + self.err_to.Hide() + + def on_dest_tag_edit(self, event): + """ + When the user edits the Destination Tag field, strip non-numeric + characters from it. + """ + v = self.txt_dtag.GetValue().strip() + v = re.sub(r"[^0-9]", "", v) + self.txt_dtag.ChangeValue(v) # SetValue would generate another EVT_TEXT + self.txt_dtag.SetInsertionPointEnd() + + def update_dest_info(self, dest_status): + """ + Update the UI with details provided by a background job to check the + destination address. + """ + # Keep existing error message if there is one + try: + err_msg = self.err_to.GetToolTip().GetTip().strip() + except RuntimeError: + # This method can be called after the dialog it belongs to has been + # closed. In that case, there's nothing to do here. + return + + if not dest_status["funded"]: + err_msg = ("Warning: this account does not exist. The payment will " + "fail unless you send enough to fund it.") + elif dest_status["disallow_xrp"]: + err_msg = "This account does not want to receive XRP." + + # Domain verification + bmp_err = wx.ArtProvider.GetBitmap(wx.ART_ERROR, wx.ART_CMN_DIALOG, size=(16,16)) + bmp_check = wx.ArtProvider.GetBitmap(wx.ART_TICK_MARK, wx.ART_CMN_DIALOG, size=(16,16)) + domain = dest_status["domain_str"] + verified = dest_status["domain_verified"] + if not domain: + self.domain_text.Hide() + self.domain_verified.Hide() + elif verified: + self.domain_text.SetLabel(domain) + self.domain_text.Show() + self.domain_verified.SetToolTip("Domain verified") + self.domain_verified.SetBitmap(bmp_check) + self.domain_verified.Show() + else: + self.domain_text.SetLabel(domain) + self.domain_text.Show() + self.domain_verified.SetToolTip("Failed to verify domain") + self.domain_verified.SetBitmap(bmp_err) + self.domain_verified.Show() + + if err_msg: + # Disabling the button is optional. These types of errors can be + # benign, so you could let the user "click through" them. + self.btn_send.Disable() + self.err_to.SetToolTip(err_msg) + self.err_to.Show() + else: + self.btn_send.Enable() + self.err_to.SetToolTip("") + self.err_to.Hide() + + +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 + # This account's total XRP reserve including base + owner amounts + self.reserve_xrp = 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. + """ + self.tabs = wx.Notebook(self, style=wx.BK_DEFAULT) + # Tab 1: "Summary" pane ------------------------------------------------ + main_panel = wx.Panel(self.tabs) + self.tabs.AddPage(main_panel, "Summary") + + 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)) ) + + + # Send XRP button. Disabled until we have a secret key & network connection + self.sxb = wx.Button(main_panel, label="Send XRP") + self.sxb.SetToolTip("Disabled in read-only mode.") + self.sxb.Disable() + self.Bind(wx.EVT_BUTTON, self.click_send_xrp, source=self.sxb) + + 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.sxb, 0, flag=wx.ALL, border=5) + main_sizer.Add(self.ledger_info, 1, flag=wx.EXPAND|wx.ALL, border=5) + main_panel.SetSizer(main_sizer) + + # Tab 2: "Transaction History" pane ------------------------------------ + txhistory_panel = wx.Panel(self.tabs) + self.tabs.AddPage(txhistory_panel, "Transaction History") + txhistory_sizer = wx.BoxSizer(wx.VERTICAL) + + self.tx_list = wx.dataview.DataViewListCtrl(txhistory_panel) + self.tx_list.AppendTextColumn("Confirmed") + self.tx_list.AppendTextColumn("Type") + self.tx_list.AppendTextColumn("From") + self.tx_list.AppendTextColumn("To") + self.tx_list.AppendTextColumn("Value Delivered") + self.tx_list.AppendTextColumn("Identifying Hash") + self.tx_list.AppendTextColumn("Raw JSON") + txhistory_sizer.Add(self.tx_list, 1, wx.EXPAND|wx.ALL) + self.pending_tx_rows = {} # Map pending tx hashes to rows in the history UI + txhistory_panel.SetSizer(txhistory_sizer) + + # Tab 3: "Tokens" pane ------------------------------------------------- + tokens_panel = wx.Panel(self.tabs) + self.tabs.AddPage(tokens_panel, "Tokens") + tokens_sizer = wx.BoxSizer(wx.VERTICAL) + + self.tkn_list = wx.dataview.DataViewListCtrl(tokens_panel) + self.tkn_list.AppendTextColumn("Currency") + self.tkn_list.AppendTextColumn("Issuer") + self.tkn_list.AppendTextColumn("Balance") + self.tkn_list.AppendTextColumn("Limit") + self.tkn_list.AppendTextColumn("Peer Limit") + self.tkn_list.AppendToggleColumn("Allows Rippling?", mode=wx.dataview.DATAVIEW_CELL_INERT) + self.tkn_list.AppendToggleColumn("Frozen?", mode=wx.dataview.DATAVIEW_CELL_INERT) + self.tkn_list.AppendToggleColumn("Authorized?", mode=wx.dataview.DATAVIEW_CELL_INERT) + self.tkn_list.AppendToggleColumn("Peer Allows Rippling?", mode=wx.dataview.DATAVIEW_CELL_INERT) + self.tkn_list.AppendToggleColumn("Frozen by Peer?", mode=wx.dataview.DATAVIEW_CELL_INERT) + self.tkn_list.AppendToggleColumn("Authorized by Peer?", mode=wx.dataview.DATAVIEW_CELL_INERT) + tokens_sizer.Add(self.tkn_list, 1, wx.EXPAND|wx.ALL) + tokens_panel.SetSizer(tokens_sizer) + + # Tab 4: "Objects" pane ------------------------------------------------ + objs_panel = wx.Panel(self.tabs) + self.tabs.AddPage(objs_panel, "Other Objects") + objs_sizer = wx.BoxSizer(wx.VERTICAL) + self.o_list = wx.dataview.DataViewListCtrl(objs_panel) + self.o_list.AppendTextColumn("Type") + self.o_list.AppendTextColumn("Summary") + objs_sizer.Add(self.o_list, 1, wx.EXPAND|wx.ALL) + objs_panel.SetSizer(objs_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 and save for calculating max send. + reserve_xrp = self.calculate_reserve_xrp(acct.get("OwnerCount", 0)) + if reserve_xrp != None: + self.st_reserve.SetLabel(str(reserve_xrp)) + self.reserve_xrp = reserve_xrp + + def enable_readwrite(self): + """ + Enable buttons for sending transactions. + """ + self.sxb.Enable() + self.sxb.SetToolTip("") + + def displayable_amount(self, a): + """ + Convert an arbitrary amount value from the XRPL to a string to be + displayed to the user: + - Convert drops of XRP to 6-decimal XRP (e.g. '12.345000 XRP') + - For issued tokens, show amount, currency code, and issuer. For + example, 100 USD issued by address r12345... is returned as + '100 USD.r12345...' + + Leaves non-standard (hex) currency codes as-is. + """ + if a == "unavailable": + # Special case for pre-2014 partial payments. + return a + elif type(a) == str: + # It's an XRP amount in drops. Convert to decimal. + return f"{xrpl.utils.drops_to_xrp(a)} XRP" + else: + # It's a token amount. + return f"{a['value']} {a['currency']}.{a['issuer']}" + + def add_tx_row(self, t, prepend=False): + """ + Add one row to the account transaction history control. Helper function + called by other methods. + """ + conf_dt = xrpl.utils.ripple_time_to_datetime(t["tx"]["date"]) + # Convert datetime to locale-default representation & time zone + confirmation_time = conf_dt.astimezone().strftime("%c") + + tx_hash = t["tx"]["hash"] + tx_type = t["tx"]["TransactionType"] + from_acct = t["tx"].get("Account") or "" + if from_acct == self.classic_address: + from_acct = "(Me)" + to_acct = t["tx"].get("Destination") or "" + if to_acct == self.classic_address: + to_acct = "(Me)" + + delivered_amt = t["meta"].get("delivered_amount") + if delivered_amt: + delivered_amt = self.displayable_amount(delivered_amt) + else: + delivered_amt = "" + + cols = (confirmation_time, tx_type, from_acct, to_acct, delivered_amt, + tx_hash, str(t)) + if prepend: + self.tx_list.PrependItem(cols) + else: + self.tx_list.AppendItem(cols) + + def update_account_tx(self, data): + """ + Update the transaction history tab with information from an account_tx + response. + """ + txs = data["transactions"] + # Note: if you extend the code to do paginated responses, you might want + # to keep previous history instead of deleting the contents first. + self.tx_list.DeleteAllItems() + for t in txs: + self.add_tx_row(t) + + def add_tx_from_sub(self, t): + """ + Add 1 transaction to the history based on a subscription stream message. + Assumes only validated transaction streams (e.g. transactions, accounts) + not proposed transaction streams. + + Also, send a notification to the user about it. + """ + # Convert to same format as account_tx results + t["tx"] = t["transaction"] + if t["tx"]["hash"] in self.pending_tx_rows.keys(): + dvi = self.pending_tx_rows[t["tx"]["hash"]] + pending_row = self.tx_list.ItemToRow(dvi) + self.tx_list.DeleteItem(pending_row) + + self.add_tx_row(t, prepend=True) + # Scroll to top of list. + self.tx_list.EnsureVisible(self.tx_list.RowToItem(0)) + + # Send a notification message (aka a "toast") about the transaction. + # Note the transaction stream and account_tx include all transactions + # that "affect" the account, no just ones directly from/to the account. + # For example, if the account has issued tokens, it gets notified when + # other users transfer those tokens among themselves. + notif = wx.adv.NotificationMessage(title="New Transaction", message = + f"New {t['tx']['TransactionType']} transaction confirmed!") + notif.SetFlags(wx.ICON_INFORMATION) + notif.Show() + + def add_pending_tx(self, txm): + """ + Add a "pending" transaction to the history based on a transaction model + that was (presumably) just submitted. + """ + confirmation_time = "(pending)" + tx_type = txm.transaction_type + from_acct = txm.account + if from_acct == self.classic_address: + from_acct = "(Me)" + # Some transactions don't have a destination, so we need to handle that. + to_acct = getattr(txm, "destination", "") + if to_acct == self.classic_address: + to_acct = "(Me)" + # Delivered amount is only known after a transaction is processed, so + # leave this column empty in the display for pending transactions. + delivered_amt = "" + tx_hash = txm.get_hash() + cols = (confirmation_time, tx_type, from_acct, to_acct, delivered_amt, + tx_hash, str(txm.to_xrpl())) + self.tx_list.PrependItem(cols) + self.pending_tx_rows[tx_hash] = self.tx_list.RowToItem(0) + + def click_send_xrp(self, event): + """ + Pop up a dialog for the user to input how much XRP to send where, and + send the transaction (if the user doesn't cancel). + """ + xrp_bal = Decimal(self.st_xrp_balance.GetLabelText()) + tx_cost = Decimal("0.000010") + reserve = self.reserve_xrp or Decimal(0.000000) + dlg = SendXRPDialog(self, max_send=float(xrp_bal - reserve - tx_cost)) + dlg.CenterOnScreen() + resp = dlg.ShowModal() + if resp != wx.ID_OK: + print("Send XRP canceled") + dlg.Destroy() + return + + paydata = dlg.get_payment_data() + dlg.Destroy() + self.run_bg_job(self.worker.send_xrp(paydata)) + notif = wx.adv.NotificationMessage(title="Sending!", message = + f"Sending a payment for {paydata['amt']} XRP!") + notif.SetFlags(wx.ICON_INFORMATION) + notif.Show() + + def update_account_lines(self, lines): + """ + Update the Tokens tab based on an account_lines result. + This doesn't handle pagination. + """ + self.tkn_list.DeleteAllItems() + for l in lines: + self.tkn_list.AppendItem([ + l["currency"], + l["account"], + l["balance"], + l["limit"], + l["limit_peer"], + not l.get("no_ripple", False), + l.get("freeze", False), + l.get("authorized", False), + l.get("freeze_peer", False), + not l.get("no_ripple_peer", False), + l.get("peer_authorized", False), + ]) + + def update_account_objects(self, objs): + """ + Update the tab of objects owned with the results of an account_objects + call, skipping RippleState objects since those are represented in the + Tokens tab. This doesn't handle pagination. + """ + self.o_list.DeleteAllItems() + for o in objs: + if o["LedgerEntryType"] == "RippleState": + continue + elif o["LedgerEntryType"] == "Check": + check_amt = self.displayable_amount(o["SendMax"]) + summary = f"Deliver up to {check_amt}" + if o["Account"] == self.classic_address: + # Outgoing check + summary += f" to {o['Destination']}" + else: + summary += f" from {o['Account']}" + elif o["LedgerEntryType"] == "DepositPreauth": + if o["Account"] == self.classic_address: + # We authorized them + summary = f"Authorized {o['Authorize']}" + else: + summary = f"Authorized by {o['Account']}" + elif o["LedgerEntryType"] == "Escrow": + escrow_amt = self.displayable_amount(o["Amount"]) + summary = f"Hold {escrow_amt} " + if o["Account"] == self.classic_address: + # Outgoing escrow + summary += f"for {o['Destination']} " + else: + summary += f"from {o['Account']} " + if o.get("Condition"): + summary += "with condition " + if o.get("FinishAfter"): + fa_dt = xrpl.utils.ripple_time_to_datetime(o["FinishAfter"]) + # Convert datetime to locale-default representation & time zone + fa_time = fa_dt.astimezone().strftime("%c") + summary += f"until {fa_time} " + if o.get("CancelAfter"): + ca_dt = xrpl.utils.ripple_time_to_datetime(o["CancelAfter"]) + # Convert datetime to locale-default representation & time zone + ca_time = ca_dt.astimezone().strftime("%c") + summary += f"or cancel at {ca_time} " + elif o["LedgerEntryType"] == "Offer": + # An order we placed in the decentralized exchange + sell_amt = self.displayable_amount(o["TakerGets"]) + buy_amt = self.displayable_amount(o["TakerPays"]) + summary = f"Trade {sell_amt} to receive {buy_amt}" + elif o["LedgerEntryType"] == "PayChannel": + # Payment channels' balance is determined by the amount paid + # out of the amount funded + amt_dec = xrpl.utils.drops_to_xrp(o["Amount"]) + bal_dec = xrpl.utils.drops_to_xrp(o["Balance"]) + summary = f"{bal_dec} paid of {amt_dec} XRP" + if o["Account"] == self.classic_address: + # Outgoing channel + summary += f" to {o['Destination']}" + else: + summary += f" from {o['Account']}" + elif o["LedgerEntryType"] == "SignerList": + summary = f"Quorum: {o['SignerQuorum']}. Signers: " + summary += ", ".join([ + f"{se['SignerEntry']['Account']} " + f"(Weight: {se['SignerEntry']['SignerWeight']})" + for se in o["SignerEntries"]]) + elif o["LedgerEntryType"] == "Ticket": + summary = f"Ticket #{o['TicketSequence']}" + else: + summary = "" + cols = (o["LedgerEntryType"], summary) + self.o_list.AppendItem(cols) + +if __name__ == "__main__": + WS_URL = "wss://s.altnet.rippletest.net:51233" # Testnet + #WS_URL = "wss://xrplcluster.com" # Mainnet + app = wx.App() + frame = TWaXLFrame(WS_URL, test_network=True) + frame.Show() + app.MainLoop() diff --git a/content/_code-samples/build-a-wallet/py/x_owned_objects.py b/content/_code-samples/build-a-wallet/py/x_owned_objects.py deleted file mode 100644 index a2b79a02cf..0000000000 --- a/content/_code-samples/build-a-wallet/py/x_owned_objects.py +++ /dev/null @@ -1,304 +0,0 @@ -# "Build a Wallet" tutorial, extra: List various objects attached to an account - -import xrpl -import wx -import wx.lib.newevent -import wx.dataview -from threading import Thread -from decimal import Decimal - -class WSResponseError(Exception): - pass - -class SmartWSClient(xrpl.clients.WebsocketClient): - def __init__(self, *args, **kwargs): - self._handlers = {} - self._pending_requests = {} - self._id = 0 - super().__init__(*args, **kwargs) - - def on(self, event_type, callback): - """ - Map a callback function to a type of event message from the connected - server. Only supports one callback function per event type. - """ - self._handlers[event_type] = callback - - def request(self, req_dict, callback): - if "id" not in req_dict: - req_dict["id"] = f"__auto_{self._id}" - self._id += 1 - # Work around xrpl-py quirk where it won't let you instantiate a request - # in proper WebSocket format because WS uses "command" instead of - # "method" but xrpl-py checks for "method": - req_dict["method"] = req_dict["command"] - del req_dict["command"] - - req = xrpl.models.requests.request.Request.from_xrpl(req_dict) - req.validate() - self._pending_requests[req.id] = callback - self.send(req) - - def handle_messages(self): - for message in self: - if message.get("type") == "response": - if message.get("status") == "success": - del message["status"] - else: - raise WSResponseError("Unsuccessful response:", message) - - msg_id = message.get("id") - if msg_id in self._pending_requests: - self._pending_requests[msg_id](message) - del self._pending_requests[msg_id] - else: - raise WSResponseError("Response to unknown request:", message) - - elif message.get("type") in self._handlers: - self._handlers[message.get("type")](message) - - -# Set up event types to pass info from the worker thread to the main UI thread -GotNewLedger, EVT_NEW_LEDGER = wx.lib.newevent.NewEvent() -GotAccountInfo, EVT_ACCT_INFO = wx.lib.newevent.NewEvent() -GotAccountObjects, EVT_ACCT_OBJ = wx.lib.newevent.NewEvent() - -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, ws_url, notify_window, classic_address): - Thread.__init__(self, daemon=True) - self.notify_window = notify_window - self.ws_url = ws_url - self.account = classic_address - - def notify_ledger(self, message): - wx.QueueEvent(self.notify_window, GotNewLedger(data=message)) - - def notify_account(self, message): - wx.QueueEvent(self.notify_window, GotAccountInfo(data=message["result"])) - - def notify_objects(self, message): - wx.QueueEvent(self.notify_window, GotAccountObjects(data=message["result"])) - - def on_transaction(self, client, message): - """ - Re-check our balance and owned objects whenever a new transaction - touches our account. - """ - client.request({ - "command": "account_info", - "account": self.account, - "ledger_index": message["ledger_index"] - }, self.notify_account) - client.request({ - "command": "account_objects", - "account": self.account, - "ledger_index": message["ledger_index"] - }, self.notify_objects) - - def run(self): - with SmartWSClient(self.ws_url) as client: - # Subscribe to ledger updates - client.request({ - "command": "subscribe", - "streams": ["ledger"], - "accounts": [self.account] - }, - lambda message: self.notify_ledger(message["result"]) - ) - client.on("ledgerClosed", self.notify_ledger) - client.on("transaction", lambda message: self.on_transaction(client, message)) - - # Look up our balance right away - client.request({ - "command": "account_info", - "account": self.account, - "ledger_index": "validated" - }, - self.notify_account - ) - # Look up our owned account objects - client.request({ - "command": "account_objects", - "account": self.account, - "ledger_index": "validated" - }, - self.notify_objects - ) - # Start looping through messages received. This runs indefinitely. - client.handle_messages() - -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 - - self.tabs = wx.Notebook(self, style=wx.BK_DEFAULT) - - # Tab 1: "Summary" pane ------------------------------------------------ - main_panel = wx.Panel(self.tabs) - self.tabs.AddPage(main_panel, "Summary") - main_sizer = wx.BoxSizer(wx.VERTICAL) - - self.acct_info_area = wx.StaticBox(main_panel, label="Account Info") - aia_sizer = wx.GridBagSizer(vgap=5, hgap=5) - self.acct_info_area.SetSizer(aia_sizer) - aia_sizer.Add(wx.StaticText(self.acct_info_area, label="Classic Address:"), (0,0)) - self.st_classic_address = wx.StaticText(self.acct_info_area, label="TBD") - aia_sizer.Add(self.st_classic_address, (0,1)) - aia_sizer.Add(wx.StaticText(self.acct_info_area, label="X-Address:"), (1,0)) - self.st_x_address = wx.StaticText(self.acct_info_area, label="TBD") - aia_sizer.Add(self.st_x_address, (1,1), flag=wx.EXPAND) - aia_sizer.Add(wx.StaticText(self.acct_info_area, label="XRP Balance:"), (2,0)) - self.st_xrp_balance = wx.StaticText(self.acct_info_area, label="TBD") - aia_sizer.Add(self.st_xrp_balance, (2,1), flag=wx.EXPAND) - - main_sizer.Add(self.acct_info_area, 1, wx.EXPAND|wx.ALL, 25) - - self.ledger_info = wx.StaticText(main_panel, label="Not connected") - main_sizer.Add(self.ledger_info, 1, wx.EXPAND|wx.ALL, 25) - - main_panel.SetSizer(main_sizer) - - # Tab 2: "Objects" pane ------------------------------------------------ - objs_panel = wx.Panel(self.tabs) - self.tabs.AddPage(objs_panel, "Objects") - objs_sizer = wx.BoxSizer(wx.VERTICAL) - - self.o_list = wx.dataview.DataViewListCtrl(objs_panel) - self.o_list.AppendTextColumn("Type") - self.o_list.AppendTextColumn("From") - self.o_list.AppendTextColumn("Balance") - self.o_list.AppendTextColumn("JSON") - objs_sizer.Add(self.o_list, 1, wx.EXPAND|wx.ALL) - - objs_panel.SetSizer(objs_sizer) - - - # Pop up to ask user for their account --------------------------------- - 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") - - if account_dialog.ShowModal() == wx.ID_OK: - self.set_up_account(account_dialog.GetValue()) - account_dialog.Destroy() - else: - # If the user presses Cancel on the account entry, exit the app. - exit(1) - - # Attach handlers and start bg thread for updates from the ledger ------ - self.Bind(EVT_NEW_LEDGER, self.update_ledger) - self.Bind(EVT_ACCT_INFO, self.update_account) - self.Bind(EVT_ACCT_OBJ, self.update_account_objects) - XRPLMonitorThread(url, self, self.classic_address).start() - - def set_up_account(self, value): - value = value.strip() - - if xrpl.core.addresscodec.is_valid_xaddress(value): - classic_address, dest_tag, test_network = xrpl.core.addresscodec.xaddress_to_classic_address(value) - if test_network != self.test_network: - print(f"X-address {value} is meant for a different network type" - f"than this client is connected to." - f"(Client is on: {'a test network' if self.test_network else 'Mainnet'})") - exit(1) - self.xaddress = value - self.classic_address = classic_address - self.wallet = None - - elif xrpl.core.addresscodec.is_valid_classic_address(value): - self.xaddress = xrpl.core.addresscodec.classic_address_to_xaddress( - value, tag=None, is_test_network=self.test_network) - self.classic_address = value - self.wallet = None - - else: - try: - # Check if it's a valid seed - seed_bytes, alg = xrpl.core.addresscodec.decode_seed(value) - self.wallet = xrpl.wallet.Wallet(seed=value, sequence=0) - # We'll fill in the actual sequence later. - self.xaddress = self.wallet.get_xaddress(is_test=self.test_network) - self.classic_address = self.wallet.classic_address - except Exception as e: - print(e) - exit(1) - self.st_classic_address.SetLabel(self.classic_address) - self.st_x_address.SetLabel(self.xaddress) - - def update_ledger(self, event): - message = event.data - 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}") - - def update_account(self, event): - acct = event.data["account_data"] - xrp_balance = str(xrpl.utils.drops_to_xrp(acct["Balance"])) - self.st_xrp_balance.SetLabel(xrp_balance) - - def balance_with_currency(self, bal): - """ - Convert an XRPL Balance value, which might be a string (drops of XRP) or - an object (issued token with value/issuer/currency code) into a tuple: - (balance, currency code) where balance is a number. Converts drops to - decimal XRP. - """ - if type(bal) == str: - return xrpl.utils.drops_to_xrp(bal), "XRP" - - return Decimal(bal["value"]), bal["currency"] - - def update_account_objects(self, event): - objs = event.data["account_objects"] - self.o_list.DeleteAllItems() - for o in objs: - from_ = o.get("Account") or "" - if o.get("Amount"): - bal_dec, currency = self.balance_with_currency(o.get("Amount")) - balance = f"{bal_dec} {currency}" - else: - balance = "" - if o["LedgerEntryType"] == "RippleState": - # We're either the "high" or "low" node for a RippleState - bal_dec, currency = self.balance_with_currency(o.get("Balance")) - if o["HighLimit"]["issuer"] == self.classic_address: - from_ = o["LowLimit"]["issuer"] - # Balance is recorded from the perspective of the low - # account, so we need the negative of it here. - balance = f"{-bal_dec} {currency}" - else: - from_ = o["HighLimit"]["issuer"] - balance = f"{bal_dec} {currency}" - elif o["LedgerEntryType"] == "PayChannel": - # Payment channels' balance is determined by the amount paid - # out of the amount funded - amt_dec = xrpl.utils.drops_to_xrp(o["Amount"]) - bal_dec = xrpl.utils.drops_to_xrp(o["Balance"]) - balance = f"{bal_dec} paid of {amt_dec} XRP" - cols = (o["LedgerEntryType"], from_, balance, str(o)) - self.o_list.AppendItem(cols) - -if __name__ == "__main__": - #JSON_RPC_URL = "https://s.altnet.rippletest.net:51234/" - #JSON_RPC_URL = "http://localhost:5005/" - WS_URL = "wss://s.altnet.rippletest.net:51233" - - app = wx.App() - frame = TWaXLFrame(WS_URL) - frame.Show() - app.MainLoop() diff --git a/content/tutorials/build-apps/build-a-desktop-wallet-in-python.md b/content/tutorials/build-apps/build-a-desktop-wallet-in-python.md index bfa48a63a6..58745863ac 100644 --- a/content/tutorials/build-apps/build-a-desktop-wallet-in-python.md +++ b/content/tutorials/build-apps/build-a-desktop-wallet-in-python.md @@ -544,6 +544,7 @@ To test X-addresses, try the following addresses: Now that you have a functional wallet, you can take it in several new directions. The following are a few ideas: - You could support more of the XRP Ledger's [transaction types](transaction-types.html) including [tokens](issued-currencies.html) and [cross-currency payments](cross-currency-payments.html) + - Example code for displaying token balances and other objects: [`7_owned_objects.py`]({{target.github_forkurl}}/tree/{{target.github_branch}}/content/_code-samples/build-a-wallet/py/7_owned_objects.py) - Allow the user to trade in the [decentralized exchange](decentralized-exchange.html) - Add a way to request payments, such as with QR codes or URIs that open in your wallet. - Support better account security including [regular key pairs](cryptographic-keys.html#regular-key-pair) or [multi-signing](multi-signing.html).