new BaseSwapMethods class

mmgen/proto/btc/tx/new_swap.py

@@ -21,11 +21,3 @@ class NewSwap(New, TxNewSwap):
 		memo = self.update_memo(o['data'].decode(), trade_limit)
 		o['data'] = f'data:{memo}'
 		self.data_output = self.Output(self.proto, **o)
-
-	@property
-	def vault_idx(self):
-		return self._chg_output_ops('idx', 'is_vault')
-
-	@property
-	def vault_output(self):
-		return self._chg_output_ops('output', 'is_vault')

mmgen/proto/vm/tx/new_swap.py

@@ -15,11 +15,3 @@ class VmNewSwap:
 	def update_data_output(self, trade_limit):
 		self.swap_memo = str(self.update_memo(self.swap_memo, trade_limit))
 		self.set_gas_with_data(self.swap_memo.encode())
-
-	@property
-	def vault_idx(self):
-		return 0
-
-	@property
-	def vault_output(self):
-		return self.outputs[0]

mmgen/tx/base.py

@@ -24,7 +24,7 @@ from ..obj import (
 )
 from ..amt import CoinAmtChk
 from ..addr import MMGenID, CoinAddr
-from ..util import msg, ymsg, fmt, remove_dups, make_timestamp, die, cached_property
+from ..util import msg, ymsg, fmt, remove_dups, make_timestamp, die
 
 class MMGenTxIO(MMGenListItem):
 	vout     = ListItemAttr(NonNegativeInt)
@@ -68,6 +68,7 @@ class MMGenTxIOList(list, MMGenObject):
 
 class Base(MMGenObject):
 	desc         = 'transaction'
+	data_desc    = 'transaction data'
 	comment      = None
 	txid         = None
 	coin_txid    = None
@@ -126,14 +127,6 @@ class Base(MMGenObject):
 		self.twctl    = twctl
 		self.is_token = 'Token' in self.name
 
-	@property
-	def data_desc(self):
-		def gen():
-			if self.is_swap:
-				yield 'swap'
-			yield 'transaction data'
-		return ' '.join(gen())
-
 	@property
 	def coin(self):
 		return self.proto.coin
@@ -228,46 +221,6 @@ class Base(MMGenObject):
 					from ..ui import keypress_confirm
 					keypress_confirm(self.cfg, 'Continue?', default_yes=True, do_exit=True)
 
-	# swap methods:
-	@cached_property
-	def swap_proto_rpc(self):
-		from ..swap.util import get_swap_proto_rpc
-		return get_swap_proto_rpc(self.cfg, self.swap_proto_mod.coin)
-
-	@cached_property
-	def swap_proto_mod(self):
-		from ..swap.util import get_swap_proto_mod
-		return get_swap_proto_mod(self.swap_proto)
-
-	@cached_property
-	def send_asset(self):
-		spec = self.proto.coin + (f'.{self.proto.tokensym}' if self.proto.tokensym else '')
-		return self.swap_proto_mod.SwapAsset(spec, 'send')
-
-	@cached_property
-	def recv_asset(self):
-		if hasattr(self, 'swap_recv_asset_spec'):
-			return self.swap_proto_mod.SwapAsset(self.swap_recv_asset_spec, 'recv')
-		elif self.is_swap and hasattr(self, 'data_output'):
-			from ..swap.proto.thorchain.memo import THORChainMemo
-			return THORChainMemo.parse(str(self.data_output.data)).asset
-		else: # backwards-compatibility workaround
-			from ..swap.asset import SwapAsset
-			x = '[unknown]'
-			return SwapAsset._ad(x, x, x, x, x, x)
-
-	def check_swap_network(self):
-		rpc = self.swap_proto_rpc
-		data = rpc.inbound_addrs()
-		for asset in (self.send_asset, self.recv_asset):
-			if asset.name != 'RUNE':
-				for key, state in (
-						('global_trading_paused', 'paused'),
-						('chain_trading_paused',  f'paused for asset {asset.name}'),
-						('halted',                f'halted for asset {asset.name}')):
-					if rpc.data_for_coin(data, asset.coin)[key]:
-						die(2, f'Trading is {state} on the {self.swap_proto_mod.name} network')
-
 	# token methods:
 	@property
 	def token_op(self):

mmgen/tx/base_swap.py

@@ -0,0 +1,63 @@
+# MMGen Wallet, a terminal-based cryptocurrency wallet
+# Copyright (C)2013-2026 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-wallet
+#   https://gitlab.com/mmgen/mmgen-wallet
+
+"""
+tx.base_swap: base swap transaction mixin class
+"""
+
+from ..util import die, cached_property
+
+class BaseSwapMethods:
+	data_desc = 'swap transaction data'
+
+	@property
+	def vault_idx(self):
+		return self._chg_output_ops('idx', 'is_vault')
+
+	@property
+	def vault_output(self):
+		return self._chg_output_ops('output', 'is_vault')
+
+	@cached_property
+	def swap_proto_rpc(self):
+		from ..swap.util import get_swap_proto_rpc
+		return get_swap_proto_rpc(self.cfg, self.swap_proto_mod.coin)
+
+	@cached_property
+	def swap_proto_mod(self):
+		from ..swap.util import get_swap_proto_mod
+		return get_swap_proto_mod(self.swap_proto)
+
+	@cached_property
+	def send_asset(self):
+		spec = self.proto.coin + (f'.{self.proto.tokensym}' if self.proto.tokensym else '')
+		return self.swap_proto_mod.SwapAsset(spec, 'send')
+
+	@cached_property
+	def recv_asset(self):
+		if hasattr(self, 'swap_recv_asset_spec'):
+			return self.swap_proto_mod.SwapAsset(self.swap_recv_asset_spec, 'recv')
+		elif self.is_swap and hasattr(self, 'data_output'):
+			from ..swap.proto.thorchain.memo import THORChainMemo
+			return THORChainMemo.parse(str(self.data_output.data)).asset
+		else: # backwards-compatibility workaround
+			from ..swap.asset import SwapAsset
+			x = '[unknown]'
+			return SwapAsset._ad(x, x, x, x, x, x)
+
+	def check_swap_network(self):
+		rpc = self.swap_proto_rpc
+		data = rpc.inbound_addrs()
+		for asset in (self.send_asset, self.recv_asset):
+			if asset.name != 'RUNE':
+				for key, state in (
+						('global_trading_paused', 'paused'),
+						('chain_trading_paused',  f'paused for asset {asset.name}'),
+						('halted',                f'halted for asset {asset.name}')):
+					if rpc.data_for_coin(data, asset.coin)[key]:
+						die(2, f'Trading is {state} on the {self.swap_proto_mod.name} network')

mmgen/tx/completed.py

@@ -49,6 +49,12 @@ class Completed(Base):
 
 			self.infile = filename
 
+		if self.is_swap:
+			from .base_swap import BaseSwapMethods as m
+			for k in m.__dict__:
+				if not k.startswith('_'):
+					setattr(type(self), k, getattr(m, k))
+
 	@property
 	def file(self):
 		from .file import MMGenTxFile

mmgen/tx/new_swap.py

@@ -17,6 +17,7 @@ from ..amt import UniAmt
 from ..util import msg, ymsg, die
 from ..swap.util import get_swap_asset, init_swap_proto
 
+from .base_swap import BaseSwapMethods
 from .new import New
 
 def get_send_proto(cfg):
@@ -26,7 +27,7 @@ def get_send_proto(cfg):
 		cfg._usage()
 	return init_swap_proto(cfg, get_swap_asset(cfg, arg, direction='send'))
 
-class NewSwap(New):
+class NewSwap(BaseSwapMethods, New):
 	desc = 'swap transaction'
 	swap_quote_refresh_timeout = 30
 	_output = namedtuple('swap_output', ['coin', 'network', 'addr', 'mmid'])