live visualizing simple data as a grid

grid.py

#!/usr/bin/env python3
# type: ignore
import argh
import json
import sys
import time
import blessed

term = blessed.Terminal()

def chunkify(val, chunk_size):
    res = []
    for x in val:
        res.append(x)
        if len(res) == chunk_size:
            yield res
            res = []
    if res:
        yield res

def main(interval_seconds=1,
         green: 'space separated substrings to highlight in green' = '',
         exclude: 'space separated substrings to exclude' = '',
         justify: 'how big do values get' = 10,
         include: 'space separated substrings to include' = ''):
    """
    live visualizing simple data as a grid.

    usage: libaws ec2-ssh $name -c 'curl -s https://gist.githubusercontent.com/nathants/8e3b26e769abf86ece8d/raw/ | python3' 2>&1 | grid

    demo: https://r2.nathants.workers.dev/link/de7f0691-29cd-47c1-bd35-bcfc8a327aa3/grid.gif

    note: libaws ec2-ssh is from https://github.com/nathants/libaws

    """
    datas = {}
    last = time.monotonic()
    for line in sys.stdin:
        try:
            key, data = line.split(None, 1)
            data = json.loads(data)
            datas[key] = data
        except:
            pass
        else:
            if time.monotonic() - last > interval_seconds:
                screen = []
                panes = []
                for key in sorted(datas):
                    pane = []
                    pane.append(key)
                    just = max(len(k) for k in datas[key])
                    for k in sorted(datas[key]):
                        v = datas[key][k]
                        k = k.rjust(just)
                        if exclude and any(token in k for token in exclude.split()):
                            continue
                        if include and not any(token in k for token in include.split()):
                            continue
                        pane.append(f'{k}: {str(v)}')
                    panes.append(pane)
                num_chunks = term.width // (just + justify)
                for chunk in chunkify(panes, num_chunks):
                    if len(screen) + len(chunk[0]) > term.height:
                        break
                    for lines in zip(*[c for c in chunk]):
                        screen.append(' | '.join(l.rstrip().ljust(just + justify) for l in lines)[:term.width])
                    screen.append('-' * term.width)
                colorized_screen = []
                for line in screen:
                    if set(line) == {'-'}:
                        colorized_screen.append(line)
                    else:
                        new_line = []
                        for part in line.split('|'):
                            if 2 == len(part.rstrip(' :').split(':')):
                                head, tail = part.split(':')
                                if green and any(token in head for token in green.split()):
                                    new_line.append(f'{term.yellow(head)}:{term.green(tail)}')
                                else:
                                    new_line.append(f'{term.yellow(head)}:{term.red(tail)}')
                            else:
                                new_line.append(term.blue(part))
                        colorized_screen.append('|'.join(new_line))
                print(term.clear + '\n'.join(colorized_screen))
                last = time.monotonic()

if __name__ == '__main__':
    with term.fullscreen():
        with term.hidden_cursor():
            argh.dispatch_command(main)