mmgen/mmgen-wallet
1
Fork 0
Code Releases 28 Wiki Activity

tw/unspent.py: move shared code to tw/common.py

Browse source
This commit is contained in:
The MMGen Project 2022-05-23 16:28:52 +00:00
commit 9e1fe9d862
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
5 changed files with 175 additions and 163 deletions
Download patch file Download diff file Expand all files Collapse all files

3
mmgen/base_proto/bitcoin/tw/unspent.py
View file

@ -24,9 +24,10 @@ class BitcoinTwUnspentOutputs(TwUnspentOutputs):
has_age = True
can_group = True
hdr_fmt = 'UNSPENT OUTPUTS (sort order: {}) Total {}: {}'
hdr_fmt = 'UNSPENT OUTPUTS (sort order: {a}) Total {b}: {c}'
desc = 'unspent outputs'
item_desc = 'unspent output'
no_data_errmsg = 'No unspent outputs in tracking wallet!'
dump_fn_pfx = 'listunspent'
prompt_fs = 'Total to spend, excluding fees: {} {}\n\n'
prompt = """

6
mmgen/base_proto/ethereum/tw/unspent.py
View file

@ -33,7 +33,7 @@ class EthereumTwUnspentOutputs(TwUnspentOutputs):
has_age = False
can_group = False
col_adj = 29
hdr_fmt = 'TRACKED ACCOUNTS (sort order: {})\nTotal {}: {}'
hdr_fmt = 'TRACKED ACCOUNTS (sort order: {a})\nTotal {b}: {c}'
desc = 'account balances'
item_desc = 'account'
dump_fn_pfx = 'balances'
@ -51,6 +51,7 @@ Actions: [q]uit view, [p]rint to file, pager [v]iew, [w]ide view,
display_fs_fs = ' {{n:{cw}}} {{a}} {{A}}'
print_fs_fs = ' {{n:4}} {{a}} {{m}} {{A:{aw}}} {{l}}'
display_hdr_fs_fs = display_fs_fs
no_data_errmsg = 'No accounts in tracking wallet!'
async def __init__(self,proto,*args,**kwargs):
from ....globalvars import g
@ -89,7 +90,8 @@ class EthereumTokenTwUnspentOutputs(EthereumTwUnspentOutputs):
await super().__init__(proto,*args,**kwargs)
self.proto.tokensym = self.wallet.symbol
def get_display_precision(self):
@property
def disp_prec(self):
return 10 # truncate precision for narrow display
async def get_data(self,*args,**kwargs):

159
mmgen/tw/common.py
View file

@ -22,14 +22,27 @@ tw: Tracking wallet dependency classes and helper functions
import time
from ..globalvars import g
from ..objmethods import Hilite,InitErrors,MMGenObject
from ..obj import TwComment
from ..color import red,yellow
from ..util import msg,msg_r,die,line_input,do_pager,capfirst
from ..addr import MMGenID
# mixin class for TwUnspentOutputs,TwAddrList:
class TwCommon:
fmt_display = ''
fmt_print = ''
cols = None
reverse = False
group = False
sort_key = 'age'
age_fmts = ('confs','block','days','date','date_time')
age_fmts_date_dependent = ('days','date','date_time')
age_fmts_interactive = ('confs','block','days','date')
_age_fmt = 'confs'
date_formatter = {
'days': lambda rpc,secs: (rpc.cur_date - secs) // 86400,
@ -53,6 +66,152 @@ class TwCommon:
for idx,o in enumerate(us):
o.date = dates[idx]
@property
def age_fmt(self):
return self._age_fmt
@age_fmt.setter
def age_fmt(self,val):
if val not in self.age_fmts:
die( 'BadAgeFormat', f'{val!r}: invalid age format (must be one of {self.age_fmts!r})' )
self._age_fmt = val
@property
def disp_prec(self):
return self.proto.coin_amt.max_prec
def set_term_columns(self):
from ..term import get_terminal_size
while True:
self.cols = g.terminal_width or get_terminal_size().width
if self.cols >= g.min_screen_width:
break
line_input(
'Screen too narrow to display the tracking wallet\n'
+ f'Please resize your screen to at least {g.min_screen_width} characters and hit ENTER ' )
def sort_info(self,include_group=True):
ret = ([],['Reverse'])[self.reverse]
ret.append(capfirst(self.sort_key).replace('Twmmid','MMGenID'))
if include_group and self.group and (self.sort_key in ('addr','txid','twmmid')):
ret.append('Grouped')
return ret
def do_sort(self,key=None,reverse=False):
sort_funcs = {
'addr': lambda i: i.addr,
'age': lambda i: 0 - i.confs,
'amt': lambda i: i.amt,
'txid': lambda i: f'{i.txid} {i.vout:04}',
'twmmid': lambda i: i.twmmid.sort_key
}
key = key or self.sort_key
if key not in sort_funcs:
die(1,f'{key!r}: invalid sort key. Valid options: {" ".join(sort_funcs.keys())}')
self.sort_key = key
assert type(reverse) == bool
self.data.sort(key=sort_funcs[key],reverse=reverse or self.reverse)
async def view_and_sort(self,tx):
from ..opts import opt
from ..term import get_char
prompt = self.prompt.strip() + '\b'
no_output = False
oneshot_msg = None
CUR_HOME = '\033[H'
ERASE_ALL = '\033[0J'
CUR_RIGHT = lambda n: f'\033[{n}C'
while True:
msg_r('' if no_output else '\n\n' if opt.no_blank else CUR_HOME+ERASE_ALL)
reply = get_char(
'' if no_output else await self.format_for_display()+'\n'+(oneshot_msg or '')+prompt,
immed_chars=''.join(self.key_mappings.keys())
)
no_output = False
oneshot_msg = '' if oneshot_msg else None # tristate, saves previous state
if reply not in self.key_mappings:
msg_r('\ninvalid keypress ')
time.sleep(0.5)
continue
action = self.key_mappings[reply]
if action.startswith('s_'):
self.do_sort(action[2:])
if action == 's_twmmid':
self.show_mmid = True
elif action == 'd_days':
af = self.age_fmts_interactive
self.age_fmt = af[(af.index(self.age_fmt) + 1) % len(af)]
elif action == 'd_mmid':
self.show_mmid = not self.show_mmid
elif action == 'd_group':
if self.can_group:
self.group = not self.group
elif action == 'd_redraw':
pass
elif action == 'd_reverse':
self.data.reverse()
self.reverse = not self.reverse
elif action == 'a_quit':
msg('')
return self.data
elif action == 'a_balance_refresh':
idx = self.get_idx_from_user(action)
if idx:
e = self.data[idx-1]
bal = await self.wallet.get_balance(e.addr,force_rpc=True)
await self.get_data()
oneshot_msg = yellow(f'{self.proto.dcoin} balance for account #{idx} refreshed\n\n')
self.display_constants = self.get_display_constants()
elif action == 'a_lbl_add':
idx,lbl = self.get_idx_from_user(action)
if idx:
e = self.data[idx-1]
if await self.wallet.add_label(e.twmmid,lbl,addr=e.addr):
await self.get_data()
oneshot_msg = yellow('Label {} {} #{}\n\n'.format(
('added to' if lbl else 'removed from'),
self.item_desc,
idx ))
else:
oneshot_msg = red('Label could not be added\n\n')
self.display_constants = self.get_display_constants()
elif action == 'a_addr_delete':
idx = self.get_idx_from_user(action)
if idx:
e = self.data[idx-1]
if await self.wallet.remove_address(e.addr):
await self.get_data()
oneshot_msg = yellow(f'{capfirst(self.item_desc)} #{idx} removed\n\n')
else:
oneshot_msg = red('Address could not be removed\n\n')
self.display_constants = self.get_display_constants()
elif action == 'a_print':
of = '{}-{}[{}].out'.format(
self.dump_fn_pfx,
self.proto.dcoin,
','.join(self.sort_info(include_group=False)).lower() )
msg('')
from ..fileutil import write_data_to_file
from ..exception import UserNonConfirmation
try:
write_data_to_file(
of,
await self.format_for_printing(color=False),
desc = f'{self.desc} listing' )
except UserNonConfirmation as e:
oneshot_msg = red(f'File {of!r} not overwritten by user request\n\n')
else:
oneshot_msg = yellow(f'Data written to {of!r}\n\n')
elif action in ('a_view','a_view_wide'):
do_pager(
self.fmt_display if action == 'a_view' else
await self.format_for_printing(color=True) )
if g.platform == 'linux' and oneshot_msg == None:
msg_r(CUR_RIGHT(len(prompt.split('\n')[-1])-2))
no_output = True
class TwMMGenID(str,Hilite,InitErrors,MMGenObject):
color = 'orange'
width = 0

166
mmgen/tw/unspent.py
View file

@ -40,7 +40,7 @@ from ..util import (
)
from ..base_obj import AsyncInit
from ..objmethods import MMGenObject
from ..obj import ImmutableAttr,ListItemAttr,MMGenListItem,TwComment,get_obj,HexStr,CoinTxID,MMGenIdx
from ..obj import ImmutableAttr,ListItemAttr,MMGenListItem,TwComment,get_obj,HexStr,CoinTxID,MMGenIdx,MMGenList
from ..addr import CoinAddr,MMGenID
from ..rpc import rpc_init
from .common import TwCommon,TwMMGenID,get_tw_label
@ -51,11 +51,6 @@ class TwUnspentOutputs(MMGenObject,TwCommon,metaclass=AsyncInit):
return MMGenObject.__new__(base_proto_tw_subclass(cls,proto,'unspent'))
txid_w = 64
age_fmts_date_dependent = ('days','date','date_time')
age_fmts_interactive = ('confs','block','days','date')
_age_fmt = 'confs'
class MMGenTwOutputList(list,MMGenObject): pass
class MMGenTwUnspentOutput(MMGenListItem):
txid = ListItemAttr(CoinTxID)
@ -82,35 +77,15 @@ class TwUnspentOutputs(MMGenObject,TwCommon,metaclass=AsyncInit):
async def __init__(self,proto,minconf=1,addrs=[]):
self.proto = proto
self.data = self.MMGenTwOutputList()
self.fmt_display = ''
self.fmt_print = ''
self.cols = None
self.reverse = False
self.group = False
self.data = MMGenList()
self.show_mmid = True
self.minconf = minconf
self.addrs = addrs
self.sort_key = 'age'
self.disp_prec = self.get_display_precision()
self.rpc = await rpc_init(proto)
from .ctl import TrackingWallet
self.wallet = await TrackingWallet(proto,mode='w')
@property
def age_fmt(self):
return self._age_fmt
@age_fmt.setter
def age_fmt(self,val):
if val not in self.age_fmts:
die( 'BadAgeFormat', f'{val!r}: invalid age format (must be one of {self.age_fmts!r})' )
self._age_fmt = val
def get_display_precision(self):
return self.proto.coin_amt.max_prec
@property
def total(self):
return sum(i.amt for i in self.data)
@ -144,45 +119,13 @@ class TwUnspentOutputs(MMGenObject,TwCommon,metaclass=AsyncInit):
self.proto,
**{ k:v for k,v in o.items() if k in self.MMGenTwUnspentOutput.valid_attrs } )
self.data = self.MMGenTwOutputList(gen_unspent())
self.data = MMGenList(gen_unspent())
if not self.data:
die(1, f'No tracked {self.item_desc}s in tracking wallet!')
die(1,self.no_data_errmsg)
self.do_sort(key=sort_key,reverse=reverse_sort)
def do_sort(self,key=None,reverse=False):
sort_funcs = {
'addr': lambda i: i.addr,
'age': lambda i: 0 - i.confs,
'amt': lambda i: i.amt,
'txid': lambda i: f'{i.txid} {i.vout:04}',
'twmmid': lambda i: i.twmmid.sort_key
}
key = key or self.sort_key
if key not in sort_funcs:
die(1,f'{key!r}: invalid sort key. Valid options: {" ".join(sort_funcs.keys())}')
self.sort_key = key
assert type(reverse) == bool
self.data.sort(key=sort_funcs[key],reverse=reverse or self.reverse)
def sort_info(self,include_group=True):
ret = ([],['Reverse'])[self.reverse]
ret.append(capfirst(self.sort_key).replace('Twmmid','MMGenID'))
if include_group and self.group and (self.sort_key in ('addr','txid','twmmid')):
ret.append('Grouped')
return ret
def set_term_columns(self):
from ..term import get_terminal_size
while True:
self.cols = g.terminal_width or get_terminal_size().width
if self.cols >= g.min_screen_width:
break
line_input(
'Screen too narrow to display the tracking wallet\n'
+ f'Please resize your screen to at least {g.min_screen_width} characters and hit ENTER ' )
def get_display_constants(self):
data = self.data
for i in data:
@ -221,7 +164,10 @@ class TwUnspentOutputs(MMGenObject,TwCommon,metaclass=AsyncInit):
b.skip = (k,'addr')[k=='twmmid']
def gen_output():
yield self.hdr_fmt.format(' '.join(self.sort_info()),self.proto.dcoin,self.total.hl())
yield self.hdr_fmt.format(
a = ' '.join(self.sort_info()),
b = self.proto.dcoin,
c = self.total.hl() if hasattr(self,'total') else None )
if self.proto.chain_name != 'mainnet':
yield 'Chain: '+green(self.proto.chain_name.upper())
fs = self.display_fs_fs.format( cw=c.col1_w, tw=c.tx_w )
@ -385,99 +331,3 @@ class TwUnspentOutputs(MMGenObject,TwCommon,metaclass=AsyncInit):
fs = 'Refreshing tracking wallet {} #{}. Is this what you want?'
if keypress_confirm(fs.format(self.item_desc,n)):
return n
async def view_and_sort(self,tx):
from ..term import get_char
prompt = self.prompt.strip() + '\b'
no_output,oneshot_msg = False,None
from ..opts import opt
CUR_HOME,ERASE_ALL = '\033[H','\033[0J'
CUR_RIGHT = lambda n: f'\033[{n}C'
while True:
msg_r('' if no_output else '\n\n' if opt.no_blank else CUR_HOME+ERASE_ALL)
reply = get_char(
'' if no_output else await self.format_for_display()+'\n'+(oneshot_msg or '')+prompt,
immed_chars=''.join(self.key_mappings.keys())
)
no_output = False
oneshot_msg = '' if oneshot_msg else None # tristate, saves previous state
if reply not in self.key_mappings:
msg_r('\ninvalid keypress ')
time.sleep(0.5)
continue
action = self.key_mappings[reply]
if action[:2] == 's_':
self.do_sort(action[2:])
if action == 's_twmmid': self.show_mmid = True
elif action == 'd_days':
af = self.age_fmts_interactive
self.age_fmt = af[(af.index(self.age_fmt) + 1) % len(af)]
elif action == 'd_mmid':
self.show_mmid = not self.show_mmid
elif action == 'd_group':
if self.can_group:
self.group = not self.group
elif action == 'd_redraw':
pass
elif action == 'd_reverse':
self.data.reverse()
self.reverse = not self.reverse
elif action == 'a_quit':
msg('')
return self.data
elif action == 'a_balance_refresh':
idx = self.get_idx_from_user(action)
if idx:
e = self.data[idx-1]
bal = await self.wallet.get_balance(e.addr,force_rpc=True)
await self.get_data()
oneshot_msg = yellow(f'{self.proto.dcoin} balance for account #{idx} refreshed\n\n')
self.display_constants = self.get_display_constants()
elif action == 'a_lbl_add':
idx,lbl = self.get_idx_from_user(action)
if idx:
e = self.data[idx-1]
if await self.wallet.add_label(e.twmmid,lbl,addr=e.addr):
await self.get_data()
oneshot_msg = yellow('Label {} {} #{}\n\n'.format(
('added to' if lbl else 'removed from'),
self.item_desc,
idx ))
else:
oneshot_msg = red('Label could not be added\n\n')
self.display_constants = self.get_display_constants()
elif action == 'a_addr_delete':
idx = self.get_idx_from_user(action)
if idx:
e = self.data[idx-1]
if await self.wallet.remove_address(e.addr):
await self.get_data()
oneshot_msg = yellow(f'{capfirst(self.item_desc)} #{idx} removed\n\n')
else:
oneshot_msg = red('Address could not be removed\n\n')
self.display_constants = self.get_display_constants()
elif action == 'a_print':
of = '{}-{}[{}].out'.format(
self.dump_fn_pfx,
self.proto.dcoin,
','.join(self.sort_info(include_group=False)).lower() )
msg('')
from ..fileutil import write_data_to_file
try:
write_data_to_file(
of,
await self.format_for_printing(),
desc = f'{self.desc} listing' )
except UserNonConfirmation as e:
oneshot_msg = red(f'File {of!r} not overwritten by user request\n\n')
else:
oneshot_msg = yellow(f'Data written to {of!r}\n\n')
elif action in ('a_view','a_view_wide'):
do_pager(
self.fmt_display if action == 'a_view' else
await self.format_for_printing(color=True) )
if g.platform == 'linux' and oneshot_msg == None:
msg_r(CUR_RIGHT(len(prompt.split('\n')[-1])-2))
no_output = True

4
mmgen/tx/new.py
View file

@ -16,7 +16,7 @@ from ..globalvars import *
from ..opts import opt
from .base import Base
from ..color import pink
from ..obj import get_obj
from ..obj import get_obj,MMGenList
from ..util import msg,qmsg,fmt,die,suf,remove_dups,get_extension,keypress_confirm,do_license_msg,line_input
from ..addr import is_mmgen_id,CoinAddr,is_coin_addr
@ -293,7 +293,7 @@ class New(Base):
sel_nums = us_f(self.twuo.data)
msg(f'Selected output{suf(sel_nums)}: {{}}'.format(' '.join(str(n) for n in sel_nums)))
sel_unspent = self.twuo.MMGenTwOutputList([self.twuo.data[i-1] for i in sel_nums])
sel_unspent = MMGenList(self.twuo.data[i-1] for i in sel_nums)
inputs_sum = sum(s.amt for s in sel_unspent)
if not await self.precheck_sufficient_funds(inputs_sum,sel_unspent,outputs_sum):