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

Build wallet: Backport refactor from step 6

Browse Source
This commit is contained in:
mDuo13
2022-01-13 18:36:52 -08:00
parent e8a5847cc5
commit 773cabbe5c
6 changed files with 848 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
content/_code-samples/build-a-wallet/py/1_hello.py
View File

@@ -38,7 +38,6 @@ class TWaXLFrame(wx.Frame):
if __name__ == "__main__":
JSON_RPC_URL = "https://s.altnet.rippletest.net:51234/"
#JSON_RPC_URL = "http://localhost:5005/"
app = wx.App()
frame = TWaXLFrame(JSON_RPC_URL)

118
content/_code-samples/build-a-wallet/py/2_threaded.py
View File

@@ -1,74 +1,112 @@
# "Build a Wallet" tutorial, step 2: Watch ledger closes from a worker thread.
import xrpl
import wx
# New dependencies
from threading import Thread
import wx.lib.newevent
# Set up an event type to pass info from the worker thread to the main thread
GotNewLedger, EVT_NEW_LEDGER = wx.lib.newevent.NewEvent()
import asyncio
import re
import wx
from threading import Thread
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, gui):
def __init__(self, url, gui, loop):
Thread.__init__(self, daemon=True)
self.gui = gui
self.ws_url = ws_url
self.client = xrpl.clients.WebsocketClient(self.ws_url)
self.url = url
self.loop = loop
def run(self):
self.client.open()
# Subscribe to ledger updates
#TODO: use request/response and on for this
self.client.send(xrpl.models.requests.Subscribe(
id="ledger_sub",
streams=[xrpl.models.requests.StreamParameter.LEDGER]
"""
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.
"""
asyncio.set_event_loop(self.loop)
self.loop.run_forever()
async def watch_xrpl(self):
"""
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.
"""
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)
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"]
))
# Watch for messages in the client
for message in self.client:
if message.get("id") == "ledger_sub":
# Immediate response to our subscribe command.
wx.QueueEvent(self.gui, GotNewLedger(data=message["result"]))
elif message.get("type") == "ledgerClosed":
# Ongoing notifications that new ledgers have been validated.
wx.QueueEvent(self.gui, GotNewLedger(data=message))
else:
print("Unhandled message:", message)
# 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)
class TWaXLFrame(wx.Frame):
"""
Tutorial Wallet for the XRP Ledger (TWaXL)
user interface, main frame.
"""
def __init__(self, url):
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.build_ui()
# Start background thread for updates from the ledger ------------------
self.worker_loop = asyncio.new_event_loop()
self.worker = XRPLMonitorThread(url, self, self.worker_loop)
self.worker.start()
self.run_bg_job(self.worker.watch_xrpl())
def build_ui(self):
main_panel = wx.Panel(self)
self.ledger_info = wx.StaticText(main_panel, label="Not connected")
main_sizer = wx.BoxSizer(wx.VERTICAL)
main_sizer.Add(self.ledger_info, 1, flag=wx.EXPAND|wx.ALL, border=5)
main_panel.SetSizer(main_sizer)
self.st = wx.StaticText(main_panel, label="Not connected")
main_sizer.Add(self.st, wx.SizerFlags().Border(wx.TOP|wx.LEFT, 25))
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)
self.Bind(EVT_NEW_LEDGER, self.update_ledger)
XRPLMonitorThread(url, self).start()
def update_ledger(self, event):
message = event.data
self.st.SetLabel(f"Latest validated ledger:\n"
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']}")
f"Ledger Hash: {message['ledger_hash']}\n"
f"Close time: {close_time_iso}")
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"
WS_URL = "wss://s.altnet.rippletest.net:51233" # Testnet
app = wx.App()
frame = TWaXLFrame(WS_URL)
frame = TWaXLFrame(WS_URL, test_network=True)
frame.Show()
app.MainLoop()

288
content/_code-samples/build-a-wallet/py/3_account.py
View File

@@ -1,60 +1,107 @@
# "Build a Wallet" tutorial, step 3: Take account input & show account info
import xrpl
import asyncio
import re
import wx
from threading import Thread
import wx.lib.newevent
from decimal import Decimal
# 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()
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):
def __init__(self, url, gui, loop):
Thread.__init__(self, daemon=True)
self.notify_window = notify_window
self.ws_url = ws_url
self.account = classic_address
self.gui = gui
self.url = url
self.loop = loop
def run(self):
with xrpl.clients.WebsocketClient(self.ws_url) as client:
# Subscribe to ledger updates
client.send(xrpl.models.requests.Subscribe(
id="ledger_sub",
streams=[xrpl.models.requests.StreamParameter.LEDGER],
accounts=[self.account]
))
client.send(xrpl.models.requests.AccountInfo(
id="acct_info",
account=self.account,
ledger_index="validated"
))
# Watch for messages in the client
i = 0 # nonce so we don't reuse request IDs
for message in client:
#print(message)
if message.get("id") == "ledger_sub":
# Immediate response to our subscribe command.)
wx.QueueEvent(self.notify_window, GotNewLedger(data=message["result"]))
elif message.get("id") and "acct_info" in message.get("id"):
wx.QueueEvent(self.notify_window, GotAccountInfo(data=message["result"]))
elif message.get("type") == "ledgerClosed":
# Ongoing notifications that new ledgers have been validated.
wx.QueueEvent(self.notify_window, GotNewLedger(data=message))
elif message.get("type") == "transaction":
# Got a new transaction. Check for updated balances
i+=1
client.send(xrpl.models.requests.AccountInfo(
id=f"acct_info{i}",
"""
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.
"""
asyncio.set_event_loop(self.loop)
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"]
))
else:
print("Unhandled message:", message)
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):
"""
@@ -65,101 +112,178 @@ class TWaXLFrame(wx.Frame):
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_loop = asyncio.new_event_loop()
self.worker = XRPLMonitorThread(url, self, self.worker_loop)
self.worker.start()
self.run_bg_job(self.worker.watch_xrpl_account(address, wallet))
def build_ui(self):
main_panel = wx.Panel(self)
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))
lbl_address = wx.StaticText(self.acct_info_area, label="Classic Address:")
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))
lbl_xaddress = wx.StaticText(self.acct_info_area, label="X-Address:")
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))
lbl_xrp_bal = wx.StaticText(self.acct_info_area, label="XRP Balance:")
self.st_xrp_balance = wx.StaticText(self.acct_info_area, label="TBD")
aia_sizer.Add(self.st_xrp_balance, (2,1), flag=wx.EXPAND)
lbl_reserve = wx.StaticText(self.acct_info_area, label="XRP Reserved:")
self.st_reserve = wx.StaticText(self.acct_info_area, label="TBD")
main_sizer.Add(self.acct_info_area, 1, wx.EXPAND|wx.ALL, 25)
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)) )
# Ledger info text. One multi-line static text, unlike the account area.
self.ledger_info = wx.StaticText(main_panel, label="Not connected")
main_sizer.Add(self.ledger_info, 1, wx.EXPAND|wx.ALL, 25)
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:
self.set_up_account(account_dialog.GetValue())
account_dialog.Destroy()
else:
if account_dialog.ShowModal() != wx.ID_OK:
# If the user presses Cancel on the account entry, exit the app.
exit(1)
self.Bind(EVT_NEW_LEDGER, self.update_ledger)
self.Bind(EVT_ACCT_INFO, self.update_account)
self.worker = XRPLMonitorThread(url, self, self.classic_address)
self.worker.start()
value = account_dialog.GetValue().strip()
account_dialog.Destroy()
def set_up_account(self, value):
value = value.strip()
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: {'a test network' if self.test_network else 'Mainnet'})")
f"(Client is on: {on_net})")
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(
classic_address = value
x_address = 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
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)
self.st_classic_address.SetLabel(self.classic_address)
self.st_x_address.SetLabel(self.xaddress)
def update_ledger(self, event):
message = event.data
# 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 (in drops of XRP) so we can calc account reserve
self.reserve_base = Decimal(message["reserve_base"])
self.reserve_inc = Decimal(message["reserve_inc"])
def update_account(self, event):
acct = event.data["account_data"]
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_drops = self.reserve_base + (self.reserve_inc * oc_decimal)
reserve_xrp = xrpl.utils.drops_to_xrp(str(reserve_drops))
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)
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"
# 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)
frame = TWaXLFrame(WS_URL, test_network=True)
frame.Show()
app.MainLoop()

399
content/_code-samples/build-a-wallet/py/4_tx_history.py
View File

@@ -1,135 +1,117 @@
# "Build a Wallet" tutorial, step 4: Show transaction history
import xrpl
import asyncio
import re
import wx
import wx.lib.newevent
import wx.dataview
import wx.adv
from threading import Thread
from decimal import Decimal
class WSResponseError(Exception):
pass
WSC_TIMEOUT = 0.2
class SmartWSClient(xrpl.clients.WebsocketClient):
def __init__(self, *args, **kwargs):
self._handlers = {}
self._pending_requests = {}
self._id = 0
super().__init__(*args, **kwargs, timeout=WSC_TIMEOUT)
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 run_forever(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()
GotAccountTx, EVT_ACCT_TX = wx.lib.newevent.NewEvent()
GotTxSub, EVT_TX_SUB = 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):
def __init__(self, url, gui, loop):
Thread.__init__(self, daemon=True)
self.notify_window = notify_window
self.ws_url = ws_url
self.account = classic_address
self.client = SmartWSClient(self.ws_url)
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_account_tx(self, message):
wx.QueueEvent(self.notify_window, GotAccountTx(data=message["result"]))
def on_transaction(self, message):
"""
Update our account history and re-check our balance whenever a new
transaction touches our account.
"""
self.client.request({
"command": "account_info",
"account": self.account,
"ledger_index": message["ledger_index"]
}, self.notify_account)
wx.QueueEvent(self.notify_window, GotTxSub(data=message))
self.gui = gui
self.url = url
self.loop = loop
def run(self):
self.client.open()
# Subscribe to ledger updates
self.client.request({
"command": "subscribe",
"streams": ["ledger"],
"accounts": [self.account]
},
lambda message: self.notify_ledger(message["result"])
)
self.client.on("ledgerClosed", self.notify_ledger)
self.client.on("transaction", self.on_transaction)
"""
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.
"""
asyncio.set_event_loop(self.loop)
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"])
# 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)
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)
# Look up our balance right away
self.client.request({
"command": "account_info",
"account": self.account,
"ledger_index": "validated"
},
self.notify_account
)
# Look up our transaction history
self.client.request({
"command": "account_tx",
"account": self.account
},
self.notify_account_tx
)
# Start looping through messages received. This runs indefinitely.
self.client.run_forever()
class TWaXLFrame(wx.Frame):
"""
@@ -140,32 +122,51 @@ class TWaXLFrame(wx.Frame):
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_loop = asyncio.new_event_loop()
self.worker = XRPLMonitorThread(url, self, self.worker_loop)
self.worker.start()
self.run_bg_job(self.worker.watch_xrpl_account(address, wallet))
def build_ui(self):
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))
lbl_address = wx.StaticText(self.acct_info_area, label="Classic Address:")
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))
lbl_xaddress = wx.StaticText(self.acct_info_area, label="X-Address:")
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))
lbl_xrp_bal = wx.StaticText(self.acct_info_area, label="XRP Balance:")
self.st_xrp_balance = wx.StaticText(self.acct_info_area, label="TBD")
aia_sizer.Add(self.st_xrp_balance, (2,1), flag=wx.EXPAND)
lbl_reserve = wx.StaticText(self.acct_info_area, label="XRP Reserved:")
self.st_reserve = wx.StaticText(self.acct_info_area, label="TBD")
main_sizer.Add(self.acct_info_area, 1, wx.EXPAND|wx.ALL, 25)
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)) )
# Ledger info text. One multi-line static text, unlike the account area.
self.ledger_info = wx.StaticText(main_panel, label="Not connected")
main_sizer.Add(self.ledger_info, 1, wx.EXPAND|wx.ALL, 25)
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)
# Tab 2: "Transaction History" pane ------------------------------------
@@ -185,81 +186,135 @@ class TWaXLFrame(wx.Frame):
objs_panel.SetSizer(objs_sizer)
# Pop up to ask user for their account ---------------------------------
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:
self.set_up_account(account_dialog.GetValue())
account_dialog.Destroy()
else:
if account_dialog.ShowModal() != wx.ID_OK:
# 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_TX, self.update_account_tx)
self.Bind(EVT_TX_SUB, self.add_tx_from_sub)
self.worker = XRPLMonitorThread(url, self, self.classic_address)
self.worker.start()
value = account_dialog.GetValue().strip()
account_dialog.Destroy()
def set_up_account(self, value):
value = value.strip()
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: {'a test network' if self.test_network else 'Mainnet'})")
f"(Client is on: {on_net})")
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(
classic_address = value
x_address = 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
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)
self.st_classic_address.SetLabel(self.classic_address)
self.st_x_address.SetLabel(self.xaddress)
def update_ledger(self, event):
message = event.data
# 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 (in drops of XRP) so we can calc account reserve
self.reserve_base = Decimal(message["reserve_base"])
self.reserve_inc = Decimal(message["reserve_inc"])
def update_account(self, event):
acct = event.data["account_data"]
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_drops = self.reserve_base + (self.reserve_inc * oc_decimal)
reserve_xrp = xrpl.utils.drops_to_xrp(str(reserve_drops))
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))
@staticmethod
def displayable_amount(a):
"""
Convert an arbitrary amount value from the XRPL to a string to be
displayed to the user:
- Convert drops of XRP to 6-digit decimal XRP (e.g. '12.345000 XRP')
- 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...'
@@ -278,7 +333,8 @@ class TWaXLFrame(wx.Frame):
def add_tx_row(self, t, prepend=False):
"""
Add one row to the account transaction history control.
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
@@ -306,48 +362,47 @@ class TWaXLFrame(wx.Frame):
else:
self.tx_list.AppendItem(cols)
def update_account_tx(self, event):
def update_account_tx(self, data):
"""
Update the transaction history tab with information from an account_tx
response.
"""
txs = event.data["transactions"]
# TODO: with pagination, we should leave existing items
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, event):
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.
Also, send a notification to the user about it.
"""
t = event.data
# Convert to same format as account_tx results
t["tx"] = t["transaction"]
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 ("toast") about the transaction
# 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, an issuer gets notified when users transfer its tokens
# among themselves.
# 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()
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"
if __name__ == "__main__":
WS_URL = "wss://s.altnet.rippletest.net:51233" # Testnet
app = wx.App()
frame = TWaXLFrame(WS_URL)
frame = TWaXLFrame(WS_URL, test_network=True)
frame.Show()
app.MainLoop()

594
content/_code-samples/build-a-wallet/py/5_send_xrp.py
View File

@@ -1,80 +1,15 @@
# "Build a Wallet" tutorial, step 5: Send XRP button.
# This step finally introduces events from the GUI to the worker thread.
import re
import xrpl
import asyncio
import re
import wx
import wx.lib.newevent
import wx.dataview
import wx.adv
from threading import Thread
from decimal import Decimal
from queue import Queue, Empty
class WSResponseError(Exception):
pass
WSC_TIMEOUT = 0.2
class SmartWSClient(xrpl.clients.WebsocketClient):
def __init__(self, *args, **kwargs):
self._handlers = {}
self._pending_requests = {}
self._id = 0
self.jobq = Queue() # for incoming UI events
super().__init__(*args, **kwargs, timeout=WSC_TIMEOUT)
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 run_forever(self):
while True:
try:
req, callback = self.jobq.get(block=False)
self.request(req, callback)
except Empty:
pass
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()
GotAccountTx, EVT_ACCT_TX = wx.lib.newevent.NewEvent()
GotTxSub, EVT_TX_SUB = wx.lib.newevent.NewEvent()
class XRPLMonitorThread(Thread):
"""
@@ -82,116 +17,200 @@ class XRPLMonitorThread(Thread):
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):
def __init__(self, url, gui, loop):
Thread.__init__(self, daemon=True)
self.notify_window = notify_window
self.ws_url = ws_url
self.account = classic_address
self.client = SmartWSClient(self.ws_url)
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_account_tx(self, message):
wx.QueueEvent(self.notify_window, GotAccountTx(data=message["result"]))
def on_transaction(self, message):
"""
Update our account history and re-check our balance whenever a new
transaction touches our account.
"""
self.client.request({
"command": "account_info",
"account": self.account,
"ledger_index": message["ledger_index"]
}, self.notify_account)
wx.QueueEvent(self.notify_window, GotTxSub(data=message))
self.gui = gui
self.url = url
self.loop = loop
def run(self):
self.client.open()
# Subscribe to ledger updates
self.client.request({
"command": "subscribe",
"streams": ["ledger"],
"accounts": [self.account]
},
lambda message: self.notify_ledger(message["result"])
)
self.client.on("ledgerClosed", self.notify_ledger)
self.client.on("transaction", self.on_transaction)
"""
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.
"""
asyncio.set_event_loop(self.loop)
self.loop.run_forever()
# Look up our balance right away
self.client.request({
"command": "account_info",
"account": self.account,
"ledger_index": "validated"
},
self.notify_account
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)
async def send_xrp(self, paydata):
"""
Prepare, sign, and send an XRP payment with the provided parameters.
"""
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
)
# Look up our transaction history
self.client.request({
"command": "account_tx",
"account": self.account
},
self.notify_account_tx
)
# Start looping through messages received. This runs indefinitely.
self.client.run_forever()
# 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):
def __init__(self, parent, max_send=100000000.0):
"""
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):
wx.Dialog.__init__(self, parent, title="Send XRP")
sizer = wx.GridBagSizer(vgap=5, hgap=5)
self.SetSizer(sizer)
sizer = AutoGridBagSizer(self)
self.parent = parent
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 = wx.SpinCtrlDouble(self, value="20.0", min=0.000001)
self.txt_amt.SetDigits(6)
self.txt_amt.SetIncrement(1.0)
btn_send = wx.Button(self, wx.ID_OK, label="Send")
# 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)
# Lay out the controls in a 2x3 grid
ctrls = ((lbl_to, self.txt_to),
(lbl_dtag, self.txt_dtag),
(lbl_amt, self.txt_amt),
(btn_cancel, btn_send))
for x, row in enumerate(ctrls):
for y, ctrl in enumerate(row):
sizer.Add(ctrl, (x,y), flag=wx.EXPAND|wx.ALL, border=5)
sizer.BulkAdd(((lbl_to, self.txt_to),
(lbl_dtag, self.txt_dtag),
(lbl_amt, self.txt_amt),
(btn_cancel, self.btn_send)) )
sizer.Fit(self)
self.txt_dtag.Bind(wx.EVT_TEXT, self.onDestTagEdit)
self.txt_to.Bind(wx.EVT_TEXT, self.onToEdit)
self.txt_dtag.Bind(wx.EVT_TEXT, self.on_dest_tag_edit)
self.txt_to.Bind(wx.EVT_TEXT, self.on_to_edit)
def onToEdit(self, event):
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()
if xrpl.core.addresscodec.is_valid_xaddress(v):
cl_addr, tag, is_test = xrpl.core.addresscodec.xaddress_to_classic_address(v)
self.txt_dtag.ChangeValue(str(tag))
# self.txt_dtag.SetEditable(False)
self.txt_dtag.Disable()
elif not self.txt_dtag.IsEditable():
# Maybe the user erased an X-address from here
self.txt_dtag.Clear()
# self.txt_dtag.SetEditable(True)
self.txt_dtag.Enable()
def onDestTagEdit(self, event):
if not (xrpl.core.addresscodec.is_valid_classic_address(v) or
xrpl.core.addresscodec.is_valid_xaddress(v) ):
self.btn_send.Disable()
elif v == self.parent.classic_address:
self.btn_send.Disable()
else:
self.btn_send.Enable()
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 GetPaymentData(self):
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(),
@@ -199,8 +218,6 @@ class SendXRPDialog(wx.Dialog):
}
class TWaXLFrame(wx.Frame):
"""
Tutorial Wallet for the XRP Ledger (TWaXL)
@@ -210,37 +227,60 @@ class TWaXLFrame(wx.Frame):
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_loop = asyncio.new_event_loop()
self.worker = XRPLMonitorThread(url, self, self.worker_loop)
self.worker.start()
self.run_bg_job(self.worker.watch_xrpl_account(address, wallet))
def build_ui(self):
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))
lbl_address = wx.StaticText(self.acct_info_area, label="Classic Address:")
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))
lbl_xaddress = wx.StaticText(self.acct_info_area, label="X-Address:")
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))
lbl_xrp_bal = wx.StaticText(self.acct_info_area, label="XRP Balance:")
self.st_xrp_balance = wx.StaticText(self.acct_info_area, label="TBD")
aia_sizer.Add(self.st_xrp_balance, (2,1), flag=wx.EXPAND)
lbl_reserve = wx.StaticText(self.acct_info_area, label="XRP Reserved:")
self.st_reserve = wx.StaticText(self.acct_info_area, label="TBD")
main_sizer.Add(self.acct_info_area, 1, wx.EXPAND|wx.ALL, 25)
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.
# 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()
main_sizer.Add(self.sxb, 1, wx.ALL, 25)
self.Bind(wx.EVT_BUTTON, self.click_send_xrp, source=self.sxb)
# Ledger info text. One multi-line static text, unlike the account area.
self.ledger_info = wx.StaticText(main_panel, label="Not connected")
main_sizer.Add(self.ledger_info, 1, wx.EXPAND|wx.ALL, 25)
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 ------------------------------------
@@ -261,88 +301,142 @@ class TWaXLFrame(wx.Frame):
objs_panel.SetSizer(objs_sizer)
# Pop up to ask user for their account ---------------------------------
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")
value="snX6rmeLQasF2fLswCB7C4PwMSPD7")#TODO: remove test secret
value="rPT1Sjq2YGrBMTttX4GZHjKu9dyfzbpAYe")
account_dialog.Bind(wx.EVT_TEXT, self.toggle_dialog_style)
if account_dialog.ShowModal() == wx.ID_OK:
self.set_up_account(account_dialog.GetValue())
account_dialog.Destroy()
else:
if account_dialog.ShowModal() != wx.ID_OK:
# 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(wx.EVT_BUTTON, self.send_xrp, source=self.sxb)
self.Bind(EVT_NEW_LEDGER, self.update_ledger)
self.Bind(EVT_ACCT_INFO, self.update_account)
self.Bind(EVT_ACCT_TX, self.update_account_tx)
self.Bind(EVT_TX_SUB, self.add_tx_from_sub)
self.worker = XRPLMonitorThread(url, self, self.classic_address)
self.worker.start()
value = account_dialog.GetValue().strip()
account_dialog.Destroy()
def set_up_account(self, value):
value = value.strip()
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: {'a test network' if self.test_network else 'Mainnet'})")
f"(Client is on: {on_net})")
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(
classic_address = value
x_address = 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
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)
self.st_classic_address.SetLabel(self.classic_address)
self.st_x_address.SetLabel(self.xaddress)
def update_ledger(self, event):
message = event.data
# 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 (in drops of XRP) so we can calc account reserve
self.reserve_base = Decimal(message["reserve_base"])
self.reserve_inc = Decimal(message["reserve_inc"])
def update_account(self, event):
acct = event.data["account_data"]
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_drops = self.reserve_base + (self.reserve_inc * oc_decimal)
reserve_xrp = xrpl.utils.drops_to_xrp(str(reserve_drops))
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)
self.wallet.sequence = acct["Sequence"]
# Now that we have a sequence number we can enable the Send XRP button,
# if we aren't read-only.
if self.wallet:
self.sxb.Enable()
# Display account reserve.
reserve_xrp = self.calculate_reserve_xrp(acct.get("OwnerCount", 0))
if reserve_xrp != None:
self.st_reserve.SetLabel(str(reserve_xrp))
def enable_readwrite(self):
"""
Enable buttons for sending transactions.
"""
self.sxb.Enable()
self.sxb.SetToolTip("")
@staticmethod
def displayable_amount(a):
"""
Convert an arbitrary amount value from the XRPL to a string to be
displayed to the user:
- Convert drops of XRP to 6-digit decimal XRP (e.g. '12.345000 XRP')
- 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...'
@@ -361,7 +455,8 @@ class TWaXLFrame(wx.Frame):
def add_tx_row(self, t, prepend=False):
"""
Add one row to the account transaction history control.
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
@@ -389,26 +484,26 @@ class TWaXLFrame(wx.Frame):
else:
self.tx_list.AppendItem(cols)
def update_account_tx(self, event):
def update_account_tx(self, data):
"""
Update the transaction history tab with information from an account_tx
response.
"""
txs = event.data["transactions"]
# TODO: with pagination, we should leave existing items
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, event):
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.
Also, send a notification to the user about it.
"""
t = event.data
# Convert to same format as account_tx results
t["tx"] = t["transaction"]
if t["tx"]["hash"] in self.pending_tx_rows.keys():
@@ -420,11 +515,11 @@ class TWaXLFrame(wx.Frame):
# Scroll to top of list.
self.tx_list.EnsureVisible(self.tx_list.RowToItem(0))
# Send a notification message ("toast") about the transaction
# 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, an issuer gets notified when users transfer its tokens
# among themselves.
# 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)
@@ -435,79 +530,46 @@ class TWaXLFrame(wx.Frame):
Add a "pending" transaction to the history based on a transaction model
that was (presumably) just submitted.
"""
tx = txm.to_xrpl()
confirmation_time = "(pending)"
tx_type = tx["TransactionType"]
from_acct = tx.get("Account") or ""
tx_type = txm.transaction_type
from_acct = txm.account
if from_acct == self.classic_address:
from_acct = "(Me)"
to_acct = tx.get("Destination") or ""
# 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(tx))
tx_hash, str(txm.to_xrpl()))
self.tx_list.PrependItem(cols)
self.pending_tx_rows[tx_hash] = self.tx_list.RowToItem(0)
def send_xrp(self, event):
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")
dlg = SendXRPDialog(self, max_send=float(xrp_bal - tx_cost))
dlg = SendXRPDialog(self)
dlg.CenterOnScreen()
resp = dlg.ShowModal()
if resp != wx.ID_OK:
print("Send XRP canceled")
return
paydata = dlg.GetPaymentData()
# TODO: can we safely autofill with the client in another thread??
tx = {
"TransactionType": "Payment",
"Account": self.classic_address,
"Sequence": self.wallet.sequence,
"Destination": paydata["to"],
"Amount": xrpl.utils.xrp_to_drops(paydata["amt"]),
"Fee": "10",
#TODO: LLS
"Flags": 0
}
dtag = paydata.get("dtag")
if dtag is not None and dtag != "":
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["DestinationTag"] = dtag
txm = xrpl.models.transactions.transaction.Transaction.from_xrpl(tx)
signed_tx = xrpl.transaction.safe_sign_transaction(txm, self.wallet)
tx_blob = xrpl.core.binarycodec.encode(signed_tx.to_xrpl())
req = {
"command": "submit",
"tx_blob": tx_blob
}
nop = lambda x: x # TODO: actually handle response from sending
self.worker.client.jobq.put( (req, nop) )
self.add_pending_tx(signed_tx)
paydata = dlg.get_payment_data()
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()
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"
WS_URL = "wss://s.altnet.rippletest.net:51233" # Testnet
app = wx.App()
frame = TWaXLFrame(WS_URL)
frame = TWaXLFrame(WS_URL, test_network=True)
frame.Show()
app.MainLoop()

11
content/_code-samples/build-a-wallet/py/6_verification_and_polish.py
View File

@@ -243,14 +243,14 @@ class SendXRPDialog(wx.Dialog):
self.txt_amt.SetIncrement(1.0)
# The "Send" button is functionally an "OK" button except for the text.
btn_send = wx.Button(self, wx.ID_OK, label="Send")
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, btn_send)) )
(btn_cancel, self.btn_send)) )
sizer.Fit(self)
self.txt_dtag.Bind(wx.EVT_TEXT, self.on_dest_tag_edit)
@@ -287,8 +287,10 @@ class SendXRPDialog(wx.Dialog):
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))
@@ -359,6 +361,7 @@ class SendXRPDialog(wx.Dialog):
self.err_to.SetToolTip(err_msg)
self.err_to.Show()
else:
self.btn_send.Enable()
self.err_to.Hide()
@@ -712,6 +715,10 @@ class TWaXLFrame(wx.Frame):
paydata = dlg.get_payment_data()
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()
if __name__ == "__main__":
WS_URL = "wss://s.altnet.rippletest.net:51233" # Testnet