finish modularization of tracking wallet classes

- protocol-independent base classes remain in `tw*.py`
- protocol-dependent subclasses are in `base_proto/{name}/tw*.py`
This commit is contained in:
The MMGen Project 2022-02-06 13:28:45 +00:00
commit 1fb022d151
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
18 changed files with 347 additions and 237 deletions

View file

@ -28,15 +28,6 @@ from .addrlist import AddrListEntry,AddrListData,AddrList
class AddrData(MMGenObject):
msgs = {
'multiple_acct_addrs': """
ERROR: More than one address found for account: {acct!r}.
Your 'wallet.dat' file appears to have been altered by a non-{proj} program.
Please restore your tracking wallet from a backup or create a new one and
re-import your addresses.
"""
}
def __init__(self,proto,*args,**kwargs):
self.al_ids = {}
self.proto = proto
@ -107,15 +98,3 @@ class TwAddrData(AddrData,metaclass=AsyncInit):
for al_id in out:
self.add(AddrList(self.proto,al_id=al_id,adata=AddrListData(sorted(out[al_id],key=lambda a: a.idx))))
async def get_tw_data(self,wallet=None):
vmsg('Getting address data from tracking wallet')
c = self.rpc
if 'label_api' in c.caps:
accts = await c.call('listlabels')
ll = await c.batch_call('getaddressesbylabel',[(k,) for k in accts])
alists = [list(a.keys()) for a in ll]
else:
accts = await c.call('listaccounts',0,True)
alists = await c.batch_call('getaddressesbyaccount',[(k,) for k in accts])
return list(zip(accts,alists))

39
mmgen/base_proto/bitcoin/tw.py Executable file
View file

@ -0,0 +1,39 @@
#!/usr/bin/env python3
#
# mmgen = Multi-Mode GENerator, a command-line cryptocurrency wallet
# Copyright (C)2013-2022 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
# https://gitlab.com/mmgen/mmgen
"""
base_proto.bitcoin.tw: Bitcoin base protocol tracking wallet dependency classes
"""
from ...addrdata import TwAddrData
from ...util import vmsg
class BitcoinTwAddrData(TwAddrData):
msgs = {
'multiple_acct_addrs': """
ERROR: More than one address found for account: {acct!r}.
Your 'wallet.dat' file appears to have been altered by a non-{proj} program.
Please restore your tracking wallet from a backup or create a new one and
re-import your addresses.
"""
}
async def get_tw_data(self,wallet=None):
vmsg('Getting address data from tracking wallet')
c = self.rpc
if 'label_api' in c.caps:
accts = await c.call('listlabels')
ll = await c.batch_call('getaddressesbylabel',[(k,) for k in accts])
alists = [list(a.keys()) for a in ll]
else:
accts = await c.call('listaccounts',0,True)
alists = await c.batch_call('getaddressesbyaccount',[(k,) for k in accts])
return list(zip(accts,alists))

View file

@ -0,0 +1,112 @@
#!/usr/bin/env python3
#
# mmgen = Multi-Mode GENerator, a command-line cryptocurrency wallet
# Copyright (C)2013-2022 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
# https://gitlab.com/mmgen/mmgen
"""
base_proto.bitcoin.twaddrs: Bitcoin base protocol tracking wallet address list class
"""
from ...twaddrs import TwAddrList
from ...util import msg,die
from ...obj import MMGenList
from ...addr import CoinAddr
from ...rpc import rpc_init
from ...tw import get_tw_label
class BitcoinTwAddrList(TwAddrList):
has_age = True
async def __init__(self,proto,usr_addr_list,minconf,showempty,showbtcaddrs,all_labels,wallet=None):
def check_dup_mmid(acct_labels):
mmid_prev,err = None,False
for mmid in sorted(a.mmid for a in acct_labels if a):
if mmid == mmid_prev:
err = True
msg(f'Duplicate MMGen ID ({mmid}) discovered in tracking wallet!\n')
mmid_prev = mmid
if err:
die(4,'Tracking wallet is corrupted!')
def check_addr_array_lens(acct_pairs):
err = False
for label,addrs in acct_pairs:
if not label:
continue
if len(addrs) != 1:
err = True
if len(addrs) == 0:
msg(f'Label {label!r}: has no associated address!')
else:
msg(f'{addrs!r}: more than one {proto.coin} address in account!')
if err:
die(4,'Tracking wallet is corrupted!')
self.rpc = await rpc_init(proto)
self.total = proto.coin_amt('0')
self.proto = proto
lbl_id = ('account','label')['label_api' in self.rpc.caps]
for d in await self.rpc.call('listunspent',0):
if not lbl_id in d:
continue # skip coinbase outputs with missing account
if d['confirmations'] < minconf:
continue
label = get_tw_label(proto,d[lbl_id])
if label:
lm = label.mmid
if usr_addr_list and (lm not in usr_addr_list):
continue
if lm in self:
if self[lm]['addr'] != d['address']:
die(2,'duplicate {} address ({}) for this MMGen address! ({})'.format(
proto.coin,
d['address'],
self[lm]['addr'] ))
else:
lm.confs = d['confirmations']
lm.txid = d['txid']
lm.date = None
self[lm] = {
'amt': proto.coin_amt('0'),
'lbl': label,
'addr': CoinAddr(proto,d['address']) }
amt = proto.coin_amt(d['amount'])
self[lm]['amt'] += amt
self.total += amt
# We use listaccounts only for empty addresses, as it shows false positive balances
if showempty or all_labels:
# for compatibility with old mmids, must use raw RPC rather than native data for matching
# args: minconf,watchonly, MUST use keys() so we get list, not dict
if 'label_api' in self.rpc.caps:
acct_list = await self.rpc.call('listlabels')
aa = await self.rpc.batch_call('getaddressesbylabel',[(k,) for k in acct_list])
acct_addrs = [list(a.keys()) for a in aa]
else:
acct_list = list((await self.rpc.call('listaccounts',0,True)).keys()) # raw list, no 'L'
acct_addrs = await self.rpc.batch_call('getaddressesbyaccount',[(a,) for a in acct_list]) # use raw list here
acct_labels = MMGenList([get_tw_label(proto,a) for a in acct_list])
check_dup_mmid(acct_labels)
assert len(acct_list) == len(acct_addrs),(
'listaccounts() and getaddressesbyaccount() not equal in length')
addr_pairs = list(zip(acct_labels,acct_addrs))
check_addr_array_lens(addr_pairs)
for label,addr_arr in addr_pairs:
if not label:
continue
if all_labels and not showempty and not label.comment:
continue
if usr_addr_list and (label.mmid not in usr_addr_list):
continue
if label.mmid not in self:
self[label.mmid] = { 'amt':proto.coin_amt('0'), 'lbl':label, 'addr':'' }
if showbtcaddrs:
self[label.mmid]['addr'] = CoinAddr(proto,addr_arr[0])

View file

@ -0,0 +1,49 @@
#!/usr/bin/env python3
#
# mmgen = Multi-Mode GENerator, a command-line cryptocurrency wallet
# Copyright (C)2013-2022 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
# https://gitlab.com/mmgen/mmgen
"""
base_proto.bitcoin.twbal: Bitcoin base protocol tracking wallet balance class
"""
from ...twbal import TwGetBalance
from ...tw import get_tw_label
class BitcoinTwGetBalance(TwGetBalance):
fs = '{w:13} {u:<16} {p:<16} {c}'
async def create_data(self):
# 0: unconfirmed, 1: below minconf, 2: confirmed, 3: spendable (privkey in wallet)
lbl_id = ('account','label')['label_api' in self.rpc.caps]
for d in await self.rpc.call('listunspent',0):
lbl = get_tw_label(self.proto,d[lbl_id])
if lbl:
if lbl.mmid.type == 'mmgen':
key = lbl.mmid.obj.sid
if key not in self.data:
self.data[key] = [self.proto.coin_amt('0')] * 4
else:
key = 'Non-MMGen'
else:
lbl,key = None,'Non-wallet'
amt = self.proto.coin_amt(d['amount'])
if not d['confirmations']:
self.data['TOTAL'][0] += amt
self.data[key][0] += amt
conf_level = (1,2)[d['confirmations'] >= self.minconf]
self.data['TOTAL'][conf_level] += amt
self.data[key][conf_level] += amt
if d['spendable']:
self.data[key][3] += amt

View file

@ -0,0 +1,48 @@
#!/usr/bin/env python3
#
# mmgen = Multi-Mode GENerator, a command-line cryptocurrency wallet
# Copyright (C)2013-2022 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
# https://gitlab.com/mmgen/mmgen
"""
base_proto.bitcoin.twctl: Bitcoin base protocol tracking wallet control class
"""
from ...twctl import TrackingWallet
from ...util import rmsg,write_mode
class BitcoinTrackingWallet(TrackingWallet):
def init_empty(self):
self.data = { 'coin': self.proto.coin, 'addresses': {} }
def upgrade_wallet_maybe(self):
pass
async def rpc_get_balance(self,addr):
raise NotImplementedError('not implemented')
@write_mode
async def import_address(self,addr,label,rescan):
return await self.rpc.call('importaddress',addr,label,rescan,timeout=(False,3600)[rescan])
@write_mode
def batch_import_address(self,arg_list):
return self.rpc.batch_call('importaddress',arg_list)
@write_mode
async def remove_address(self,addr):
raise NotImplementedError(f'address removal not implemented for coin {self.proto.coin}')
@write_mode
async def set_label(self,coinaddr,lbl):
args = self.rpc.daemon.set_label_args( self.rpc, coinaddr, lbl )
try:
return await self.rpc.call(*args)
except Exception as e:
rmsg(e.args[0])
return False

View file

@ -0,0 +1,57 @@
#!/usr/bin/env python3
#
# mmgen = Multi-Mode GENerator, a command-line cryptocurrency wallet
# Copyright (C)2013-2022 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
# https://gitlab.com/mmgen/mmgen
"""
base_proto.bitcoin.twuo: Bitcoin base protocol tracking wallet unspent outputs class
"""
from ...twuo import TwUnspentOutputs
from ...addr import CoinAddr
class BitcoinTwUnspentOutputs(TwUnspentOutputs):
class MMGenTwUnspentOutput(TwUnspentOutputs.MMGenTwUnspentOutput):
# required by gen_unspent(); setting valid_attrs explicitly is also more efficient
valid_attrs = {'txid','vout','amt','amt2','label','twmmid','addr','confs','date','scriptPubKey','skip'}
invalid_attrs = {'proto'}
has_age = True
can_group = True
hdr_fmt = 'UNSPENT OUTPUTS (sort order: {}) Total {}: {}'
desc = 'unspent outputs'
item_desc = 'unspent output'
dump_fn_pfx = 'listunspent'
prompt_fs = 'Total to spend, excluding fees: {} {}\n\n'
prompt = """
Sort options: [t]xid, [a]mount, a[d]dress, [A]ge, [r]everse, [M]mgen addr
Display options: toggle [D]ays/date, show [g]roup, show [m]mgen addr, r[e]draw
Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
"""
key_mappings = {
't':'s_txid','a':'s_amt','d':'s_addr','A':'s_age','r':'d_reverse','M':'s_twmmid',
'D':'d_days','g':'d_group','m':'d_mmid','e':'d_redraw',
'q':'a_quit','p':'a_print','v':'a_view','w':'a_view_wide','l':'a_lbl_add' }
col_adj = 38
display_fs_fs = ' {{n:{cw}}} {{t:{tw}}} {{v:2}} {{a}} {{A}} {{c:<}}'
display_hdr_fs_fs = ' {{n:{cw}}} {{t:{tw}}} {{a}} {{A}} {{c:<}}'
print_fs_fs = ' {{n:4}} {{t:{tw}}} {{a}} {{m}} {{A:{aw}}} {cf}{{b:<8}} {{D:<19}} {{l}}'
async def get_unspent_rpc(self):
# bitcoin-cli help listunspent:
# Arguments:
# 1. minconf (numeric, optional, default=1) The minimum confirmations to filter
# 2. maxconf (numeric, optional, default=9999999) The maximum confirmations to filter
# 3. addresses (json array, optional, default=empty array) A json array of bitcoin addresses
# 4. include_unsafe (boolean, optional, default=true) Include outputs that are not safe to spend
# 5. query_options (json object, optional) JSON with query options
# for now, self.addrs is just an empty list for Bitcoin and friends
add_args = (9999999,self.addrs) if self.addrs else ()
return await self.rpc.call('listunspent',self.minconf,*add_args)

View file

@ -21,9 +21,17 @@ base_proto.ethereum.tw: Ethereum tracking wallet dependency classes
"""
from ...addrdata import TwAddrData
from ...util import vmsg
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,wallet=None):
from ...twctl import TrackingWallet
from ...util import vmsg

View file

@ -30,7 +30,7 @@ class EthereumTwUnspentOutputs(TwUnspentOutputs):
valid_attrs = {'txid','vout','amt','amt2','label','twmmid','addr','confs','skip'}
invalid_attrs = {'proto'}
disp_type = 'eth'
has_age = False
can_group = False
col_adj = 29
hdr_fmt = 'TRACKED ACCOUNTS (sort order: {})\nTotal {}: {}'
@ -48,6 +48,9 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view,
'm':'d_mmid','e':'d_redraw',
'q':'a_quit','p':'a_print','v':'a_view','w':'a_view_wide',
'l':'a_lbl_add','D':'a_addr_delete','R':'a_balance_refresh' }
display_fs_fs = ' {{n:{cw}}} {{a}} {{A}}'
print_fs_fs = ' {{n:4}} {{a}} {{m}} {{A:{aw}}} {{l}}'
display_hdr_fs_fs = display_fs_fs
async def __init__(self,proto,*args,**kwargs):
from ...globalvars import g
@ -76,9 +79,15 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view,
class EthereumTokenTwUnspentOutputs(EthereumTwUnspentOutputs):
disp_type = 'token'
prompt_fs = 'Total to spend: {} {}\n\n'
col_adj = 37
display_fs_fs = ' {{n:{cw}}} {{a}} {{A}} {{A2}}'
print_fs_fs = ' {{n:4}} {{a}} {{m}} {{A:{aw}}} {{A2:{aw}}} {{l}}'
display_hdr_fs_fs = display_fs_fs
async def __init__(self,proto,*args,**kwargs):
await super().__init__(proto,*args,**kwargs)
self.proto.tokensym = self.wallet.symbol
def get_display_precision(self):
return 10 # truncate precision for narrow display

View file

@ -1 +1 @@
13.1.dev014
13.1.dev015

View file

@ -47,8 +47,6 @@ class TwCommon:
@staticmethod
async def set_dates(rpc,us):
if rpc.proto.base_proto != 'Bitcoin':
return
if us and us[0].date is None:
# 'blocktime' differs from 'time', is same as getblockheader['time']
dates = [o['blocktime'] for o in await rpc.gathered_call('gettransaction',[(o.txid,) for o in us])]

View file

@ -23,99 +23,15 @@ twaddrs: Tracking wallet listaddresses class for the MMGen suite
from .color import green
from .util import msg,die,base_proto_subclass
from .base_obj import AsyncInit
from .obj import MMGenList,MMGenDict,TwComment
from .obj import MMGenDict,TwComment
from .addr import CoinAddr,MMGenID
from .rpc import rpc_init
from .tw import TwCommon,get_tw_label
from .tw import TwCommon
class TwAddrList(MMGenDict,TwCommon,metaclass=AsyncInit):
has_age = True
def __new__(cls,proto,*args,**kwargs):
return MMGenDict.__new__(base_proto_subclass(cls,proto,'twaddrs'),*args,**kwargs)
async def __init__(self,proto,usr_addr_list,minconf,showempty,showbtcaddrs,all_labels,wallet=None):
def check_dup_mmid(acct_labels):
mmid_prev,err = None,False
for mmid in sorted(a.mmid for a in acct_labels if a):
if mmid == mmid_prev:
err = True
msg(f'Duplicate MMGen ID ({mmid}) discovered in tracking wallet!\n')
mmid_prev = mmid
if err:
die(4,'Tracking wallet is corrupted!')
def check_addr_array_lens(acct_pairs):
err = False
for label,addrs in acct_pairs:
if not label: continue
if len(addrs) != 1:
err = True
if len(addrs) == 0:
msg(f'Label {label!r}: has no associated address!')
else:
msg(f'{addrs!r}: more than one {proto.coin} address in account!')
if err:
die(4,'Tracking wallet is corrupted!')
self.rpc = await rpc_init(proto)
self.total = proto.coin_amt('0')
self.proto = proto
lbl_id = ('account','label')['label_api' in self.rpc.caps]
for d in await self.rpc.call('listunspent',0):
if not lbl_id in d: continue # skip coinbase outputs with missing account
if d['confirmations'] < minconf: continue
label = get_tw_label(proto,d[lbl_id])
if label:
lm = label.mmid
if usr_addr_list and (lm not in usr_addr_list):
continue
if lm in self:
if self[lm]['addr'] != d['address']:
die(2,'duplicate {} address ({}) for this MMGen address! ({})'.format(
proto.coin,
d['address'],
self[lm]['addr'] ))
else:
lm.confs = d['confirmations']
lm.txid = d['txid']
lm.date = None
self[lm] = {
'amt': proto.coin_amt('0'),
'lbl': label,
'addr': CoinAddr(proto,d['address']) }
amt = proto.coin_amt(d['amount'])
self[lm]['amt'] += amt
self.total += amt
# We use listaccounts only for empty addresses, as it shows false positive balances
if showempty or all_labels:
# for compatibility with old mmids, must use raw RPC rather than native data for matching
# args: minconf,watchonly, MUST use keys() so we get list, not dict
if 'label_api' in self.rpc.caps:
acct_list = await self.rpc.call('listlabels')
aa = await self.rpc.batch_call('getaddressesbylabel',[(k,) for k in acct_list])
acct_addrs = [list(a.keys()) for a in aa]
else:
acct_list = list((await self.rpc.call('listaccounts',0,True)).keys()) # raw list, no 'L'
acct_addrs = await self.rpc.batch_call('getaddressesbyaccount',[(a,) for a in acct_list]) # use raw list here
acct_labels = MMGenList([get_tw_label(proto,a) for a in acct_list])
check_dup_mmid(acct_labels)
assert len(acct_list) == len(acct_addrs),(
'listaccounts() and getaddressesbyaccount() not equal in length')
addr_pairs = list(zip(acct_labels,acct_addrs))
check_addr_array_lens(addr_pairs)
for label,addr_arr in addr_pairs:
if not label: continue
if all_labels and not showempty and not label.comment: continue
if usr_addr_list and (label.mmid not in usr_addr_list): continue
if label.mmid not in self:
self[label.mmid] = { 'amt':proto.coin_amt('0'), 'lbl':label, 'addr':'' }
if showbtcaddrs:
self[label.mmid]['addr'] = CoinAddr(proto,addr_arr[0])
def raw_list(self):
return [((k if k.type == 'mmgen' else 'Non-MMGen'),self[k]['addr'],self[k]['amt']) for k in self]

View file

@ -25,12 +25,9 @@ from .util import base_proto_subclass
from .base_obj import AsyncInit
from .objmethods import MMGenObject
from .rpc import rpc_init
from .tw import get_tw_label
class TwGetBalance(MMGenObject,metaclass=AsyncInit):
fs = '{w:13} {u:<16} {p:<16} {c}'
def __new__(cls,proto,*args,**kwargs):
return MMGenObject.__new__(base_proto_subclass(cls,proto,'twbal'))
@ -43,35 +40,6 @@ class TwGetBalance(MMGenObject,metaclass=AsyncInit):
self.proto = proto
await self.create_data()
async def create_data(self):
# 0: unconfirmed, 1: below minconf, 2: confirmed, 3: spendable (privkey in wallet)
lbl_id = ('account','label')['label_api' in self.rpc.caps]
for d in await self.rpc.call('listunspent',0):
lbl = get_tw_label(self.proto,d[lbl_id])
if lbl:
if lbl.mmid.type == 'mmgen':
key = lbl.mmid.obj.sid
if key not in self.data:
self.data[key] = [self.proto.coin_amt('0')] * 4
else:
key = 'Non-MMGen'
else:
lbl,key = None,'Non-wallet'
amt = self.proto.coin_amt(d['amount'])
if not d['confirmations']:
self.data['TOTAL'][0] += amt
self.data[key][0] += amt
conf_level = (1,2)[d['confirmations'] >= self.minconf]
self.data['TOTAL'][conf_level] += amt
self.data[key][conf_level] += amt
if d['spendable']:
self.data[key][3] += amt
def format(self):
def gen_output():
if self.proto.chain_name != 'mainnet':

View file

@ -69,9 +69,6 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
self.conv_types(self.data[self.data_key])
self.cur_balances = {} # cache balances to prevent repeated lookups per program invocation
def init_empty(self):
self.data = { 'coin': self.proto.coin, 'addresses': {} }
def init_from_wallet_file(self):
import os,json
tw_dir = (
@ -130,9 +127,6 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
elif g.debug:
msg('read-only wallet, doing nothing')
def upgrade_wallet_maybe(self):
pass
def conv_types(self,ad):
for k,v in ad.items():
if k not in ('params','coin'):
@ -170,9 +164,6 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
self.cache_balance(addr,ret,self.cur_balances,self.data_root)
return ret
async def rpc_get_balance(self,addr):
raise NotImplementedError('not implemented')
@property
def sorted_list(self):
return sorted(
@ -186,14 +177,6 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
def mmid_ordered_dict(self):
return dict((x['mmid'],{'addr':x['addr'],'comment':x['comment']}) for x in self.sorted_list)
@write_mode
async def import_address(self,addr,label,rescan):
return await self.rpc.call('importaddress',addr,label,rescan,timeout=(False,3600)[rescan])
@write_mode
def batch_import_address(self,arg_list):
return self.rpc.batch_call('importaddress',arg_list)
def force_write(self):
mode_save = self.mode
self.mode = 'w'
@ -236,15 +219,6 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
from .twaddrs import TwAddrList
return addr in (await TwAddrList(self.proto,[],0,True,True,True,wallet=self)).coinaddr_list()
@write_mode
async def set_label(self,coinaddr,lbl):
args = self.rpc.daemon.set_label_args( self.rpc, coinaddr, lbl )
try:
return await self.rpc.call(*args)
except Exception as e:
rmsg(e.args[0])
return False
# returns on failure
@write_mode
async def add_label(self,arg1,label='',addr=None,silent=False,on_fail='return'):
@ -305,7 +279,3 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
@write_mode
async def remove_label(self,mmaddr):
await self.add_label(mmaddr,'')
@write_mode
async def remove_address(self,addr):
raise NotImplementedError(f'address removal not implemented for coin {self.proto.coin}')

View file

@ -51,23 +51,6 @@ class TwUnspentOutputs(MMGenObject,TwCommon,metaclass=AsyncInit):
return MMGenObject.__new__(base_proto_subclass(cls,proto,'twuo'))
txid_w = 64
disp_type = 'btc'
can_group = True
hdr_fmt = 'UNSPENT OUTPUTS (sort order: {}) Total {}: {}'
desc = 'unspent outputs'
item_desc = 'unspent output'
dump_fn_pfx = 'listunspent'
prompt_fs = 'Total to spend, excluding fees: {} {}\n\n'
prompt = """
Sort options: [t]xid, [a]mount, a[d]dress, [A]ge, [r]everse, [M]mgen addr
Display options: toggle [D]ays/date, show [g]roup, show [m]mgen addr, r[e]draw
Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
"""
key_mappings = {
't':'s_txid','a':'s_amt','d':'s_addr','A':'s_age','r':'d_reverse','M':'s_twmmid',
'D':'d_days','g':'d_group','m':'d_mmid','e':'d_redraw',
'q':'a_quit','p':'a_print','v':'a_view','w':'a_view_wide','l':'a_lbl_add' }
col_adj = 38
age_fmts_date_dependent = ('days','date','date_time')
age_fmts_interactive = ('confs','block','days','date')
_age_fmt = 'confs'
@ -87,10 +70,6 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
scriptPubKey = ImmutableAttr(HexStr)
skip = ListItemAttr(str,typeconv=False,reassign_ok=True)
# required by gen_unspent(); setting valid_attrs explicitly is also more efficient
valid_attrs = {'txid','vout','amt','amt2','label','twmmid','addr','confs','date','scriptPubKey','skip'}
invalid_attrs = {'proto'}
def __init__(self,proto,**kwargs):
self.__dict__['proto'] = proto
MMGenListItem.__init__(self,**kwargs)
@ -118,8 +97,6 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
from .twctl import TrackingWallet
self.wallet = await TrackingWallet(proto,mode='w')
if self.disp_type == 'token':
self.proto.tokensym = self.wallet.symbol
@property
def age_fmt(self):
@ -138,19 +115,6 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
def total(self):
return sum(i.amt for i in self.unspent)
async def get_unspent_rpc(self):
# bitcoin-cli help listunspent:
# Arguments:
# 1. minconf (numeric, optional, default=1) The minimum confirmations to filter
# 2. maxconf (numeric, optional, default=9999999) The maximum confirmations to filter
# 3. addresses (json array, optional, default=empty array) A json array of bitcoin addresses
# 4. include_unsafe (boolean, optional, default=true) Include outputs that are not safe to spend
# 5. query_options (json object, optional) JSON with query options
# for now, self.addrs is just an empty list for Bitcoin and friends
add_args = (9999999,self.addrs) if self.addrs else ()
return await self.rpc.call('listunspent',self.minconf,*add_args)
async def get_unspent_data(self,sort_key=None,reverse_sort=False):
us_raw = await self.get_unspent_rpc()
@ -242,7 +206,7 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
async def format_for_display(self):
unsp = self.unspent
if self.age_fmt in self.age_fmts_date_dependent:
if self.has_age and self.age_fmt in self.age_fmts_date_dependent:
await self.set_dates(self.rpc,unsp)
self.set_term_columns()
@ -260,24 +224,21 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
yield self.hdr_fmt.format(' '.join(self.sort_info()),self.proto.dcoin,self.total.hl())
if self.proto.chain_name != 'mainnet':
yield 'Chain: '+green(self.proto.chain_name.upper())
fs = { 'btc': ' {n:%s} {t:%s} {v:2} {a} {A} {c:<}' % (c.col1_w,c.tx_w),
'eth': ' {n:%s} {a} {A}' % c.col1_w,
'token': ' {n:%s} {a} {A} {A2}' % c.col1_w }[self.disp_type]
fs_hdr = ' {n:%s} {t:%s} {a} {A} {c:<}' % (c.col1_w,c.tx_w) if self.disp_type == 'btc' else fs
date_hdr = {
'confs': 'Confs',
'block': 'Block',
'days': 'Age(d)',
'date': 'Date',
'date_time': 'Date',
}
yield fs_hdr.format(
fs = self.display_fs_fs.format( cw=c.col1_w, tw=c.tx_w )
hdr_fs = self.display_hdr_fs_fs.format( cw=c.col1_w, tw=c.tx_w )
yield hdr_fs.format(
n = 'Num',
t = 'TXid'.ljust(c.tx_w - 2) + ' Vout',
a = 'Address'.ljust(c.addr_w),
A = f'Amt({self.proto.dcoin})'.ljust(self.disp_prec+5),
A2 = f' Amt({self.proto.coin})'.ljust(self.disp_prec+4),
c = date_hdr[self.age_fmt],
c = {
'confs': 'Confs',
'block': 'Block',
'days': 'Age(d)',
'date': 'Date',
'date_time': 'Date',
}[self.age_fmt],
).rstrip()
for n,i in enumerate(unsp):
@ -321,16 +282,14 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view, add [l]abel:
return self.fmt_display
async def format_for_printing(self,color=False,show_confs=True):
await self.set_dates(self.rpc,self.unspent)
if self.has_age:
await self.set_dates(self.rpc,self.unspent)
addr_w = max(len(i.addr) for i in self.unspent)
mmid_w = max(len(('',i.twmmid)[i.twmmid.type=='mmgen']) for i in self.unspent) or 12 # DEADBEEF:S:1
amt_w = self.proto.coin_amt.max_prec + 5
cfs = '{c:<8} ' if show_confs else ''
fs = {
'btc': (' {n:4} {t:%s} {a} {m} {A:%s} ' + cfs + '{b:<8} {D:<19} {l}') % (self.txid_w+3,amt_w),
'eth': ' {n:4} {a} {m} {A:%s} {l}' % amt_w,
'token': ' {n:4} {a} {m} {A:%s} {A2:%s} {l}' % (amt_w,amt_w)
}[self.disp_type]
fs = self.print_fs_fs.format(
tw = self.txid_w + 3,
cf = '{c:<8} ' if show_confs else '',
aw = self.proto.coin_amt.max_prec + 5 )
def gen_output():
yield fs.format(

View file

@ -669,9 +669,6 @@ def base_proto_subclass(cls,proto,modname):
"""
magic module loading and class selection
"""
if proto.base_proto != 'Ethereum':
return cls
modname = f'mmgen.base_proto.{proto.base_proto.lower()}.{modname}'
clsname = (
proto.mod_clsname

View file

@ -35,6 +35,7 @@ from mmgen.common import *
from mmgen.addrlist import *
from mmgen.passwdlist import *
from mmgen.tx.base import Base
from mmgen.base_proto.bitcoin.twuo import BitcoinTwUnspentOutputs
opts_data = {
'sets': [

View file

@ -110,7 +110,7 @@ tests = {
{},
),
# twuo.py
'TwUnspentOutputs.MMGenTwUnspentOutput': atd({
'BitcoinTwUnspentOutputs.MMGenTwUnspentOutput': atd({
'txid': (0b001, CoinTxID),
'vout': (0b001, int),
'amt': (0b001, BTCAmt),

View file

@ -9,4 +9,4 @@ if os.getenv('MMGEN_BOGUS_WALLET_DATA'):
from mmgen.fileutil import get_data_from_file
return json.loads(get_data_from_file(os.getenv('MMGEN_BOGUS_WALLET_DATA')),parse_float=Decimal)
TwUnspentOutputs.get_unspent_rpc = fake_get_unspent_rpc
BitcoinTwUnspentOutputs.get_unspent_rpc = fake_get_unspent_rpc