addr.py: move AddrData and TwAddrData classes to addrdata.py
This commit is contained in:
parent
012fea543f
commit
5961d1c36d
GPG key ID:
3F8B1861E32B7DA2
6 changed files with 127 additions and 100 deletions
|
|
@ -828,97 +828,3 @@ Record this checksum: it will be used to verify the password file in the future
|
|||
from .crypto import scramble_seed
|
||||
dmsg_sc('str',scramble_key)
|
||||
return scramble_seed(seed,scramble_key.encode())
|
||||
|
||||
class AddrData(MMGenObject):
|
||||
msgs = {
|
||||
'too_many_acct_addresses': f"""
|
||||
ERROR: More than one address found for account: '{{}}'.
|
||||
Your 'wallet.dat' file appears to have been altered by a non-{pnm} program.
|
||||
Please restore your tracking wallet from a backup or create a new one and
|
||||
re-import your addresses.
|
||||
""".strip()
|
||||
}
|
||||
|
||||
def __new__(cls,proto,*args,**kwargs):
|
||||
return MMGenObject.__new__(altcoin_subclass(cls,proto,'tw'))
|
||||
|
||||
def __init__(self,proto,*args,**kwargs):
|
||||
self.al_ids = {}
|
||||
self.proto = proto
|
||||
self.rpc = None
|
||||
|
||||
def seed_ids(self):
|
||||
return list(self.al_ids.keys())
|
||||
|
||||
def addrlist(self,al_id):
|
||||
# TODO: Validate al_id
|
||||
if al_id in self.al_ids:
|
||||
return self.al_ids[al_id]
|
||||
|
||||
def mmaddr2coinaddr(self,mmaddr):
|
||||
al_id,idx = MMGenID(self.proto,mmaddr).rsplit(':',1)
|
||||
coinaddr = ''
|
||||
if al_id in self.al_ids:
|
||||
coinaddr = self.addrlist(al_id).coinaddr(int(idx))
|
||||
return coinaddr or None
|
||||
|
||||
def coinaddr2mmaddr(self,coinaddr):
|
||||
d = self.make_reverse_dict([coinaddr])
|
||||
return (list(d.values())[0][0]) if d else None
|
||||
|
||||
def add(self,addrlist):
|
||||
if type(addrlist) == AddrList:
|
||||
self.al_ids[addrlist.al_id] = addrlist
|
||||
return True
|
||||
else:
|
||||
raise TypeError(f'Error: object {addrlist!r} is not of type AddrList')
|
||||
|
||||
def make_reverse_dict(self,coinaddrs):
|
||||
d = MMGenDict()
|
||||
for al_id in self.al_ids:
|
||||
d.update(self.al_ids[al_id].make_reverse_dict_addrlist(coinaddrs))
|
||||
return d
|
||||
|
||||
class TwAddrData(AddrData,metaclass=AsyncInit):
|
||||
|
||||
def __new__(cls,proto,*args,**kwargs):
|
||||
return MMGenObject.__new__(altcoin_subclass(cls,proto,'tw'))
|
||||
|
||||
async def __init__(self,proto,wallet=None):
|
||||
self.proto = proto
|
||||
from .rpc import rpc_init
|
||||
self.rpc = await rpc_init(proto)
|
||||
self.al_ids = {}
|
||||
await self.add_tw_data(wallet)
|
||||
|
||||
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))
|
||||
|
||||
async def add_tw_data(self,wallet):
|
||||
|
||||
twd = await self.get_tw_data(wallet)
|
||||
out,i = {},0
|
||||
for acct,addr_array in twd:
|
||||
l = get_obj(TwLabel,proto=self.proto,text=acct,silent=True)
|
||||
if l and l.mmid.type == 'mmgen':
|
||||
obj = l.mmid.obj
|
||||
if len(addr_array) != 1:
|
||||
die(2,self.msgs['too_many_acct_addresses'].format(acct))
|
||||
al_id = AddrListID(SeedID(sid=obj.sid),self.proto.addr_type(obj.mmtype))
|
||||
if al_id not in out:
|
||||
out[al_id] = []
|
||||
out[al_id].append(AddrListEntry(self.proto,idx=obj.idx,addr=addr_array[0],label=l.comment))
|
||||
i += 1
|
||||
|
||||
vmsg(f'{i} {pnm} addresses found, {len(twd)} accounts total')
|
||||
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))))
|
||||
|
|
|
|||
119
mmgen/addrdata.py
Executable file
119
mmgen/addrdata.py
Executable file
|
|
@ -0,0 +1,119 @@
|
|||
#!/usr/bin/env python3
|
||||
#
|
||||
# mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
|
||||
# Copyright (C)2013-2022 The MMGen Project <mmgen@tuta.io>
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
"""
|
||||
addrdata.py: MMGen AddrData and related classes
|
||||
"""
|
||||
|
||||
from .util import vmsg,altcoin_subclass
|
||||
from .base_obj import AsyncInit
|
||||
from .obj import MMGenObject,MMGenDict,get_obj,AddrListID,AddrListData
|
||||
from .addr import MMGenID,AddrListEntry,AddrList
|
||||
|
||||
class AddrData(MMGenObject):
|
||||
msgs = {
|
||||
'too_many_acct_addresses': """
|
||||
ERROR: More than one address found for account: '{}'.
|
||||
Your 'wallet.dat' file appears to have been altered by a non-{} program.
|
||||
Please restore your tracking wallet from a backup or create a new one and
|
||||
re-import your addresses.
|
||||
""".strip()
|
||||
}
|
||||
|
||||
def __new__(cls,proto,*args,**kwargs):
|
||||
return MMGenObject.__new__(altcoin_subclass(cls,proto,'tw'))
|
||||
|
||||
def __init__(self,proto,*args,**kwargs):
|
||||
self.al_ids = {}
|
||||
self.proto = proto
|
||||
self.rpc = None
|
||||
|
||||
def seed_ids(self):
|
||||
return list(self.al_ids.keys())
|
||||
|
||||
def addrlist(self,al_id):
|
||||
# TODO: Validate al_id
|
||||
if al_id in self.al_ids:
|
||||
return self.al_ids[al_id]
|
||||
|
||||
def mmaddr2coinaddr(self,mmaddr):
|
||||
al_id,idx = MMGenID(self.proto,mmaddr).rsplit(':',1)
|
||||
coinaddr = ''
|
||||
if al_id in self.al_ids:
|
||||
coinaddr = self.addrlist(al_id).coinaddr(int(idx))
|
||||
return coinaddr or None
|
||||
|
||||
def coinaddr2mmaddr(self,coinaddr):
|
||||
d = self.make_reverse_dict([coinaddr])
|
||||
return (list(d.values())[0][0]) if d else None
|
||||
|
||||
def add(self,addrlist):
|
||||
if type(addrlist) == AddrList:
|
||||
self.al_ids[addrlist.al_id] = addrlist
|
||||
return True
|
||||
else:
|
||||
raise TypeError(f'Error: object {addrlist!r} is not of type AddrList')
|
||||
|
||||
def make_reverse_dict(self,coinaddrs):
|
||||
d = MMGenDict()
|
||||
for al_id in self.al_ids:
|
||||
d.update(self.al_ids[al_id].make_reverse_dict_addrlist(coinaddrs))
|
||||
return d
|
||||
|
||||
class TwAddrData(AddrData,metaclass=AsyncInit):
|
||||
|
||||
def __new__(cls,proto,*args,**kwargs):
|
||||
return MMGenObject.__new__(altcoin_subclass(cls,proto,'tw'))
|
||||
|
||||
async def __init__(self,proto,wallet=None):
|
||||
from .rpc import rpc_init
|
||||
from .obj import TwLabel
|
||||
from .globalvars import g
|
||||
from .seed import SeedID
|
||||
self.proto = proto
|
||||
self.rpc = await rpc_init(proto)
|
||||
self.al_ids = {}
|
||||
twd = await self.get_tw_data(wallet)
|
||||
out,i = {},0
|
||||
for acct,addr_array in twd:
|
||||
l = get_obj(TwLabel,proto=self.proto,text=acct,silent=True)
|
||||
if l and l.mmid.type == 'mmgen':
|
||||
obj = l.mmid.obj
|
||||
if len(addr_array) != 1:
|
||||
die(2,self.msgs['too_many_acct_addresses'].format(acct,g.prog_name))
|
||||
al_id = AddrListID(SeedID(sid=obj.sid),self.proto.addr_type(obj.mmtype))
|
||||
if al_id not in out:
|
||||
out[al_id] = []
|
||||
out[al_id].append(AddrListEntry(self.proto,idx=obj.idx,addr=addr_array[0],label=l.comment))
|
||||
i += 1
|
||||
|
||||
vmsg(f'{i} {g.prog_name} addresses found, {len(twd)} accounts total')
|
||||
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))
|
||||
|
|
@ -23,7 +23,7 @@ altcoins.eth.tw: Ethereum tracking wallet and related classes for the MMGen suit
|
|||
from mmgen.common import *
|
||||
from mmgen.obj import TwLabel,is_coin_addr,is_mmgen_id,ListItemAttr,ImmutableAttr
|
||||
from mmgen.tw import TrackingWallet,TwAddrList,TwUnspentOutputs,TwGetBalance
|
||||
from mmgen.addr import AddrData,TwAddrData
|
||||
from mmgen.addrdata import AddrData,TwAddrData
|
||||
from .contract import Token,TokenResolve
|
||||
from .obj import ETHAmt
|
||||
|
||||
|
|
|
|||
|
|
@ -938,7 +938,7 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
|
|||
mmaddr = TwMMGenID(self.proto,arg1)
|
||||
|
||||
if mmaddr and not coinaddr:
|
||||
from .addr import TwAddrData
|
||||
from .addrdata import TwAddrData
|
||||
coinaddr = (await TwAddrData(self.proto)).mmaddr2coinaddr(mmaddr)
|
||||
|
||||
try:
|
||||
|
|
@ -953,7 +953,7 @@ class TrackingWallet(MMGenObject,metaclass=AsyncInit):
|
|||
# Allow for the possibility that BTC addr of MMGen addr was entered.
|
||||
# Do reverse lookup, so that MMGen addr will not be marked as non-MMGen.
|
||||
if not mmaddr:
|
||||
from .addr import TwAddrData
|
||||
from .addrdata import TwAddrData
|
||||
mmaddr = (await TwAddrData(proto=self.proto)).coinaddr2mmaddr(coinaddr)
|
||||
|
||||
if not mmaddr:
|
||||
|
|
|
|||
|
|
@ -723,7 +723,8 @@ class MMGenTX:
|
|||
die(2,'At least one output must be specified on the command line')
|
||||
|
||||
async def get_outputs_from_cmdline(self,cmd_args):
|
||||
from .addr import AddrList,AddrData,TwAddrData
|
||||
from .addr import AddrList
|
||||
from .addrdata import AddrData,TwAddrData
|
||||
from .addrfile import AddrFile
|
||||
addrfiles = remove_dups(
|
||||
tuple(a for a in cmd_args if get_extension(a) == AddrFile.ext),
|
||||
|
|
@ -1612,7 +1613,7 @@ class MMGenTX:
|
|||
#
|
||||
# async def get_outputs_from_cmdline(self,mmid): # TODO: check that addr is empty
|
||||
#
|
||||
# from .addr import TwAddrData
|
||||
# from .addrdata import TwAddrData
|
||||
# ad_w = await TwAddrData()
|
||||
#
|
||||
# if is_mmgen_id(self.proto,mmid):
|
||||
|
|
|
|||
|
|
@ -397,8 +397,9 @@ class TestSuiteMain(TestSuiteBase,TestSuiteShared):
|
|||
return out
|
||||
|
||||
def _create_tx_data(self,sources,addrs_per_wallet=addrs_per_wallet):
|
||||
from mmgen.addr import AddrData,AddrList
|
||||
from mmgen.addr import AddrList
|
||||
from mmgen.obj import AddrIdxList
|
||||
from mmgen.addrdata import AddrData
|
||||
tx_data,ad = {},AddrData(self.proto)
|
||||
for s in sources:
|
||||
afile = get_file_with_ext(self.cfgs[s]['tmpdir'],'addrs')
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue