rpc.py 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. #!/usr/bin/env python3
  2. #
  3. # MMGen Wallet, a terminal-based cryptocurrency wallet
  4. # Copyright (C)2013-2025 The MMGen Project <mmgen@tuta.io>
  5. # Licensed under the GNU General Public License, Version 3:
  6. # https://www.gnu.org/licenses
  7. # Public project repositories:
  8. # https://github.com/mmgen/mmgen-wallet
  9. # https://gitlab.com/mmgen/mmgen-wallet
  10. """
  11. proto.btc.tw.rpc: Bitcoin base protocol tracking wallet RPC classes
  12. """
  13. from ....addr import CoinAddr
  14. from ....util import die, msg
  15. from ....tw.shared import get_tw_label
  16. from ....tw.rpc import TwRPC
  17. from ....tw.ctl import label_addr_pair
  18. class BitcoinTwRPC(TwRPC):
  19. async def get_label_addr_pairs(self):
  20. """
  21. Get all the accounts in the tracking wallet and their associated addresses.
  22. Returns list of (label, address) tuples.
  23. """
  24. def check_dup_mmid(acct_labels):
  25. mmid_prev, err = None, False
  26. for mmid in sorted(label.mmid for label in acct_labels if label):
  27. if mmid == mmid_prev:
  28. err = True
  29. msg(f'Duplicate MMGen ID ({mmid}) discovered in tracking wallet!\n')
  30. mmid_prev = mmid
  31. if err:
  32. die(4, 'Tracking wallet is corrupted!')
  33. async def get_acct_list():
  34. if 'label_api' in self.rpc.caps:
  35. return await self.rpc.call('listlabels')
  36. else:
  37. return (await self.rpc.call('listaccounts', 0, True)).keys()
  38. async def get_acct_addrs(acct_list):
  39. if 'label_api' in self.rpc.caps:
  40. return [list(a.keys())
  41. for a in await self.rpc.batch_call('getaddressesbylabel', [(k,) for k in acct_list])]
  42. else:
  43. return await self.rpc.batch_call('getaddressesbyaccount', [(a,) for a in acct_list])
  44. acct_labels = [get_tw_label(self.proto, a) for a in await get_acct_list()]
  45. if not acct_labels:
  46. return []
  47. check_dup_mmid(acct_labels)
  48. acct_addrs = await get_acct_addrs(acct_labels)
  49. for n, a in enumerate(acct_addrs):
  50. if len(a) != 1:
  51. raise ValueError(f'{a}: label {acct_labels[n]!r} has != 1 associated address!')
  52. return [label_addr_pair(label, CoinAddr(self.proto, addrs[0]))
  53. for label, addrs in zip(acct_labels, acct_addrs)]
  54. async def get_unspent_by_mmid(self, minconf=1, mmid_filter=[]):
  55. """
  56. get unspent outputs in tracking wallet, compute balances per address
  57. and return a dict with elements {'twmmid': {'addr', 'lbl', 'amt'}}
  58. """
  59. data = {}
  60. lbl_id = ('account', 'label')['label_api' in self.rpc.caps]
  61. amt0 = self.proto.coin_amt('0')
  62. for d in await self.rpc.call('listunspent', 0):
  63. if not lbl_id in d:
  64. continue # skip coinbase outputs with missing account
  65. if d['confirmations'] < minconf:
  66. continue
  67. label = get_tw_label(self.proto, d[lbl_id])
  68. if label:
  69. lm = label.mmid
  70. if mmid_filter and (lm not in mmid_filter):
  71. continue
  72. if lm in data:
  73. if data[lm]['addr'] != d['address']:
  74. die(2, 'duplicate {} address ({}) for this MMGen address! ({})'.format(
  75. self.proto.coin,
  76. d['address'],
  77. data[lm]['addr']))
  78. else:
  79. lm.confs = d['confirmations']
  80. lm.txid = d['txid']
  81. lm.vout = d['vout']
  82. lm.date = None
  83. data[lm] = {
  84. 'amt': amt0,
  85. 'lbl': label,
  86. 'addr': CoinAddr(self.proto, d['address'])}
  87. data[lm]['amt'] += self.proto.coin_amt(d['amount'])
  88. return data