mmgen-wallet/mmgen/proto/btc/tw/rpc.py

#!/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

"""
proto.btc.tw.rpc: Bitcoin base protocol tracking wallet RPC classes
"""

from ....addr import CoinAddr
from ....util import die,msg,rmsg
from ....obj import MMGenList
from ....tw.shared import get_tw_label

class BitcoinTwRPC:

	async def get_addr_label_pairs(self,twmmid=None):
		"""
		Get all the accounts in the tracking wallet and their associated addresses.
		Returns list of (label,address) tuples.
		"""
		def check_dup_mmid(acct_labels):
			mmid_prev,err = None,False
			for mmid in sorted(label.mmid for label in acct_labels if label):
				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!')

		async def get_acct_list():
			if 'label_api' in self.rpc.caps:
				return await self.rpc.call('listlabels')
			else:
				return (await self.rpc.call('listaccounts',0,True)).keys()

		async def get_acct_addrs(acct_list):
			if 'label_api' in self.rpc.caps:
				return [list(a.keys())
					for a in await self.rpc.batch_call('getaddressesbylabel',[(k,) for k in acct_list])]
			else:
				return await self.rpc.batch_call('getaddressesbyaccount',[(a,) for a in acct_list])

		acct_labels = [get_tw_label(self.proto,a) for a in await get_acct_list()]

		if twmmid:
			acct_labels = [lbl for lbl in acct_labels if lbl.mmid == twmmid]

		if not acct_labels:
			return None

		check_dup_mmid(acct_labels)

		acct_addrs = await get_acct_addrs(acct_labels)

		for n,a in enumerate(acct_addrs):
			if len(a) != 1:
				raise ValueError(f'{a}: label {acct_labels[n]!r} has != 1 associated address!')

		return [(
			label,
			CoinAddr(self.proto,addrs[0])
		) for label,addrs in zip(acct_labels,acct_addrs)]

	async def get_unspent_by_mmid(self,minconf=1,mmid_filter=[]):
		"""
		get unspent outputs in tracking wallet, compute balances per address
		and return a dict with elements { 'twmmid': {'addr','lbl','amt'} }
		"""
		data = {}
		lbl_id = ('account','label')['label_api' in self.rpc.caps]
		amt0 = self.proto.coin_amt('0')

		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(self.proto,d[lbl_id])

			if label:
				lm = label.mmid
				if mmid_filter and (lm not in mmid_filter):
					continue
				if lm in data:
					if data[lm]['addr'] != d['address']:
						die(2,'duplicate {} address ({}) for this MMGen address! ({})'.format(
							self.proto.coin,
							d['address'],
							data[lm]['addr'] ))
				else:
					lm.confs = d['confirmations']
					lm.txid = d['txid']
					lm.vout = d['vout']
					lm.date = None
					data[lm] = {
						'amt': amt0,
						'lbl': label,
						'addr': CoinAddr(self.proto,d['address']) }
				amt = self.proto.coin_amt(d['amount'])
				data[lm]['amt'] += amt

		return data