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

Build a wallet tutorial: show account objects

Browse Source
This commit is contained in:
mDuo13
2021-10-29 18:10:50 -07:00
parent b70b4d3f64
commit 7eeeb4c659
3 changed files with 319 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,11 +1,11 @@
# "Build a Wallet" tutorial, step 2: Watch ledger closes from a worker thread.
# "Build a Wallet" tutorial, step 3: Take account input & show account info
import xrpl
import wx
from threading import Thread
import wx.lib.newevent
# Set up an event type to pass info from the worker thread to the main thread
# 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):
@@ -99,7 +99,7 @@ class TWaXLFrame(wx.Frame):
self.set_up_account(account_dialog.GetValue())
account_dialog.Destroy()
else:
# If the user presses Cancel, exit the app.
# If the user presses Cancel on the account entry, exit the app.
exit(1)
self.Bind(EVT_NEW_LEDGER, self.update_ledger)
@@ -112,7 +112,6 @@ class TWaXLFrame(wx.Frame):
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:
# TODO: handle network mismatch error better
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'})")
@@ -136,7 +135,6 @@ class TWaXLFrame(wx.Frame):
self.xaddress = self.wallet.get_xaddress(is_test=self.test_network)
self.classic_address = self.wallet.classic_address
except Exception as e:
# TODO: handle invalid value better
print(e)
exit(1)
self.st_classic_address.SetLabel(self.classic_address)

12
content/_code-samples/build-a-wallet/py/README.md
View File

@@ -1,3 +1,15 @@
# Build a Wallet Sample Code (Python)
This folder contains sample code for a non-custodial XRP Ledger wallet application in Python.
Setup:
```sh
pip install -r requirements.txt
```
Run any of the Python scripts (higher numbers are more complete/advanced examples):
```sh
python3 1_hello.py
```

304
content/_code-samples/build-a-wallet/py/x_owned_objects.py Normal file
View File

@@ -0,0 +1,304 @@
# "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()