I am experimenting with showing progress on the terminal for a subcommand that is being run, showing what is happening without scrolling away the output of the main program, and I came out with this little toy. It shows the last X lines of a subcommand output, then gets rid of everything after the command has ended.
Usability-wise, it feels like a tease to me: it looks like I'm being shown all sorts of information then they are taken away from me before I managed to make sense of them. However, I find it cute enough to share:
#!/usr/bin/env python3 #coding: utf-8 # Copyright 2015 Enrico Zini <enrico@enricozini.org>. Licensed under the terms # of the GNU General Public License, version 2 or any later version. import argparse import fcntl import select import curses import contextlib import subprocess import os import sys import collections import shlex import shutil import logging def stream_output(proc): """ Take a subprocess.Popen object and generate its output, line by line, annotated with "stdout" or "stderr". At process termination it generates one last element: ("result", return_code) with the return code of the process. """ fds = [proc.stdout, proc.stderr] bufs = [b"", b""] types = ["stdout", "stderr"] # Set both pipes as non-blocking for fd in fds: fcntl.fcntl(fd, fcntl.F_SETFL, os.O_NONBLOCK) # Multiplex stdout and stderr with different prefixes while len(fds) > 0: s = select.select(fds, (), ()) for fd in s[0]: idx = fds.index(fd) buf = fd.read() if len(buf) == 0: fds.pop(idx) if len(bufs[idx]) != 0: yield types[idx], bufs.pop(idx) types.pop(idx) else: bufs[idx] += buf lines = bufs[idx].split(b"\n") bufs[idx] = lines.pop() for l in lines: yield types[idx], l res = proc.wait() yield "result", res @contextlib.contextmanager def miniscreen(has_fancyterm, name, maxlines=3, silent=False): """ Show the output of a process scrolling in a portion of the screen. has_fancyterm: true if the terminal supports fancy features; if false, just write lines to standard output name: name of the process being run, to use as a header maxlines: maximum height of the miniscreen silent: do nothing whatsoever, used to disable this without needing to change the code structure Usage: with miniscreen(True, "my process", 5) as print_line: for i in range(10): print_line(("stdout", "stderr")[i % 2], "Line #{}".format(i)) """ if not silent and has_fancyterm: # Discover all the terminal control sequences that we need output_normal = str(curses.tigetstr("sgr0"), "ascii") output_up = str(curses.tigetstr("cuu1"), "ascii") output_clreol = str(curses.tigetstr("el"), "ascii") cols, lines = shutil.get_terminal_size() output_width = cols fg_color = (curses.tigetstr("setaf") or curses.tigetstr("setf") or "") sys.stdout.write(str(curses.tparm(fg_color, 6), "ascii")) output_lines = collections.deque(maxlen=maxlines) def print_lines(): """ Print the lines in our buffer, then move back to the beginning """ sys.stdout.write("{} progress:".format(name)) sys.stdout.write(output_clreol) for msg in output_lines: sys.stdout.write("\n") sys.stdout.write(msg) sys.stdout.write(output_clreol) sys.stdout.write(output_up * len(output_lines)) sys.stdout.write("\r") try: print_lines() def _progress_line(type, line): """ Print a new line to the miniscreen """ # Add the new line to our output buffer msg = "{} {}".format("." if type == "stdout" else "!", line) if len(msg) > output_width - 4: msg = msg[:output_width - 4] + "..." output_lines.append(msg) # Update the miniscreen print_lines() yield _progress_line # Clear the miniscreen by filling our ring buffer with empty lines # then printing them out for i in range(maxlines): output_lines.append("") print_lines() finally: sys.stdout.write(output_normal) elif not silent: def _progress_line(type, line): print("{}: {}".format(type, line)) yield _progress_line else: def _progress_line(type, line): pass yield _progress_line def run_command_fancy(name, cmd, env=None, logfd=None, fancy=True, debug=False): quoted_cmd = " ".join(shlex.quote(x) for x in cmd) log.info("%s running command %s", name, quoted_cmd) if logfd: print("runcmd:", quoted_cmd, file=logfd) # Run the script itself on an empty environment, so that what was # documented is exactly what was run proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) with miniscreen(fancy, name, silent=debug) as progress: stderr = [] for type, val in stream_output(proc): if type == "stdout": val = val.decode("utf-8") if logfd: print("stdout:", val, file=logfd) log.debug("%s stdout: %s", name, val) progress(type, val) elif type == "stderr": val = val.decode("utf-8") if logfd: print("stderr:", val, file=logfd) stderr.append(val) log.debug("%s stderr: %s", name, val) progress(type, val) elif type == "result": if logfd: print("retval:", val, file=logfd) log.debug("%s retval: %d", name, val) retval = val if retval != 0: lastlines = min(len(stderr), 5) log.error("%s exited with code %s", name, retval) log.error("Last %d lines of standard error:", lastlines) for line in stderr[-lastlines:]: log.error("%s: %s", name, line) return retval parser = argparse.ArgumentParser(description="run a command showing only a portion of its output") parser.add_argument("--logfile", action="store", help="specify a file where the full execution log will be written") parser.add_argument("--debug", action="store_true", help="debugging output on the terminal") parser.add_argument("--verbose", action="store_true", help="verbose output on the terminal") parser.add_argument("command", nargs="*", help="command to run") args = parser.parse_args() if args.debug: loglevel = logging.DEBUG elif args.verbose: loglevel = logging.INFO else: loglevel = logging.WARN logging.basicConfig(level=loglevel, stream=sys.stderr) log = logging.getLogger() fancy = False if not args.debug and sys.stdout.isatty(): curses.setupterm() if curses.tigetnum("colors") > 0: fancy = True if args.logfile: logfd = open("output.log", "wt") else: logfd = None retval = run_command_fancy("miniscreen example", args.command, logfd=logfd) sys.exit(retval)