cfg.py 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  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. swap.cfg: swap configuration class for the MMGen Wallet suite
  12. """
  13. import re
  14. from collections import namedtuple
  15. from ..amt import UniAmt
  16. from ..util import die
  17. _mmd = namedtuple('swap_config_option', ['min', 'max', 'dfl'])
  18. class SwapCfg:
  19. # The trade limit, i.e., set 100000000 to get a minimum of 1 full asset, else a refund
  20. # Optional. 1e8 or scientific notation
  21. trade_limit = None
  22. # Swap interval for streaming swap in blocks. Optional. If 0, do not stream
  23. si = _mmd(1, 20, 3) # stream_interval
  24. # Swap quantity for streaming swap.
  25. # The interval value determines the frequency of swaps in blocks
  26. # Optional. If 0, network will determine the number of swaps
  27. stream_quantity = 0
  28. def __init__(self, cfg):
  29. self.cfg = cfg
  30. if cfg.trade_limit is not None:
  31. self.set_trade_limit(desc='parameter for --trade-limit')
  32. if cfg.stream_interval is None:
  33. self.stream_interval = self.si.dfl
  34. else:
  35. self.set_stream_interval(desc='parameter for --stream-interval')
  36. def set_trade_limit(self, *, desc):
  37. s = self.cfg.trade_limit
  38. if re.match(r'-*[0-9]+(\.[0-9]+)*%*$', s):
  39. self.trade_limit = 1 - float(s[:-1]) / 100 if s.endswith('%') else UniAmt(s)
  40. else:
  41. die('SwapCfgValueError', f'{s}: invalid {desc}')
  42. def set_stream_interval(self, *, desc):
  43. s = self.cfg.stream_interval
  44. from ..util import is_int
  45. if not is_int(s):
  46. die('SwapCfgValueError', f'{s}: invalid {desc} (not an integer)')
  47. self.stream_interval = si = int(s)
  48. if si < self.si.min:
  49. die('SwapCfgValueError', f'{si}: invalid {desc} (< {self.si.min})')
  50. if si > self.si.max:
  51. die('SwapCfgValueError', f'{si}: invalid {desc} (> {self.si.max})')