From a231bbac6e3aa7651a324ffe90a7edd684dcb42a Mon Sep 17 00:00:00 2001 From: Christopher Haster Date: Sun, 2 Jun 2024 00:07:35 -0500 Subject: [PATCH] Added -^/--head to watch.py and other scripts, RingIO tweaks This lets you view the first n lines of output instead of the last n lines, as though the output was piped through head. This is how the standard watch command works, and can be more useful when most of the information is at the top, such as in our dbg*.py scripts (watch.py was originally used as a sort of inotifywait-esque build runner, which is the main reason it's different). To make this work, RingIO (renamed from LinesIO) now uses terminal height as a part of its canvas rendering. This has the added benefit of more rigorously enforcing the canvas boundaries, but risks breaking when not associated with a terminal. But that raises the question, does RingIO even make sense without a terminal? Worst case you can bypass all of this with -z/--cat. --- scripts/plot.py | 66 +++++++++++++++++++++++-------------------- scripts/tailpipe.py | 58 +++++++++++++++++++++---------------- scripts/tracebd.py | 57 ++++++++++++++++++++++--------------- scripts/watch.py | 69 ++++++++++++++++++++++++++++----------------- 4 files changed, 147 insertions(+), 103 deletions(-) diff --git a/scripts/plot.py b/scripts/plot.py index 4b704262..2adf3211 100755 --- a/scripts/plot.py +++ b/scripts/plot.py @@ -167,9 +167,10 @@ else: else: self.add_watch(path, flags) -class LinesIO: - def __init__(self, maxlen=None): +class RingIO: + def __init__(self, maxlen=None, head=False): self.maxlen = maxlen + self.head = head self.lines = co.deque(maxlen=maxlen) self.tail = io.StringIO() @@ -177,6 +178,9 @@ class LinesIO: if maxlen == 0: self.resize(0) + def __len__(self): + return len(self.lines) + def write(self, s): # note using split here ensures the trailing string has no newline lines = s.split('\n') @@ -204,35 +208,37 @@ class LinesIO: if self.maxlen == 0: self.resize(0) + # copy lines + lines = self.lines.copy() + # pad to fill any existing canvas, but truncate to terminal size + h = shutil.get_terminal_size((80, 5))[1] + lines.extend('' for _ in range( + len(lines), + min(RingIO.canvas_lines, h))) + while len(lines) > h: + if self.head: + lines.pop() + else: + lines.popleft() + # first thing first, give ourself a canvas - while LinesIO.canvas_lines < len(self.lines): + while RingIO.canvas_lines < len(lines): sys.stdout.write('\n') - LinesIO.canvas_lines += 1 + RingIO.canvas_lines += 1 - # clear the bottom of the canvas if we shrink - shrink = LinesIO.canvas_lines - len(self.lines) - if shrink > 0: - for i in range(shrink): - sys.stdout.write('\r') - if shrink-1-i > 0: - sys.stdout.write('\x1b[%dA' % (shrink-1-i)) - sys.stdout.write('\x1b[K') - if shrink-1-i > 0: - sys.stdout.write('\x1b[%dB' % (shrink-1-i)) - sys.stdout.write('\x1b[%dA' % shrink) - LinesIO.canvas_lines = len(self.lines) - - for i, line in enumerate(self.lines): + # write lines from top to bottom so later lines overwrite earlier + # lines, note [xA/[xB stop at terminal boundaries + for i, line in enumerate(lines): # move cursor, clear line, disable/reenable line wrapping sys.stdout.write('\r') - if len(self.lines)-1-i > 0: - sys.stdout.write('\x1b[%dA' % (len(self.lines)-1-i)) + if len(lines)-1-i > 0: + sys.stdout.write('\x1b[%dA' % (len(lines)-1-i)) sys.stdout.write('\x1b[K') sys.stdout.write('\x1b[?7l') sys.stdout.write(line) sys.stdout.write('\x1b[?7h') - if len(self.lines)-1-i > 0: - sys.stdout.write('\x1b[%dB' % (len(self.lines)-1-i)) + if len(lines)-1-i > 0: + sys.stdout.write('\x1b[%dB' % (len(lines)-1-i)) sys.stdout.flush() @@ -845,6 +851,7 @@ def main(csv_paths, *, legend_below=False, subplot={}, subplots=[], + head=False, cat=False, keep_open=False, sleep=None, @@ -1402,7 +1409,7 @@ def main(csv_paths, *, if cat: draw(sys.stdout) else: - ring = LinesIO() + ring = RingIO(head=head) draw(ring) ring.draw() @@ -1419,13 +1426,8 @@ def main(csv_paths, *, except KeyboardInterrupt: pass - if cat: - draw(sys.stdout) - else: - ring = LinesIO() - draw(ring) - ring.draw() - sys.stdout.write('\n') + if not cat: + sys.stdout.write('\n') else: draw(sys.stdout) @@ -1644,6 +1646,10 @@ if __name__ == "__main__": '--subplot', type=AppendSubplot.parse, help="Add subplot-specific arguments to the main plot.") + parser.add_argument( + '-^', '--head', + action='store_true', + help="Show the first n lines.") parser.add_argument( '-z', '--cat', action='store_true', diff --git a/scripts/tailpipe.py b/scripts/tailpipe.py index 9bd9d1d0..9f0af4d3 100755 --- a/scripts/tailpipe.py +++ b/scripts/tailpipe.py @@ -29,9 +29,10 @@ def openio(path, mode='r', buffering=-1): else: return open(path, mode, buffering) -class LinesIO: - def __init__(self, maxlen=None): +class RingIO: + def __init__(self, maxlen=None, head=False): self.maxlen = maxlen + self.head = head self.lines = co.deque(maxlen=maxlen) self.tail = io.StringIO() @@ -39,6 +40,9 @@ class LinesIO: if maxlen == 0: self.resize(0) + def __len__(self): + return len(self.lines) + def write(self, s): # note using split here ensures the trailing string has no newline lines = s.split('\n') @@ -66,43 +70,49 @@ class LinesIO: if self.maxlen == 0: self.resize(0) + # copy lines + lines = self.lines.copy() + # pad to fill any existing canvas, but truncate to terminal size + h = shutil.get_terminal_size((80, 5))[1] + lines.extend('' for _ in range( + len(lines), + min(RingIO.canvas_lines, h))) + while len(lines) > h: + if self.head: + lines.pop() + else: + lines.popleft() + # first thing first, give ourself a canvas - while LinesIO.canvas_lines < len(self.lines): + while RingIO.canvas_lines < len(lines): sys.stdout.write('\n') - LinesIO.canvas_lines += 1 + RingIO.canvas_lines += 1 - # clear the bottom of the canvas if we shrink - shrink = LinesIO.canvas_lines - len(self.lines) - if shrink > 0: - for i in range(shrink): - sys.stdout.write('\r') - if shrink-1-i > 0: - sys.stdout.write('\x1b[%dA' % (shrink-1-i)) - sys.stdout.write('\x1b[K') - if shrink-1-i > 0: - sys.stdout.write('\x1b[%dB' % (shrink-1-i)) - sys.stdout.write('\x1b[%dA' % shrink) - LinesIO.canvas_lines = len(self.lines) - - for i, line in enumerate(self.lines): + # write lines from top to bottom so later lines overwrite earlier + # lines, note [xA/[xB stop at terminal boundaries + for i, line in enumerate(lines): # move cursor, clear line, disable/reenable line wrapping sys.stdout.write('\r') - if len(self.lines)-1-i > 0: - sys.stdout.write('\x1b[%dA' % (len(self.lines)-1-i)) + if len(lines)-1-i > 0: + sys.stdout.write('\x1b[%dA' % (len(lines)-1-i)) sys.stdout.write('\x1b[K') sys.stdout.write('\x1b[?7l') sys.stdout.write(line) sys.stdout.write('\x1b[?7h') - if len(self.lines)-1-i > 0: - sys.stdout.write('\x1b[%dB' % (len(self.lines)-1-i)) + if len(lines)-1-i > 0: + sys.stdout.write('\x1b[%dB' % (len(lines)-1-i)) sys.stdout.flush() -def main(path='-', *, lines=5, cat=False, sleep=None, keep_open=False): +def main(path='-', *, + lines=5, + cat=False, + sleep=None, + keep_open=False): if cat: ring = sys.stdout else: - ring = LinesIO(lines) + ring = RingIO(lines) # if sleep print in background thread to avoid getting stuck in a read call event = th.Event() diff --git a/scripts/tracebd.py b/scripts/tracebd.py index c51227ff..a2247910 100755 --- a/scripts/tracebd.py +++ b/scripts/tracebd.py @@ -110,9 +110,10 @@ def rbydaddr(s): return addr -class LinesIO: - def __init__(self, maxlen=None): +class RingIO: + def __init__(self, maxlen=None, head=False): self.maxlen = maxlen + self.head = head self.lines = co.deque(maxlen=maxlen) self.tail = io.StringIO() @@ -120,6 +121,9 @@ class LinesIO: if maxlen == 0: self.resize(0) + def __len__(self): + return len(self.lines) + def write(self, s): # note using split here ensures the trailing string has no newline lines = s.split('\n') @@ -147,35 +151,37 @@ class LinesIO: if self.maxlen == 0: self.resize(0) + # copy lines + lines = self.lines.copy() + # pad to fill any existing canvas, but truncate to terminal size + h = shutil.get_terminal_size((80, 5))[1] + lines.extend('' for _ in range( + len(lines), + min(RingIO.canvas_lines, h))) + while len(lines) > h: + if self.head: + lines.pop() + else: + lines.popleft() + # first thing first, give ourself a canvas - while LinesIO.canvas_lines < len(self.lines): + while RingIO.canvas_lines < len(lines): sys.stdout.write('\n') - LinesIO.canvas_lines += 1 + RingIO.canvas_lines += 1 - # clear the bottom of the canvas if we shrink - shrink = LinesIO.canvas_lines - len(self.lines) - if shrink > 0: - for i in range(shrink): - sys.stdout.write('\r') - if shrink-1-i > 0: - sys.stdout.write('\x1b[%dA' % (shrink-1-i)) - sys.stdout.write('\x1b[K') - if shrink-1-i > 0: - sys.stdout.write('\x1b[%dB' % (shrink-1-i)) - sys.stdout.write('\x1b[%dA' % shrink) - LinesIO.canvas_lines = len(self.lines) - - for i, line in enumerate(self.lines): + # write lines from top to bottom so later lines overwrite earlier + # lines, note [xA/[xB stop at terminal boundaries + for i, line in enumerate(lines): # move cursor, clear line, disable/reenable line wrapping sys.stdout.write('\r') - if len(self.lines)-1-i > 0: - sys.stdout.write('\x1b[%dA' % (len(self.lines)-1-i)) + if len(lines)-1-i > 0: + sys.stdout.write('\x1b[%dA' % (len(lines)-1-i)) sys.stdout.write('\x1b[K') sys.stdout.write('\x1b[?7l') sys.stdout.write(line) sys.stdout.write('\x1b[?7h') - if len(self.lines)-1-i > 0: - sys.stdout.write('\x1b[%dB' % (len(self.lines)-1-i)) + if len(lines)-1-i > 0: + sys.stdout.write('\x1b[%dB' % (len(lines)-1-i)) sys.stdout.flush() @@ -662,6 +668,7 @@ def main(path='-', *, width=None, height=None, lines=None, + head=False, cat=False, hilbert=False, lebesgue=False, @@ -991,7 +998,7 @@ def main(path='-', *, if cat: ring = sys.stdout else: - ring = LinesIO(lines + (1 if not no_header else 0)) + ring = RingIO(lines + (1 if not no_header else 0), head) # if sleep print in background thread to avoid getting stuck in a read call event = th.Event() @@ -1159,6 +1166,10 @@ if __name__ == "__main__": const=0, help="Show this many lines of history. 0 uses the terminal height. " "Defaults to 5.") + parser.add_argument( + '-^', '--head', + action='store_true', + help="Show the first n lines.") parser.add_argument( '-z', '--cat', action='store_true', diff --git a/scripts/watch.py b/scripts/watch.py index 1c4211bd..f6afac68 100755 --- a/scripts/watch.py +++ b/scripts/watch.py @@ -67,9 +67,10 @@ else: else: self.add_watch(path, flags) -class LinesIO: - def __init__(self, maxlen=None): +class RingIO: + def __init__(self, maxlen=None, head=False): self.maxlen = maxlen + self.head = head self.lines = co.deque(maxlen=maxlen) self.tail = io.StringIO() @@ -77,6 +78,9 @@ class LinesIO: if maxlen == 0: self.resize(0) + def __len__(self): + return len(self.lines) + def write(self, s): # note using split here ensures the trailing string has no newline lines = s.split('\n') @@ -104,40 +108,43 @@ class LinesIO: if self.maxlen == 0: self.resize(0) + # copy lines + lines = self.lines.copy() + # pad to fill any existing canvas, but truncate to terminal size + h = shutil.get_terminal_size((80, 5))[1] + lines.extend('' for _ in range( + len(lines), + min(RingIO.canvas_lines, h))) + while len(lines) > h: + if self.head: + lines.pop() + else: + lines.popleft() + # first thing first, give ourself a canvas - while LinesIO.canvas_lines < len(self.lines): + while RingIO.canvas_lines < len(lines): sys.stdout.write('\n') - LinesIO.canvas_lines += 1 + RingIO.canvas_lines += 1 - # clear the bottom of the canvas if we shrink - shrink = LinesIO.canvas_lines - len(self.lines) - if shrink > 0: - for i in range(shrink): - sys.stdout.write('\r') - if shrink-1-i > 0: - sys.stdout.write('\x1b[%dA' % (shrink-1-i)) - sys.stdout.write('\x1b[K') - if shrink-1-i > 0: - sys.stdout.write('\x1b[%dB' % (shrink-1-i)) - sys.stdout.write('\x1b[%dA' % shrink) - LinesIO.canvas_lines = len(self.lines) - - for i, line in enumerate(self.lines): + # write lines from top to bottom so later lines overwrite earlier + # lines, note [xA/[xB stop at terminal boundaries + for i, line in enumerate(lines): # move cursor, clear line, disable/reenable line wrapping sys.stdout.write('\r') - if len(self.lines)-1-i > 0: - sys.stdout.write('\x1b[%dA' % (len(self.lines)-1-i)) + if len(lines)-1-i > 0: + sys.stdout.write('\x1b[%dA' % (len(lines)-1-i)) sys.stdout.write('\x1b[K') sys.stdout.write('\x1b[?7l') sys.stdout.write(line) sys.stdout.write('\x1b[?7h') - if len(self.lines)-1-i > 0: - sys.stdout.write('\x1b[%dB' % (len(self.lines)-1-i)) + if len(lines)-1-i > 0: + sys.stdout.write('\x1b[%dB' % (len(lines)-1-i)) sys.stdout.flush() def main(command, *, lines=0, + head=False, cat=False, sleep=None, keep_open=False, @@ -145,6 +152,11 @@ def main(command, *, buffer=False, ignore_errors=False, exit_on_error=False): + if not command: + print('usage: %s [options] command' % sys.argv[0], + file=sys.stderr) + sys.exit(-1) + # if we have keep_open_paths, assume user wanted keep_open if keep_open_paths and not keep_open: keep_open = True @@ -171,7 +183,7 @@ def main(command, *, if cat: ring = sys.stdout else: - ring = LinesIO(lines) + ring = RingIO(lines, head) try: # register inotify before running the command, this avoids @@ -206,9 +218,10 @@ def main(command, *, if not line: break - ring.write(line) - if not cat and not buffer and not ignore_errors: - ring.draw() + if cat or not head or len(ring) < h: + ring.write(line) + if not cat and not buffer and not ignore_errors: + ring.draw() mpty.close() proc.wait() @@ -262,6 +275,10 @@ if __name__ == "__main__": const=0, help="Show this many lines of history. 0 uses the terminal height. " "Defaults to 0.") + parser.add_argument( + '-^', '--head', + action='store_true', + help="Show the first n lines.") parser.add_argument( '-z', '--cat', action='store_true',