"""
Terminal-related utilities
--------------------------
"""
from __future__ import division, print_function, absolute_import
import sys
import os
from plumbum import local
from .termsize import get_terminal_size
from .progress import Progress
[docs]def readline(message=""):
"""Gets a line of input from the user (stdin)"""
sys.stdout.write(message)
sys.stdout.flush()
return sys.stdin.readline()
[docs]def ask(question, default=None):
"""
Presents the user with a yes/no question.
:param question: The question to ask
:param default: If ``None``, the user must answer. If ``True`` or ``False``, lack of response is
interpreted as the default option
:returns: the user's choice
"""
question = question.rstrip().rstrip("?").rstrip() + "?"
if default is None:
question += " (y/n) "
elif default:
question += " [Y/n] "
else:
question += " [y/N] "
while True:
try:
answer = readline(question).strip().lower()
except EOFError:
answer = None
if answer in ("y", "yes"):
return True
elif answer in ("n", "no"):
return False
elif not answer and default is not None:
return default
else:
sys.stdout.write("Invalid response, please try again\n")
[docs]def choose(question, options, default=None):
"""Prompts the user with a question and a set of options, from which the user needs to choose.
:param question: The question to ask
:param options: A set of options. It can be a list (of strings or two-tuples, mapping text
to returned-object) or a dict (mapping text to returned-object).``
:param default: If ``None``, the user must answer. Otherwise, lack of response is interpreted
as this answer
:returns: The user's choice
Example::
ans = choose("What is your favorite color?", ["blue", "yellow", "green"], default = "yellow")
# `ans` will be one of "blue", "yellow" or "green"
ans = choose("What is your favorite color?",
{"blue" : 0x0000ff, "yellow" : 0xffff00 , "green" : 0x00ff00}, default = 0x00ff00)
# this will display "blue", "yellow" and "green" but return a numerical value
"""
if hasattr(options, "items"):
options = options.items()
sys.stdout.write(question.rstrip() + "\n")
choices = {}
defindex = None
for i, item in enumerate(options):
i = i + 1 # python2.5
if isinstance(item, (tuple, list)) and len(item) == 2:
text = item[0]
val = item[1]
else:
text = item
val = item
choices[i] = val
if default is not None and default == val:
defindex = i
sys.stdout.write("(%d) %s\n" % (i, text))
if default is not None:
if defindex is None:
msg = "Choice [%s]: " % (default, )
else:
msg = "Choice [%d]: " % (defindex, )
else:
msg = "Choice: "
while True:
try:
choice = readline(msg).strip()
except EOFError:
choice = ""
if not choice and default:
return default
try:
choice = int(choice)
if choice not in choices:
raise ValueError()
except ValueError:
sys.stdout.write("Invalid choice, please try again\n")
continue
return choices[choice]
[docs]def prompt(question,
type=str,
default=NotImplemented,
validator=lambda val: True):
"""
Presents the user with a validated question, keeps asking if validation does not pass.
:param question: The question to ask
:param type: The type of the answer, defaults to str
:param default: The default choice
:param validator: An extra validator called after type conversion, can raise ValueError or return False to trigger a retry.
:returns: the user's choice
"""
question = question.rstrip(" \t:")
if default is not NotImplemented:
question += " [%s]" % (default, )
question += ": "
while True:
try:
ans = readline(question).strip()
except EOFError:
ans = ""
if not ans:
if default is not NotImplemented:
#sys.stdout.write("\b%s\n" % (default,))
return default
else:
continue
try:
ans = type(ans)
except (TypeError, ValueError) as ex:
sys.stdout.write("Invalid value (%s), please try again\n" % (ex, ))
continue
try:
valid = validator(ans)
except ValueError as ex:
sys.stdout.write("%s, please try again\n" % (ex, ))
continue
if not valid:
sys.stdout.write(
"Value not in specified range, please try again\n")
continue
return ans
[docs]def hexdump(data_or_stream, bytes_per_line=16, aggregate=True):
"""Convert the given bytes (or a stream with a buffering ``read()`` method) to hexdump-formatted lines,
with possible aggregation of identical lines. Returns a generator of formatted lines.
"""
if hasattr(data_or_stream, "read"):
def read_chunk():
while True:
buf = data_or_stream.read(bytes_per_line)
if not buf:
break
yield buf
else:
def read_chunk():
for i in range(0, len(data_or_stream), bytes_per_line):
yield data_or_stream[i:i + bytes_per_line]
prev = None
skipped = False
for i, chunk in enumerate(read_chunk()):
hexd = " ".join("%02x" % (ord(ch), ) for ch in chunk)
text = "".join(ch if 32 <= ord(ch) < 127 else "." for ch in chunk)
if aggregate and prev == chunk:
skipped = True
continue
prev = chunk
if skipped:
yield "*"
yield "%06x | %s| %s" % (i * bytes_per_line,
hexd.ljust(bytes_per_line * 3, " "), text)
skipped = False
