diff --git a/mmgen/addrdata.py b/mmgen/addrdata.py index 1608a6c0..d67554fe 100755 --- a/mmgen/addrdata.py +++ b/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())] diff --git a/mmgen/proto/eth/addrdata.py b/mmgen/proto/eth/addrdata.py index 2611f4c0..182381c4 100755 --- a/mmgen/proto/eth/addrdata.py +++ b/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 diff --git a/mmgen/proto/eth/tw/ctl.py b/mmgen/proto/eth/tw/ctl.py index 30098e9e..f8bb1633 100755 --- a/mmgen/proto/eth/tw/ctl.py +++ b/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. diff --git a/mmgen/tw/store.py b/mmgen/tw/store.py new file mode 100755 index 00000000..5dd95536 --- /dev/null +++ b/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 +# 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))})