term.py: new MMGenTerm family of classes

Testing:

    $ test/misc/term.py

mmgen/crypto.py

@@ -162,7 +162,7 @@ def _get_random_data_from_user(uchars,desc):
 	key_data,time_data = '',[]
 
 	for i in range(uchars):
-		key_data += get_char_raw('\r'+prompt.format(uchars-i)).decode()
+		key_data += get_char_raw('\r'+prompt.format(uchars-i))
 		time_data.append(time.time())
 
 	if opt.quiet: msg_r('\r')

mmgen/mn_entry.py

@@ -52,7 +52,7 @@ class MnEntryMode(object):
 	def get_char(self,s):
 		did_erase = False
 		while True:
-			ch = get_char_raw('',num_chars=1).decode()
+			ch = get_char_raw('',num_chars=1)
 			if s and ch in _erase_chars:
 				s = s[:-1]
 				did_erase = True
@@ -320,7 +320,7 @@ class MnemonicEntry(object):
 				fmt(mode.choose_info,' '*14).lstrip().format(usl=self.uniq_ss_len),
 			))
 		while True:
-			uret = get_char('Entry mode: ').decode()
+			uret = get_char('Entry mode: ')
 			if uret in [str(i) for i in range(1,len(em_objs)+1)]:
 				return em_objs[int(uret)-1]
 			else:

mmgen/opts.py

@@ -78,8 +78,8 @@ def opt_postproc_debug():
 	Msg('\n=== end opts.py debug ===\n')
 
 def init_term_and_color():
-	from mmgen.term import set_terminal_vars
-	set_terminal_vars()
+	from mmgen.term import init_term
+	init_term()
 
 	if g.color: # MMGEN_DISABLE_COLOR sets this to False
 		from mmgen.color import start_mscolor,init_color

mmgen/seed.py

@@ -699,7 +699,7 @@ class SeedSourceUnenc(SeedSource):
 		def choose_len():
 			prompt = self.choose_seedlen_prompt
 			while True:
-				r = get_char('\r'+prompt).decode()
+				r = get_char('\r'+prompt)
 				if is_int(r) and 1 <= int(r) <= len(ok_lens):
 					break
 			msg_r(('\r','\n')[g.test_suite] + ' '*len(prompt) + '\r')
@@ -1044,7 +1044,7 @@ class DieRollSeedFile(SeedSourceUnenc):
 			p = prompt_fs
 			sleep = g.short_disp_timeout
 			while True:
-				ch = get_char(p.format(n),num_chars=1,sleep=sleep).decode()
+				ch = get_char(p.format(n),num_chars=1,sleep=sleep)
 				if ch in b6d_digits:
 					msg_r(CUR_HIDE + ' OK')
 					return ch

mmgen/term.py

@@ -17,10 +17,11 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 """
-term.py:  Terminal-handling routines for the MMGen suite
+term.py:  Terminal classes for the MMGen suite
 """
 
-import os,struct
+import sys,os,time
+from collections import namedtuple
 from mmgen.common import *
 
 try:
@@ -29,176 +30,211 @@ try:
 	_platform = 'linux'
 except:
 	try:
-		import msvcrt,time
-		_platform = 'win'
+		import msvcrt
+		_platform = 'mswin'
 	except:
 		die(2,'Unable to set terminal mode')
 	if not sys.stdin.isatty():
 		msvcrt.setmode(sys.stdin.fileno(),os.O_BINARY)
 
-def _kb_hold_protect_unix():
+class MMGenTerm(object):
 
-	if g.test_suite: return
+	tdim = namedtuple('terminal_dimensions',['width','height'])
 
-	fd = sys.stdin.fileno()
-	old = termios.tcgetattr(fd)
-	tty.setcbreak(fd)
-
-	timeout = float(0.3)
-
-	while True:
-		key = select([sys.stdin], [], [], timeout)[0]
-		if key: sys.stdin.read(1)
-		else:
-			termios.tcsetattr(fd, termios.TCSADRAIN, old)
-			break
-
-# Use os.read(), not file.read(), to get a variable number of bytes without blocking.
-# Request 5 bytes to cover escape sequences generated by F1, F2, .. Fn keys (5 bytes)
-# as well as UTF8 chars (4 bytes max).
-def _get_keypress_unix(prompt='',immed_chars='',prehold_protect=True,num_chars=5,sleep=None):
-	timeout = float(0.3)
-	fd = sys.stdin.fileno()
-	old = termios.tcgetattr(fd)
-	tty.setcbreak(fd)
-	if sleep:
-		time.sleep(sleep)
-	msg_r(prompt)
-	immed_chars = immed_chars.encode()
-	if g.test_suite: prehold_protect = False
-	while True:
-		# Protect against held-down key before read()
-		key = select([sys.stdin], [], [], timeout)[0]
-		s = os.read(fd,num_chars)
-		if prehold_protect:
-			if key: continue
-		if immed_chars == 'ALL' or s in immed_chars: break
-		if immed_chars == 'ALL_EXCEPT_ENTER' and not s in '\n\r': break
-		# Protect against long keypress
-		key = select([sys.stdin], [], [], timeout)[0]
-		if not key: break
-	termios.tcsetattr(fd, termios.TCSADRAIN, old)
-	return s
-
-def _get_keypress_unix_raw(prompt='',immed_chars='',prehold_protect=None,num_chars=5,sleep=None):
-	fd = sys.stdin.fileno()
-	old = termios.tcgetattr(fd)
-	tty.setcbreak(fd)
-	if sleep:
-		time.sleep(sleep)
-	msg_r(prompt)
-	ch = os.read(fd,num_chars)
-	termios.tcsetattr(fd, termios.TCSADRAIN, old)
-	return ch
-
-def _get_keypress_unix_stub(prompt='',immed_chars='',prehold_protect=None,num_chars=None,sleep=None):
-	if sleep:
-		time.sleep(0.1)
-	msg_r(prompt)
-	return sys.stdin.read(1).encode()
-
-#_get_keypress_unix_stub = _get_keypress_unix
-
-def _kb_hold_protect_mswin():
-
-	timeout = float(0.5)
-
-	while True:
-		hit_time = time.time()
-		while True:
-			if msvcrt.kbhit():
-				msvcrt.getch()
-				break
-			if float(time.time() - hit_time) > timeout:
-				return
-
-def _get_keypress_mswin(prompt='',immed_chars='',prehold_protect=True,num_chars=None,sleep=None):
-
-	if sleep:
-		time.sleep(sleep)
-
-	msg_r(prompt)
-	timeout = float(0.5)
-
-	while True:
-		if msvcrt.kbhit():
-			ch = msvcrt.getch()
-
-			if ord(ch) == 3: raise KeyboardInterrupt
-
-			if immed_chars == 'ALL' or ch.decode() in immed_chars:
-				return ch
-			if immed_chars == 'ALL_EXCEPT_ENTER' and not ch in '\n\r':
-				return ch
-
-			hit_time = time.time()
-
-			while True:
-				if msvcrt.kbhit(): break
-				if float(time.time() - hit_time) > timeout:
-					return ch
-
-def _get_keypress_mswin_raw(prompt='',immed_chars='',prehold_protect=None,num_chars=None,sleep=None):
-	if sleep:
-		time.sleep(sleep)
-	msg_r(prompt)
-	ch = msvcrt.getch()
-	if ch == b'\x03': raise KeyboardInterrupt
-	return ch
-
-def _get_keypress_mswin_stub(prompt='',immed_chars='',prehold_protect=None,num_chars=None,sleep=None):
-	if sleep:
-		time.sleep(0.1)
-	msg_r(prompt)
-	return os.read(0,1)
-
-def _get_terminal_size_linux():
-	try:
-		return tuple(os.get_terminal_size())
-	except:
-		try:
-			return (os.environ['LINES'],os.environ['COLUMNS'])
-		except:
-			return (80,25)
-
-def _get_terminal_size_mswin():
-	import sys,os,struct
-	x,y = 0,0
-	try:
-		from ctypes import windll,create_string_buffer
-		# handles - stdin: -10, stdout: -11, stderr: -12
-		csbi = create_string_buffer(22)
-		h = windll.kernel32.GetStdHandle(-12)
-		res = windll.kernel32.GetConsoleScreenBufferInfo(h, csbi)
-		if res:
-			(bufx, bufy, curx, cury, wattr, left, top, right, bottom,
-			maxx, maxy) = struct.unpack('hhhhHhhhhhh', csbi.raw)
-			x = right - left + 1
-			y = bottom - top + 1
-	except:
+	@classmethod
+	def init(cls):
 		pass
 
-	if x and y:
-		return x, y
-	else:
-		msg(yellow('Warning: could not get terminal size. Using fallback dimensions.'))
-		return 80,25
+	@classmethod
+	def kb_hold_protect(cls):
+		return None
+
+class MMGenTermLinux(MMGenTerm):
+
+	@classmethod
+	def init(cls):
+		cls.stdin_fd = sys.stdin.fileno()
+		cls.old_term = termios.tcgetattr(cls.stdin_fd)
+
+	@classmethod
+	def get_terminal_size(cls):
+		try:
+			ret = os.get_terminal_size()
+		except:
+			try:
+				ret = (os.environ['COLUMNS'],os.environ['LINES'])
+			except:
+				ret = (80,25)
+		return cls.tdim(*ret)
+
+	@classmethod
+	def kb_hold_protect(cls):
+		if g.test_suite:
+			return
+		tty.setcbreak(cls.stdin_fd)
+		timeout = 0.3
+		while True:
+			key = select([sys.stdin], [], [], timeout)[0]
+			if key:
+				sys.stdin.read(1)
+			else:
+				termios.tcsetattr(cls.stdin_fd, termios.TCSADRAIN, cls.old_term)
+				break
+
+	@classmethod
+	def get_char(cls,prompt='',immed_chars='',prehold_protect=True,num_chars=5,sleep=None):
+		"""
+		Use os.read(), not file.read(), to get a variable number of bytes without blocking.
+		Request 5 bytes to cover escape sequences generated by F1, F2, .. Fn keys (5 bytes)
+		as well as UTF8 chars (4 bytes max).
+		"""
+		timeout = 0.3
+		tty.setcbreak(cls.stdin_fd)
+		if sleep:
+			time.sleep(sleep)
+		msg_r(prompt)
+		if g.test_suite:
+			prehold_protect = False
+		while True:
+			# Protect against held-down key before read()
+			key = select([sys.stdin], [], [], timeout)[0]
+			s = os.read(cls.stdin_fd,num_chars).decode()
+			if prehold_protect and key:
+				continue
+			if s in immed_chars:
+				break
+			# Protect against long keypress
+			key = select([sys.stdin], [], [], timeout)[0]
+			if not key:
+				break
+		termios.tcsetattr(cls.stdin_fd, termios.TCSADRAIN, cls.old_term)
+		return s
+
+	@classmethod
+	def get_char_raw(cls,prompt='',num_chars=5,sleep=None):
+		tty.setcbreak(cls.stdin_fd)
+		if sleep:
+			time.sleep(sleep)
+		msg_r(prompt)
+		s = os.read(cls.stdin_fd,num_chars).decode()
+		termios.tcsetattr(cls.stdin_fd, termios.TCSADRAIN, cls.old_term)
+		return s
+
+class MMGenTermLinuxStub(MMGenTermLinux):
+
+	@classmethod
+	def init(cls):
+		pass
+
+	@classmethod
+	def get_char(cls,prompt='',immed_chars='',prehold_protect=None,num_chars=None,sleep=None):
+		if sleep:
+			time.sleep(0.1)
+		msg_r(prompt)
+		return sys.stdin.read(1)
+
+	get_char_raw = get_char
+
+class MMGenTermMSWin(MMGenTerm):
+
+	@classmethod
+	def get_terminal_size(cls):
+		import struct
+		x,y = 0,0
+		try:
+			from ctypes import windll,create_string_buffer
+			# handles - stdin: -10, stdout: -11, stderr: -12
+			csbi = create_string_buffer(22)
+			h = windll.kernel32.GetStdHandle(-12)
+			res = windll.kernel32.GetConsoleScreenBufferInfo(h,csbi)
+			assert res, 'failed to get console screen buffer info'
+			left,top,right,bottom = struct.unpack('hhhhHhhhhhh', csbi.raw)[5:9]
+			x = right - left + 1
+			y = bottom - top + 1
+		except:
+			pass
+
+		if x and y:
+			return cls.tdim(x,y)
+		else:
+			msg(yellow('Warning: could not get terminal size. Using fallback dimensions.'))
+			return cls.tdim(80,25)
+
+	@classmethod
+	def kb_hold_protect(cls):
+		timeout = 0.5
+		while True:
+			hit_time = time.time()
+			while True:
+				if msvcrt.kbhit():
+					msvcrt.getch()
+					break
+				if time.time() - hit_time > timeout:
+					return
+
+	@classmethod
+	def get_char(cls,prompt='',immed_chars='',prehold_protect=True,num_chars=None,sleep=None):
+		"""
+		always return a single character, ignore num_chars
+		first character of 2-character sequence returned by F1-F12 keys is discarded
+		prehold_protect is ignored
+		"""
+		if sleep:
+			time.sleep(sleep)
+		msg_r(prompt)
+		timeout = 0.5
+		while True:
+			if msvcrt.kbhit():
+				ch = chr(msvcrt.getch()[0])
+				if ch == '\x03':
+					raise KeyboardInterrupt
+				if ch in immed_chars:
+					return ch
+				hit_time = time.time()
+				while True:
+					if msvcrt.kbhit():
+						break
+					if time.time() - hit_time > timeout:
+						return ch
+
+	@classmethod
+	def get_char_raw(cls,prompt='',num_chars=None,sleep=None):
+		"""
+		always return a single character, ignore num_chars
+		first character of 2-character sequence returned by F1-F12 keys is discarded
+		"""
+		while True:
+			if sleep:
+				time.sleep(sleep)
+			msg_r(prompt)
+			ch = chr(msvcrt.getch()[0])
+			if ch in '\x00\xe0': # first char of 2-char sequence for F1-F12 keys
+				continue
+			if ch == '\x03':
+				raise KeyboardInterrupt
+			return ch
+
+class MMGenTermMSWinStub(MMGenTermMSWin):
+
+	@classmethod
+	def get_char(cls,prompt='',immed_chars='',prehold_protect=None,num_chars=None,sleep=None):
+		if sleep:
+			time.sleep(0.1)
+		msg_r(prompt)
+		return os.read(0,1).decode()
+
+	get_char_raw = get_char
+
+def init_term():
+
+	term = {
+		'linux': (MMGenTermLinux if sys.stdin.isatty() else MMGenTermLinuxStub),
+		'mswin': (MMGenTermMSWin if sys.stdin.isatty() else MMGenTermMSWinStub),
+	}[_platform]
+
+	term.init()
 
-def set_terminal_vars():
 	global get_char,get_char_raw,kb_hold_protect,get_terminal_size
-	if _platform == 'linux':
-		get_char        = _get_keypress_unix
-		get_char_raw    = _get_keypress_unix_raw
-		kb_hold_protect = _kb_hold_protect_unix
-		if not sys.stdin.isatty():
-			get_char = get_char_raw = _get_keypress_unix_stub
-			kb_hold_protect = lambda: None
-		get_terminal_size = _get_terminal_size_linux
-	else:
-		get_char        = _get_keypress_mswin
-		get_char_raw    = _get_keypress_mswin_raw
-		kb_hold_protect = _kb_hold_protect_mswin
-		if not sys.stdin.isatty():
-			get_char = get_char_raw = _get_keypress_mswin_stub
-			kb_hold_protect = lambda: None
-		get_terminal_size = _get_terminal_size_mswin
+
+	for var in ('get_char','get_char_raw','kb_hold_protect','get_terminal_size'):
+		globals()[var] = getattr(term,var)

mmgen/tw.py

@@ -187,7 +187,7 @@ watch-only wallet using '{}-addrimport' and then re-run this program.
 	def set_term_columns(self):
 		from mmgen.term import get_terminal_size
 		while True:
-			self.cols = g.terminal_width or get_terminal_size()[0]
+			self.cols = g.terminal_width or get_terminal_size().width
 			if self.cols >= g.min_screen_width: break
 			m1 = 'Screen too narrow to display the tracking wallet\n'
 			m2 = 'Please resize your screen to at least {} characters and hit ENTER '
@@ -345,7 +345,7 @@ watch-only wallet using '{}-addrimport' and then re-run this program.
 		while True:
 			msg_r('' if no_output else '\n\n' if opt.no_blank else CUR_HOME+ERASE_ALL)
 			reply = get_char('' if no_output else self.format_for_display()+'\n'+(oneshot_msg or '')+prompt,
-								immed_chars=''.join(self.key_mappings.keys())).decode()
+								immed_chars=''.join(self.key_mappings.keys()))
 			no_output = False
 			oneshot_msg = '' if oneshot_msg else None # tristate, saves previous state
 			if reply not in self.key_mappings:

mmgen/util.py

@@ -760,7 +760,7 @@ def keypress_confirm(prompt,default_yes=False,verbose=False,no_nl=False,complete
 
 	from mmgen.term import get_char
 	while True:
-		reply = get_char(p).decode().strip('\n\r')
+		reply = get_char(p).strip('\n\r')
 		if not reply:
 			msg_r(nl)
 			return True if default_yes else False
@@ -774,7 +774,7 @@ def prompt_and_get_char(prompt,chars,enter_ok=False,verbose=False):
 
 	from mmgen.term import get_char
 	while True:
-		reply = get_char('{}: '.format(prompt)).decode().strip('\n\r')
+		reply = get_char('{}: '.format(prompt)).strip('\n\r')
 		if reply in chars or (enter_ok and not reply):
 			msg('')
 			return reply
@@ -816,7 +816,7 @@ def do_license_msg(immed=False):
 
 	from mmgen.term import get_char
 	while True:
-		reply = get_char(prompt, immed_chars=('','wc')[bool(immed)]).decode()
+		reply = get_char(prompt, immed_chars=('','wc')[bool(immed)])
 		if reply == 'w':
 			do_pager(gpl.conditions)
 		elif reply == 'c':

test/misc/password_entry.py

@@ -6,7 +6,6 @@ parpar = os.path.dirname(os.path.dirname(pn))
 os.chdir(parpar)
 sys.path[0] = os.curdir
 
-from mmgen.util import msg
 from mmgen.common import *
 
 cmd_args = opts.init({'text': { 'desc': '', 'usage':'', 'options':'-e, --echo-passphrase foo' }})

test/misc/term.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+
+import sys,os
+pn = os.path.abspath(os.path.dirname(sys.argv[0]))
+parpar = os.path.dirname(os.path.dirname(pn))
+os.chdir(parpar)
+sys.path[0] = os.curdir
+
+from mmgen.common import *
+
+opts_data = {
+	'text': {
+		'desc': 'Interactively test MMGen terminal functionality',
+		'usage':'',
+		'options': """
+-h, --help     Print this help message
+""",
+	'notes': """
+"""
+	}
+}
+cmd_args = opts.init(opts_data)
+
+from mmgen.term import get_char,get_char_raw,get_terminal_size
+
+def cmsg(m):
+	msg('\n'+cyan(m))
+
+def confirm(m):
+	if not keypress_confirm(m):
+		if keypress_confirm('Are you sure you want to exit test?'):
+			die(1,'Exiting test at user request')
+		else:
+			msg('Continuing...')
+
+def tt_start():
+	m = fmt("""
+		We will now test MMGen’s terminal capabilities.
+		This is a non-automated test and requires user interaction.
+		Continue?
+	""")
+	confirm(m.strip())
+
+def tt_get_terminal_size():
+	cmsg('Testing get_terminal_size():')
+	msg('X' * get_terminal_size().width)
+	confirm('Do the X’s exactly fill the width of the screen?')
+
+def tt_color():
+	cmsg('Testing color:')
+	confirm(blue('THIS TEXT') + ' should be blue.  Is it?')
+
+def tt_license():
+	cmsg('Testing do_license_msg() with pager')
+	ymsg('Press "w" to test the pager, then "c" to continue')
+	do_license_msg()
+
+def tt_my_raw_input():
+	cmsg('Testing my_raw_input():')
+	msg(fmt("""
+		At the Ready? prompt type and hold down "y".
+		Then Enter some text, followed by held-down ENTER.
+		The held-down "y" and ENTER keys should be blocked, not affecting the output
+		on screen or entered text.
+	"""))
+	get_char_raw('Ready? ',num_chars=1)
+	reply = my_raw_input('\nEnter text: ')
+	confirm('Did you enter the text {!r}?'.format(reply))
+
+def tt_prompt_and_get_char():
+	cmsg('Testing prompt_and_get_char():')
+	m = 'Type some letters besides "x" or "z", then "x" or "z"'
+	reply = prompt_and_get_char(m,'xz')
+	confirm('Did you enter the letter {!r}?'.format(reply))
+
+def tt_prompt_and_get_char_enter_ok():
+	cmsg('Testing prompt_and_get_char() with blank choices and enter_ok=True:')
+	for m in (
+		'Type ENTER',
+		'Type any letter followed by a pause, followed by ENTER',
+	):
+		reply = prompt_and_get_char(m,'',enter_ok=True)
+		assert reply == ''
+		msg('OK')
+
+def tt_get_char(raw=False,one_char=False,sleep=0,immed_chars=''):
+	fname = ('get_char','get_char_raw')[raw]
+	fs = fmt("""
+		Press some keys in quick succession.
+		{}{}{}
+		{}
+		When you’re finished, use Ctrl-C to exit.
+		""").strip()
+	m1 = (
+		'You should experience a delay with quickly repeated entry.',
+		'Your entry should be repeated back to you immediately.'
+	)[raw]
+	m2 = (
+		'',
+		'\nA delay of {} seconds will added before each prompt'.format(sleep)
+	)[bool(sleep)]
+	m3 = (
+		'',
+		'\nThe characters {!r} will be repeated immediately, the others with delay.'.format(immed_chars)
+	)[bool(immed_chars)]
+	m4 = 'The F1-F12 keys will be ' + (
+		'blocked entirely.'
+			if one_char and not raw else
+		"echoed AS A SINGLE character '\\x1b'."
+			if one_char else
+		'echoed as a FULL CONTROL SEQUENCE.'
+	)
+	if g.platform == 'win':
+		m4 = 'The Escape and F1-F12 keys will be returned as single characters.'
+	kwargs = {}
+	if one_char:
+		kwargs.update({'num_chars':1})
+	if sleep:
+		kwargs.update({'sleep':sleep})
+	if immed_chars:
+		kwargs.update({'immed_chars':immed_chars})
+
+	cmsg('Testing {}({}):'.format(fname,','.join(['{}={!r}'.format(*i) for i in kwargs.items()])))
+	msg(fs.format(m1,yellow(m2),yellow(m3),yellow(m4)))
+
+	try:
+		while True:
+			ret = globals()[fname]('Enter a letter: ',**kwargs)
+			msg('You typed {!r}'.format(ret))
+	except KeyboardInterrupt:
+		msg('\nDone')
+
+if g.platform == 'linux':
+	import termios,atexit
+	fd = sys.stdin.fileno()
+	old = termios.tcgetattr(fd)
+	atexit.register(lambda: termios.tcsetattr(fd,termios.TCSADRAIN,old))
+
+tt_start()
+
+tt_get_terminal_size()
+tt_color()
+tt_license()
+tt_my_raw_input()
+tt_prompt_and_get_char()
+tt_prompt_and_get_char_enter_ok()
+
+tt_get_char(one_char=True)
+tt_get_char(one_char=True,sleep=1)
+tt_get_char(one_char=True,raw=True)
+
+if g.platform == 'linux':
+	tt_get_char(one_char=False)
+	tt_get_char(one_char=False,immed_chars='asdf')
+	tt_get_char(one_char=False,raw=True)
+else:
+	tt_get_char(one_char=True,immed_chars='asdf')
+
+gmsg('\nTest completed')

test/test.py

@@ -431,7 +431,7 @@ def create_tmp_dirs(shm_dir):
 def set_environ_for_spawned_scripts():
 
 	from mmgen.term import get_terminal_size
-	os.environ['MMGEN_TERMINAL_WIDTH'] = str(get_terminal_size()[0])
+	os.environ['MMGEN_TERMINAL_WIDTH'] = str(get_terminal_size().width)
 
 	if os.getenv('MMGEN_DEBUG_ALL'):
 		for name in g.env_opts: