tw: new `TwAddrDataWithStore`, `TwCtlWithStore` classes

mmgen/addrdata.py

@@ -105,3 +105,20 @@ class TwAddrData(AddrData, metaclass=AsyncInit):
 				al_id = al_id,
 				adata = AddrListData(sorted(out[al_id], key=lambda a: a.idx))
 			))
+
+class TwAddrDataWithStore(TwAddrData):
+
+	msgs = {
+		'multiple_acct_addrs': """
+			ERROR: More than one address found for account: {acct!r}.
+			Your tracking wallet is corrupted!
+		"""
+	}
+
+	async def get_tw_data(self, *, twctl=None):
+		self.cfg._util.vmsg('Getting address data from tracking wallet')
+		if twctl is None:
+			from .tw.ctl import TwCtl
+			twctl = await TwCtl(self.cfg, self.proto)
+		# emulate the output of RPC 'listaccounts' and 'getaddressesbyaccount'
+		return [(mmid+' '+d['comment'], [d['addr']]) for mmid, d in list(twctl.mmid_ordered_dict.items())]

mmgen/proto/eth/addrdata.py

@@ -20,23 +20,10 @@
 proto.eth.addrdata: Ethereum TwAddrData classes
 """
 
-from ...addrdata import TwAddrData
+from ...addrdata import TwAddrDataWithStore
 
-class EthereumTwAddrData(TwAddrData):
-
-	msgs = {
-		'multiple_acct_addrs': """
-			ERROR: More than one address found for account: {acct!r}.
-			Your tracking wallet is corrupted!
-		"""
-	}
-
-	async def get_tw_data(self, *, twctl=None):
-		from ...tw.ctl import TwCtl
-		self.cfg._util.vmsg('Getting address data from tracking wallet')
-		twctl = (twctl or await TwCtl(self.cfg, self.proto)).mmid_ordered_dict
-		# emulate the output of RPC 'listaccounts' and 'getaddressesbyaccount'
-		return [(mmid+' '+d['comment'], [d['addr']]) for mmid, d in list(twctl.items())]
+class EthereumTwAddrData(TwAddrDataWithStore):
+	pass
 
 class EthereumTokenTwAddrData(EthereumTwAddrData):
 	pass

mmgen/proto/eth/tw/ctl.py

@@ -21,16 +21,15 @@ proto.eth.tw.ctl: Ethereum tracking wallet control class
 """
 
 from ....util import msg, ymsg, die, cached_property
-from ....tw.ctl import TwCtl, write_mode, label_addr_pair
-from ....tw.shared import TwLabel
-from ....addr import is_coin_addr, is_mmgen_id, CoinAddr
+from ....tw.store import TwCtlWithStore
+from ....tw.ctl import write_mode
+from ....addr import is_coin_addr
+
 from ..contract import Token
 
-class EthereumTwCtl(TwCtl):
+class EthereumTwCtl(TwCtlWithStore):
 
-	caps = ('batch',)
 	data_key = 'accounts'
-	use_tw_file = True
 
 	def init_empty(self):
 		self.data = {
@@ -86,54 +85,6 @@ class EthereumTwCtl(TwCtl):
 			int(await self.rpc.call('eth_getBalance', '0x' + addr, block), 16),
 			from_unit = 'wei')
 
-	@write_mode
-	async def batch_import_address(self, args_list):
-		return [await self.import_address(a, label=b, rescan=c) for a, b, c in args_list]
-
-	async def rescan_addresses(self, coin_addrs):
-		pass
-
-	@write_mode
-	async def import_address(self, addr, *, label, rescan=False):
-		r = self.data_root
-		if addr in r:
-			if not r[addr]['mmid'] and label.mmid:
-				msg(f'Warning: MMGen ID {label.mmid!r} was missing in tracking wallet!')
-			elif r[addr]['mmid'] != label.mmid:
-				die(3, 'MMGen ID {label.mmid!r} does not match tracking wallet!')
-		r[addr] = {'mmid': label.mmid, 'comment': label.comment}
-
-	@write_mode
-	async def remove_address(self, addr):
-		r = self.data_root
-
-		if is_coin_addr(self.proto, addr):
-			have_match = lambda k: k == addr
-		elif is_mmgen_id(self.proto, addr):
-			have_match = lambda k: r[k]['mmid'] == addr
-		else:
-			die(1, f'{addr!r} is not an Ethereum address or MMGen ID')
-
-		for k in r:
-			if have_match(k):
-				# return the addr resolved to mmid if possible
-				ret = r[k]['mmid'] if is_mmgen_id(self.proto, r[k]['mmid']) else addr
-				del r[k]
-				self.write()
-				return ret
-		msg(f'Address {addr!r} not found in {self.data_root_desc!r} section of tracking wallet')
-		return None
-
-	@write_mode
-	async def set_label(self, coinaddr, lbl):
-		for addr, d in list(self.data_root.items()):
-			if addr == coinaddr:
-				d['comment'] = lbl.comment
-				self.write()
-				return True
-		msg(f'Address {coinaddr!r} not found in {self.data_root_desc!r} section of tracking wallet')
-		return False
-
 	async def addr2sym(self, req_addr):
 		for addr in self.data['tokens']:
 			if addr == req_addr:
@@ -144,29 +95,6 @@ class EthereumTwCtl(TwCtl):
 			if self.data['tokens'][addr]['params']['symbol'].upper() == sym.upper():
 				return addr
 
-	def get_token_param(self, token, param):
-		if token in self.data['tokens']:
-			return self.data['tokens'][token]['params'].get(param)
-
-	@property
-	def sorted_list(self):
-		return sorted([{
-				'addr':    x[0],
-				'mmid':    x[1]['mmid'],
-				'comment': x[1]['comment']
-			} for x in self.data_root.items() if x[0] not in ('params', 'coin')],
-			key = lambda x: x['mmid'].sort_key + x['addr'])
-
-	@property
-	def mmid_ordered_dict(self):
-		return dict((x['mmid'], {'addr': x['addr'], 'comment': x['comment']}) for x in self.sorted_list)
-
-	async def get_label_addr_pairs(self):
-		return [label_addr_pair(
-				TwLabel(self.proto, f"{mmid} {d['comment']}"),
-				CoinAddr(self.proto, d['addr'])
-			) for mmid, d in self.mmid_ordered_dict.items()]
-
 	# Since it’s nearly impossible to empty an Ethereum account, consider set of used addresses
 	# to be all accounts with balances.
 	# Token addresses might have a balance but no corresponding ETH balance, so check them too.

mmgen/tw/store.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+#
+# MMGen Wallet, a terminal-based cryptocurrency wallet
+# Copyright (C)2013-2025 The MMGen Project <mmgen@tuta.io>
+# Licensed under the GNU General Public License, Version 3:
+#   https://www.gnu.org/licenses
+# Public project repositories:
+#   https://github.com/mmgen/mmgen-wallet
+#   https://gitlab.com/mmgen/mmgen-wallet
+
+"""
+tw.store: Tracking wallet control class with store
+"""
+
+from ..util import msg, die, cached_property
+from ..addr import is_coin_addr, is_mmgen_id, CoinAddr
+
+from .shared import TwLabel
+from .ctl import TwCtl, write_mode, label_addr_pair
+
+class TwCtlWithStore(TwCtl):
+
+	caps = ('batch',)
+	data_key = 'addresses'
+	use_tw_file = True
+
+	def init_empty(self):
+		self.data = {
+			'coin': self.proto.coin,
+			'network': self.proto.network.upper(),
+			'addresses': {},
+		}
+
+	@write_mode
+	async def batch_import_address(self, args_list):
+		return [await self.import_address(a, label=b, rescan=c) for a, b, c in args_list]
+
+	async def rescan_addresses(self, coin_addrs):
+		pass
+
+	@write_mode
+	async def import_address(self, addr, *, label, rescan=False):
+		r = self.data_root
+		if addr in r:
+			if not r[addr]['mmid'] and label.mmid:
+				msg(f'Warning: MMGen ID {label.mmid!r} was missing in tracking wallet!')
+			elif r[addr]['mmid'] != label.mmid:
+				die(3, 'MMGen ID {label.mmid!r} does not match tracking wallet!')
+		r[addr] = {'mmid': label.mmid, 'comment': label.comment}
+
+	@write_mode
+	async def remove_address(self, addr):
+		r = self.data_root
+
+		if is_coin_addr(self.proto, addr):
+			have_match = lambda k: k == addr
+		elif is_mmgen_id(self.proto, addr):
+			have_match = lambda k: r[k]['mmid'] == addr
+		else:
+			die(1, f'{addr!r} is not an Ethereum address or MMGen ID')
+
+		for k in r:
+			if have_match(k):
+				# return the addr resolved to mmid if possible
+				ret = r[k]['mmid'] if is_mmgen_id(self.proto, r[k]['mmid']) else addr
+				del r[k]
+				self.write()
+				return ret
+		msg(f'Address {addr!r} not found in {self.data_root_desc!r} section of tracking wallet')
+		return None
+
+	@write_mode
+	async def set_label(self, coinaddr, lbl):
+		for addr, d in list(self.data_root.items()):
+			if addr == coinaddr:
+				d['comment'] = lbl.comment
+				self.write()
+				return True
+		msg(f'Address {coinaddr!r} not found in {self.data_root_desc!r} section of tracking wallet')
+		return False
+
+	@property
+	def sorted_list(self):
+		return sorted([{
+				'addr':    x[0],
+				'mmid':    x[1]['mmid'],
+				'comment': x[1]['comment']
+			} for x in self.data_root.items() if x[0] not in ('params', 'coin')],
+			key = lambda x: x['mmid'].sort_key + x['addr'])
+
+	@property
+	def mmid_ordered_dict(self):
+		return dict((x['mmid'], {'addr': x['addr'], 'comment': x['comment']}) for x in self.sorted_list)
+
+	async def get_label_addr_pairs(self):
+		return [label_addr_pair(
+				TwLabel(self.proto, f"{mmid} {d['comment']}"),
+				CoinAddr(self.proto, d['addr'])
+			) for mmid, d in self.mmid_ordered_dict.items()]
+
+	@cached_property
+	def used_addrs(self):
+		from decimal import Decimal
+		# TODO: for now, consider used addrs to be addrs with balance
+		return ({k for k, v in self.data['addresses'].items() if Decimal(v.get('balance', 0))})