diff --git a/scripts/codemap.py b/scripts/codemap.py index 255d4f70..a7265152 100755 --- a/scripts/codemap.py +++ b/scripts/codemap.py @@ -1382,7 +1382,7 @@ if __name__ == "__main__": action=AppendPath, help="Input *.json files.") parser.add_argument( - '-n', '--namespace-depth', + '-_', '--namespace-depth', nargs='?', type=lambda x: int(x, 0), const=0, diff --git a/scripts/codemapd3.py b/scripts/codemapd3.py index d0edc845..d67cae7d 100755 --- a/scripts/codemapd3.py +++ b/scripts/codemapd3.py @@ -2090,7 +2090,7 @@ if __name__ == "__main__": required=True, help="Output *.svg file.") parser.add_argument( - '-n', '--namespace-depth', + '-_', '--namespace-depth', nargs='?', type=lambda x: int(x, 0), const=0, diff --git a/scripts/dbgbmap.py b/scripts/dbgbmap.py index 6267fe06..90d34db7 100755 --- a/scripts/dbgbmap.py +++ b/scripts/dbgbmap.py @@ -3645,6 +3645,7 @@ else: else: self.add_watch(path, flags) +# TODO negative maxlen from terminal height? like -H nowadays? class RingIO: def __init__(self, maxlen=None, head=False): self.maxlen = maxlen @@ -4392,23 +4393,25 @@ def main_(f, disk, mroots=None, *, corrupted = not bool(lfs) # if we can't figure out the block_count, guess + block_size_ = block_size + block_count_ = block_count if block_count is None: if lfs.config.geometry is not None: - block_count = lfs.config.geometry.block_count + block_count_ = lfs.config.geometry.block_count else: f_.seek(0, os.SEEK_END) - block_count = mt.ceil(f_.tell() / block_size) + block_count_ = mt.ceil(f_.tell() / block_size) # flatten blocks, default to all blocks - blocks = list( - range(blocks.start or 0, blocks.stop or block_count) + blocks_ = list( + range(blocks.start or 0, blocks.stop or block_count_) if isinstance(blocks, slice) else range(blocks, blocks+1) if blocks - else range(block_count)) + else range(block_count_)) # traverse the filesystem and create a block map - bmap = {b: BmapBlock(b, 'unused') for b in blocks} + bmap = {b: BmapBlock(b, 'unused') for b in blocks_} mdir_count = 0 btree_count = 0 data_count = 0 @@ -4455,21 +4458,21 @@ def main_(f, disk, mroots=None, *, else: bmap[b] = BmapBlock(b, 'conflict', [bmap[b].value, child], - range(block_size)) + range(block_size_)) corrupted = True # corrupt metadata? elif (not no_ckmeta and isinstance(child, (Mdir, Rbyd)) and not child): - bmap[b] = BmapBlock(b, 'corrupt', child, range(block_size)) + bmap[b] = BmapBlock(b, 'corrupt', child, range(block_size_)) corrupted = True # corrupt data? elif (not no_ckdata and isinstance(child, Bptr) and not child): - bmap[b] = BmapBlock(b, 'corrupt', child, range(block_size)) + bmap[b] = BmapBlock(b, 'corrupt', child, range(block_size_)) corrupted = True # normal block @@ -4509,44 +4512,47 @@ def main_(f, disk, mroots=None, *, # if contiguous, compute the global curve if contiguous: - min__ = min(bmap.keys(), default=0) - curve__ = list(curve(canvas.width, canvas.height)) + global_block = min(bmap.keys(), default=0) + global_curve = list(curve(canvas.width, canvas.height)) # if blocky, figure out block sizes/locations else: - # figure out block_cols/block_rows + # figure out block_cols_/block_rows_ if block_cols is not None and block_rows is not None: - pass + block_cols_ = block_cols + block_rows_ = block_rows elif block_rows is not None: - block_cols = mt.ceil(len(bmap) / block_rows) + block_cols_ = mt.ceil(len(bmap) / block_rows) + block_rows_ = block_rows elif block_cols is not None: - block_rows = mt.ceil(len(bmap) / block_cols) + block_cols_ = block_cols + block_rows_ = mt.ceil(len(bmap) / block_cols) else: # divide by 2 until we hit our target ratio, this works # well for things that are often powers-of-two - block_cols = 1 - block_rows = len(bmap) - while (abs(((canvas.width/(block_cols * 2)) - / (canvas.height/mt.ceil(block_rows / 2))) + block_cols_ = 1 + block_rows_ = len(bmap) + while (abs(((canvas.width/(block_cols_ * 2)) + / (canvas.height/mt.ceil(block_rows_ / 2))) - block_ratio) - < abs(((canvas.width/block_cols) - / (canvas.height/block_rows))) + < abs(((canvas.width/block_cols_) + / (canvas.height/block_rows_))) - block_ratio): - block_cols *= 2 - block_rows = mt.ceil(block_rows / 2) + block_cols_ *= 2 + block_rows_ = mt.ceil(block_rows_ / 2) - block_width = canvas.width / block_cols - block_height = canvas.height / block_rows + block_width_ = canvas.width / block_cols_ + block_height_ = canvas.height / block_rows_ - # assign block locations based on block_rows/block_cols and the - # requested space filling curve + # assign block locations based on block_rows_/block_cols_ and + # the requested space filling curve for (x, y), b in zip( - curve(block_cols, block_rows), + curve(block_cols_, block_rows_), sorted(bmap.values())): - b.x = x * block_width - b.y = y * block_height - b.width = block_width - b.height = block_height + b.x = x * block_width_ + b.y = y * block_height_ + b.width = block_width_ + b.height = block_height_ # apply top padding if x == 0: @@ -4603,25 +4609,25 @@ def main_(f, disk, mroots=None, *, # skip blocks with no usage if not b.usage: continue - block__ = b.block - min__ + block__ = b.block - global_block usage__ = range( - mt.floor(((block__*block_size + b.usage.start) - / (block_size * len(bmap))) - * len(curve__)), - mt.ceil(((block__*block_size + b.usage.stop) - / (block_size * len(bmap))) - * len(curve__))) + mt.floor(((block__*block_size_ + b.usage.start) + / (block_size_ * len(bmap))) + * len(global_curve)), + mt.ceil(((block__*block_size_ + b.usage.stop) + / (block_size_ * len(bmap))) + * len(global_curve))) else: - block__ = b.block - min__ + block__ = b.block - global_block usage__ = range( - mt.floor((block__/len(bmap)) * len(curve__)), - mt.ceil((block__/len(bmap)) * len(curve__))) + mt.floor((block__/len(bmap)) * len(global_curve)), + mt.ceil((block__/len(bmap)) * len(global_curve))) # map to global curve for i in usage__: - if i >= len(curve__): + if i >= len(global_curve): continue - x__, y__ = curve__[i] + x__, y__ = global_curve[i] # flip y y__ = canvas.height - (y__+1) @@ -4647,9 +4653,9 @@ def main_(f, disk, mroots=None, *, continue # scale from bytes -> pixels usage__ = range( - mt.floor((b.usage.start/block_size) + mt.floor((b.usage.start/block_size_) * (width__*height__)), - mt.ceil((b.usage.stop/block_size) + mt.ceil((b.usage.stop/block_size_) * (width__*height__))) # map to in-block curve for i, (dx, dy) in enumerate(curve(width__, height__)): @@ -4736,6 +4742,7 @@ def main_(f, disk, mroots=None, *, def main(disk, mroots=None, *, height=None, keep_open=False, + lines=None, head=False, cat=False, sleep=False, @@ -4743,23 +4750,44 @@ def main(disk, mroots=None, *, # keep-open? if keep_open: try: + # keep track of history if lines specified + if lines is not None: + ring = RingIO(lines+1 + if not args.get('no_header') and lines > 0 + else lines) while True: # register inotify before running the command, this avoids # modification race conditions if Inotify: inotify = Inotify([disk]) + # TODO sync these comments + # cat? write directly to stdout if cat: main_(sys.stdout, disk, mroots, # make space for shell prompt height=height if height is not False else -1, **args) + # not cat? write to a bounded ring else: - ring = RingIO(head=head) - main_(ring, disk, mroots, + ring_ = RingIO(head=head) + main_(ring_, disk, mroots, height=height if height is not False else 0, **args) - ring.draw() + # no history? draw immediately + if lines is None: + ring_.draw() + # history? merge with previous lines + else: + # write header separately? + if not args.get('no_header'): + if not ring.lines: + ring.lines.append('') + ring.lines.extend(it.islice(ring_.lines, 1, None)) + ring.lines[0] = ring_.lines[0] + else: + ring.lines.extend(ring_.lines) + ring.draw() # try to inotifywait if Inotify: @@ -4952,7 +4980,7 @@ if __name__ == "__main__": '--title-littlefs', action='store_true', help="Use the littlefs mount string as the title.") - # TODO drop padding, no one is ever going to use this + # TODO drop padding in ascii scripts, no one is ever going to use this parser.add_argument( '--padding', type=float, @@ -4965,6 +4993,14 @@ if __name__ == "__main__": '-k', '--keep-open', action='store_true', help="Continue to open and redraw the CSV files in a loop.") + # TODO drop this? + parser.add_argument( + '-n', '--lines', + nargs='?', + type=lambda x: int(x, 0), + const=0, + help="Show this many lines of history. 0 uses the terminal " + "height. Defaults to 1.") parser.add_argument( '-^', '--head', action='store_true', diff --git a/scripts/dbgbmapd3.py b/scripts/dbgbmapd3.py index 89308f40..e44ce38e 100755 --- a/scripts/dbgbmapd3.py +++ b/scripts/dbgbmapd3.py @@ -4721,23 +4721,25 @@ def main(disk, output, mroots=None, *, corrupted = not bool(lfs) # if we can't figure out the block_count, guess + block_size_ = block_size + block_count_ = block_count if block_count is None: if lfs.config.geometry is not None: - block_count = lfs.config.geometry.block_count + block_count_ = lfs.config.geometry.block_count else: f.seek(0, os.SEEK_END) - block_count = mt.ceil(f.tell() / block_size) + block_count_ = mt.ceil(f_.tell() / block_size) # flatten blocks, default to all blocks - blocks = list( - range(blocks.start or 0, blocks.stop or block_count) + blocks_ = list( + range(blocks.start or 0, blocks.stop or block_count_) if isinstance(blocks, slice) else range(blocks, blocks+1) if blocks - else range(block_count)) + else range(block_count_)) # traverse the filesystem and create a block map - bmap = {b: BmapBlock(b, 'unused') for b in blocks} + bmap = {b: BmapBlock(b, 'unused') for b in blocks_} for child, path in lfs.traverse( mtree_only=mtree_only, path=True): @@ -4779,21 +4781,21 @@ def main(disk, output, mroots=None, *, else: bmap[b] = BmapBlock(b, 'conflict', [bmap[b].value, child], - range(block_size)) + range(block_size_)) corrupted = True # corrupt metadata? elif (not no_ckmeta and isinstance(child, (Mdir, Rbyd)) and not child): - bmap[b] = BmapBlock(b, 'corrupt', child, range(block_size)) + bmap[b] = BmapBlock(b, 'corrupt', child, range(block_size_)) corrupted = True # corrupt data? elif (not no_ckdata and isinstance(child, Bptr) and not child): - bmap[b] = BmapBlock(b, 'corrupt', child, range(block_size)) + bmap[b] = BmapBlock(b, 'corrupt', child, range(block_size_)) corrupted = True # normal block @@ -4838,41 +4840,44 @@ def main(disk, output, mroots=None, *, y__ += mt.ceil(FONT_SIZE * 1.3) height__ -= min(mt.ceil(FONT_SIZE * 1.3), height__) - # figure out block_cols/block_rows + # figure out block_cols_/block_rows_ if block_cols is not None and block_rows is not None: - pass + block_cols_ = block_cols + block_rows_ = block_rows elif block_rows is not None: - block_cols = mt.ceil(len(bmap) / block_rows) + block_cols_ = mt.ceil(len(bmap) / block_rows) + block_rows_ = block_rows elif block_cols is not None: - block_rows = mt.ceil(len(bmap) / block_cols) + block_cols_ = block_cols + block_rows_ = mt.ceil(len(bmap) / block_cols) else: # divide by 2 until we hit our target ratio, this works # well for things that are often powers-of-two - block_cols = 1 - block_rows = len(bmap) - while (abs(((width__/(block_cols * 2)) - / (height__/mt.ceil(block_rows / 2))) + block_cols_ = 1 + block_rows_ = len(bmap) + while (abs(((width__/(block_cols_ * 2)) + / (height__/mt.ceil(block_rows_ / 2))) - block_ratio) - < abs(((width__/block_cols) - / (height__/block_rows))) + < abs(((width__/block_cols_) + / (height__/block_rows_))) - block_ratio): - block_cols *= 2 - block_rows = mt.ceil(block_rows / 2) + block_cols_ *= 2 + block_rows_ = mt.ceil(block_rows_ / 2) - block_width = width__ / block_cols - block_height = height__ / block_rows + block_width_ = width__ / block_cols_ + block_height_ = height__ / block_rows_ - # assign block locations based on block_rows/block_cols and the - # requested space filling curve + # assign block locations based on block_rows_/block_cols_ and + # the requested space filling curve for (x, y), b in zip( (hilbert_curve if hilbert else lebesgue_curve if lebesgue - else naive_curve)(block_cols, block_rows), + else naive_curve)(block_cols_, block_rows_), sorted(bmap.values())): - b.x = x__ + (x * block_width) - b.y = y__ + (y * block_height) - b.width = block_width - b.height = block_height + b.x = x__ + (x * block_width_) + b.y = y__ + (y * block_height_) + b.width = block_width_ + b.height = block_height_ # apply top padding if x == 0: diff --git a/scripts/plot.py b/scripts/plot.py index 73d1ec60..b6ce6036 100755 --- a/scripts/plot.py +++ b/scripts/plot.py @@ -1855,7 +1855,7 @@ if __name__ == "__main__": "specific group where a group is the comma-separated " "'by' fields. Accepts %% modifiers.") parser.add_argument( - '-_', '--add-line-char', '--line-chars', + '-,', '--add-line-char', '--line-chars', dest='line_chars', action='append', type=lambda x: ( diff --git a/scripts/tailpipe.py b/scripts/tailpipe.py index 72029558..f5a07959 100755 --- a/scripts/tailpipe.py +++ b/scripts/tailpipe.py @@ -112,17 +112,48 @@ class RingIO: def main(path='-', *, lines=5, cat=False, + coalesce=None, sleep=None, keep_open=False): + lock = th.Lock() + event = th.Event() + + # TODO adopt f -> ring name in all scripts? + def main_(ring): + try: + while True: + with openio(path) as f: + count = 0 + for line in f: + with lock: + ring.write(line) + count += 1 + + # wait for coalesce number of lines + if count >= (coalesce or 1): + event.set() + count = 0 + + if not keep_open: + break + # don't just flood open calls + time.sleep(sleep or 2) + + except FileNotFoundError as e: + print("error: file not found %r" % path, + file=sys.stderr) + sys.exit(-1) + + except KeyboardInterrupt: + pass + + # cat? let main_ write directly to stdout if cat: - ring = sys.stdout + main_(sys.stdout) + + # not cat? print in a background thread else: ring = RingIO(lines) - - # if sleep print in background thread to avoid getting stuck in a read call - event = th.Event() - lock = th.Lock() - if not cat: done = False def background(): while not done: @@ -134,28 +165,13 @@ def main(path='-', *, time.sleep(sleep or 0.01) th.Thread(target=background, daemon=True).start() - try: - while True: - with openio(path) as f: - for line in f: - with lock: - ring.write(line) - event.set() + main_(ring) - if not keep_open: - break - # don't just flood open calls - time.sleep(sleep or 2) - except FileNotFoundError as e: - print("error: file not found %r" % path, - file=sys.stderr) - sys.exit(-1) - except KeyboardInterrupt: - pass - - if not cat: done = True lock.acquire() # avoids https://bugs.python.org/issue42717 + # give ourselves one last draw, helps if background is + # never triggered + ring.draw() sys.stdout.write('\n') @@ -181,10 +197,15 @@ if __name__ == "__main__": '-c', '--cat', action='store_true', help="Pipe directly to stdout.") + parser.add_argument( + '-S', '--coalesce', + type=lambda x: int(x, 0), + help="Number of lines to coalesce together.") parser.add_argument( '-s', '--sleep', type=float, - help="Seconds to sleep between reads.") + help="Seconds to sleep between draws, coalescing lines in " + "between.") parser.add_argument( '-k', '--keep-open', action='store_true', diff --git a/scripts/tracebd.py b/scripts/tracebd.py index 41fddead..483691a6 100755 --- a/scripts/tracebd.py +++ b/scripts/tracebd.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Display operations on block devices based on trace output +# Render operations on block devices based on trace output # # Example: # ./scripts/tracebd.py trace @@ -13,7 +13,9 @@ if __name__ == "__main__": __import__('sys').path.pop(0) +import bisect import collections as co +import fnmatch import functools as ft import io import itertools as it @@ -27,11 +29,37 @@ import threading as th import time -CHARS = 'rpe-' -COLORS = ['32', '35', '34', '90'] +# assign chars/colors to specific bd operations +CHARS = { + 'read': 'r', + 'prog': 'p', + 'erase': 'e', + 'noop': '-', +} +COLORS = { + 'read': '32', + 'prog': '35', + 'erase': '34', + 'noop': '90', +} -WEAR_CHARS = '-123456789' -WEAR_COLORS = ['90', '', '', '', '', '', '', '35', '35', '1;31'] +# assign chars/colors to varying levels of wear +WEAR_CHARS = '0123456789' +# TODO adopt 9x for all of these? +WEAR_COLORS = ['36', '36', '36', '', '', '', '', '31', '31', '91'] + +# give more interesting operations a higher priority +# +# note that while erase is more destructive than prog, all progs subset +# erases, so progs are more interesting +Z_ORDER = ['prog', 'erase', 'read', 'wear', 'noop'] + +# TODO rm +#CHARS = 'rpe-' +#COLORS = ['32', '35', '34', '90'] +# +#WEAR_CHARS = '-123456789' +#WEAR_COLORS = ['90', '', '', '', '', '', '', '35', '35', '1;31'] CHARS_DOTS = " .':" CHARS_BRAILLE = ( @@ -117,6 +145,7 @@ def rbydaddr(s): return tuple(addr) +# TODO negative maxlen from terminal height? like -H nowadays? class RingIO: def __init__(self, maxlen=None, head=False): self.maxlen = maxlen @@ -191,12 +220,394 @@ class RingIO: sys.stdout.write('\x1b[%dB' % (len(lines)-1-i)) sys.stdout.flush() +# TODO sync, espcially the getitem vs get changes +# a representation of optionally key-mapped attrs +class Attr: + def __init__(self, attrs, defaults=None): + if attrs is None: + attrs = [] + if isinstance(attrs, dict): + attrs = attrs.items() + + # normalize + self.attrs = [] + self.keyed = co.OrderedDict() + for attr in attrs: + if (not isinstance(attr, tuple) + or attr[0] in {None, (), (None,), ('*',)}): + attr = ((), attr) + if not isinstance(attr[0], tuple): + attr = ((attr[0],), attr[1]) + + self.attrs.append(attr) + if attr[0] not in self.keyed: + self.keyed[attr[0]] = [] + self.keyed[attr[0]].append(attr[1]) + + # create attrs object for defaults + if isinstance(defaults, Attr): + self.defaults = defaults + elif defaults is not None: + self.defaults = Attr(defaults) + else: + self.defaults = None + + def __repr__(self): + if self.defaults is None: + return 'Attr(%r)' % ( + [(','.join(attr[0]), attr[1]) + for attr in self.attrs]) + else: + return 'Attr(%r, %r)' % ( + [(','.join(attr[0]), attr[1]) + for attr in self.attrs], + [(','.join(attr[0]), attr[1]) + for attr in self.defaults.attrs]) + + def __iter__(self): + if () in self.keyed: + return it.cycle(self.keyed[()]) + elif self.defaults is not None: + return iter(self.defaults) + else: + return iter(()) + + def __bool__(self): + return bool(self.attrs) + + def __getitem__(self, key): + if isinstance(key, tuple): + if len(key) > 0 and not isinstance(key[0], str): + i, key = key + if not isinstance(key, tuple): + key = (key,) + else: + i, key = 0, key + elif isinstance(key, str): + i, key = 0, (key,) + else: + i, key = key, () + + # try to lookup by key + best = None + for ks, vs in self.keyed.items(): + prefix = [] + for j, k in enumerate(ks): + if j < len(key) and fnmatch.fnmatchcase(key[j], k): + prefix.append(k) + else: + prefix = None + break + + if prefix is not None and ( + best is None or len(prefix) >= len(best[0])): + best = (prefix, vs) + + if best is not None: + # cycle based on index + return best[1][i % len(best[1])] + + # fallback to defaults? + if self.defaults is not None: + return self.defaults[i, key] + + raise KeyError(i, key) + + def get(self, key, default=None): + try: + return self.__getitem__(key) + except KeyError: + return default + + def __contains__(self, key): + try: + self.__getitem__(key) + return True + except KeyError: + return False + + # get all results for a given key + def getall(self, key, default=None): + if not isinstance(key, tuple): + key = (key,) + + # try to lookup by key + best = None + for ks, vs in self.keyed.items(): + prefix = [] + for j, k in enumerate(ks): + if j < len(key) and fnmatch.fnmatchcase(key[j], k): + prefix.append(k) + else: + prefix = None + break + + if prefix is not None and ( + best is None or len(prefix) >= len(best[0])): + best = (prefix, vs) + + if best is not None: + return best[1] + + # fallback to defaults? + if self.defaults is not None: + return self.defaults.getall(key, default) + + raise default + + # a key function for sorting by key order + def key(self, key): + if not isinstance(key, tuple): + key = (key,) + + best = None + for i, ks in enumerate(self.keyed.keys()): + prefix = [] + for j, k in enumerate(ks): + if j < len(key) and (not k or key[j] == k): + prefix.append(k) + else: + prefix = None + break + + if prefix is not None and ( + best is None or len(prefix) >= len(best[0])): + best = (prefix, i) + + if best is not None: + return best[1] + + # fallback to defaults? + if self.defaults is not None: + return len(self.keyed) + self.defaults.key(key) + + return len(self.keyed) + +# parse %-escaped strings +def punescape(s, attrs=None): + # TODO punescape should just take something that provides __getitem__ + if attrs is None: + attrs = {} + if isinstance(attrs, dict): + attrs_ = attrs + attrs = lambda k: attrs_[k] + + pattern = re.compile( + '%[%n]' + '|' '%x..' + '|' '%u....' + '|' '%U........' + '|' '%\((?P[^)]*)\)' + '(?P[+\- #0-9\.]*[sdboxXfFeEgG])') + def unescape(m): + if m.group()[1] == '%': return '%' + elif m.group()[1] == 'n': return '\n' + elif m.group()[1] == 'x': return chr(int(m.group()[2:], 16)) + elif m.group()[1] == 'u': return chr(int(m.group()[2:], 16)) + elif m.group()[1] == 'U': return chr(int(m.group()[2:], 16)) + elif m.group()[1] == '(': + try: + v = attrs(m.group('field')) + except KeyError: + return m.group() + f = m.group('format') + if f[-1] in 'dboxX': + if isinstance(v, str): + v = dat(v, 0) + v = int(v) + elif f[-1] in 'fFeEgG': + if isinstance(v, str): + v = dat(v, 0) + v = float(v) + else: + f = ('<' if '-' in f else '>') + f.replace('-', '') + v = str(v) + # note we need Python's new format syntax for binary + return ('{:%s}' % f).format(v) + else: assert False + return re.sub(pattern, unescape, s) + +# split %-escaped strings into chars +def psplit(s): + pattern = re.compile( + '%[%n]' + '|' '%x..' + '|' '%u....' + '|' '%U........' + '|' '%\((?P[^)]*)\)' + '(?P[+\- #0-9\.]*[sdboxXfFeEgG])') + return [m.group() for m in re.finditer(pattern.pattern + '|.', s)] + + +# a little ascii renderer +class Canvas: + def __init__(self, width, height, *, + color=False, + dots=False, + braille=False): + # scale if we're printing with dots or braille + if braille: + xscale, yscale = 2, 4 + elif dots: + xscale, yscale = 1, 2 + else: + xscale, yscale = 1, 1 + + self.width_ = width + self.height_ = height + self.width = xscale*width + self.height = yscale*height + self.xscale = xscale + self.yscale = yscale + self.color_ = color + self.dots = dots + self.braille = braille + + # create initial canvas + self.chars = [0] * (width*height) + self.colors = [''] * (width*height) + + def char(self, x, y, char=None): + # ignore out of bounds + if x < 0 or y < 0 or x >= self.width or y >= self.height: + return False + + x_ = x // self.xscale + y_ = y // self.yscale + if char is not None: + c = self.chars[x_ + y_*self.width_] + # mask in sub-char pixel? + if isinstance(char, bool): + if not isinstance(c, int): + c = 0 + self.chars[x_ + y_*self.width_] = (c + | (1 + << ((y%self.yscale)*self.xscale + + (self.xscale-1)-(x%self.xscale)))) + else: + self.chars[x_ + y_*self.width_] = char + else: + c = self.chars[x_ + y_*self.width_] + if isinstance(c, int): + return ((c + >> ((y%self.yscale)*self.xscale + + (self.xscale-1)-(x%self.xscale))) + & 1) == 1 + else: + return c + + def color(self, x, y, color=None): + # ignore out of bounds + if x < 0 or y < 0 or x >= self.width or y >= self.height: + return '' + + x_ = x // self.xscale + y_ = y // self.yscale + if color is not None: + self.colors[x_ + y_*self.width_] = color + else: + return self.colors[x_ + y_*self.width_] + + def __getitem__(self, xy): + x, y = xy + return self.char(x, y) + + def __setitem__(self, xy, char): + x, y = xy + self.char(x, y, char) + + def point(self, x, y, *, + char=True, + color=''): + self.char(x, y, char) + self.color(x, y, color) + + def line(self, x1, y1, x2, y2, *, + char=True, + color=''): + # incremental error line algorithm + ex = abs(x2 - x1) + ey = -abs(y2 - y1) + dx = +1 if x1 < x2 else -1 + dy = +1 if y1 < y2 else -1 + e = ex + ey + + while True: + self.point(x1, y1, char=char, color=color) + e2 = 2*e + + if x1 == x2 and y1 == y2: + break + + if e2 > ey: + e += ey + x1 += dx + + if x1 == x2 and y1 == y2: + break + + if e2 < ex: + e += ex + y1 += dy + + self.point(x2, y2, char=char, color=color) + + def rect(self, x, y, w, h, *, + char=True, + color=''): + for j in range(h): + for i in range(w): + self.point(x+i, y+j, char=char, color=color) + + def label(self, x, y, label, width=None, height=None, *, + color=''): + x_ = x + y_ = y + for char in label: + if char == '\n': + x_ = x + y_ -= self.yscale + else: + if ((width is None or x_ < x+width) + and (height is None or y_ > y-height)): + self.point(x_, y_, char=char, color=color) + x_ += self.xscale + + def draw(self, row): + y_ = self.height_-1 - row + row_ = [] + for x_ in range(self.width_): + # char? + c = self.chars[x_ + y_*self.width_] + if isinstance(c, int): + if self.braille: + assert c < 256 + c = CHARS_BRAILLE[c] + elif self.dots: + assert c < 4 + c = CHARS_DOTS[c] + else: + assert c < 2 + c = '.' if c else ' ' + + # color? + if self.color_: + color = self.colors[x_ + y_*self.width_] + if color: + c = '\x1b[%sm%s\x1b[m' % (color, c) + + row_.append(c) + + return ''.join(row_) + + +# naive space filling curve (the default) +def naive_curve(width, height): + for y in range(height): + for x in range(width): + yield x, y # space filling Hilbert-curve -# -# note we memoize the last curve since this is a bit expensive -# -@ft.lru_cache(1) def hilbert_curve(width, height): # based on generalized Hilbert curves: # https://github.com/jakubcerveny/gilbert @@ -212,14 +623,14 @@ def hilbert_curve(width, height): # trivial row if h == 1: for _ in range(w): - yield (x,y) + yield x, y x, y = x+a_dx, y+a_dy return # trivial column if w == 1: for _ in range(h): - yield (x,y) + yield x, y x, y = x+b_dx, y+b_dy return @@ -234,448 +645,769 @@ def hilbert_curve(width, height): a_x_, a_y_ = a_x_+a_dx, a_y_+a_dy # split in two - yield from hilbert_(x, y, a_x_, a_y_, b_x, b_y) - yield from hilbert_(x+a_x_, y+a_y_, a_x-a_x_, a_y-a_y_, b_x, b_y) + yield from hilbert_( + x, y, + a_x_, a_y_, b_x, b_y) + yield from hilbert_( + x+a_x_, y+a_y_, + a_x-a_x_, a_y-a_y_, b_x, b_y) else: # prefer even steps if h_ % 2 != 0 and h > 2: b_x_, b_y_ = b_x_+b_dx, b_y_+b_dy # split in three - yield from hilbert_(x, y, b_x_, b_y_, a_x_, a_y_) - yield from hilbert_(x+b_x_, y+b_y_, a_x, a_y, b_x-b_x_, b_y-b_y_) + yield from hilbert_( + x, y, + b_x_, b_y_, a_x_, a_y_) + yield from hilbert_( + x+b_x_, y+b_y_, + a_x, a_y, b_x-b_x_, b_y-b_y_) yield from hilbert_( x+(a_x-a_dx)+(b_x_-b_dx), y+(a_y-a_dy)+(b_y_-b_dy), -b_x_, -b_y_, -(a_x-a_x_), -(a_y-a_y_)) if width >= height: - curve = hilbert_(0, 0, +width, 0, 0, +height) + yield from hilbert_(0, 0, +width, 0, 0, +height) else: - curve = hilbert_(0, 0, 0, +height, +width, 0) - - return list(curve) + yield from hilbert_(0, 0, 0, +height, +width, 0) # space filling Z-curve/Lebesgue-curve -# -# note we memoize the last curve since this is a bit expensive -# -@ft.lru_cache(1) def lebesgue_curve(width, height): - # we create a truncated Z-curve by simply filtering out the points - # that are outside our region - curve = [] + # we create a truncated Z-curve by simply filtering out the + # points that are outside our region for i in range(2**(2*mt.ceil(mt.log2(max(width, height))))): # we just operate on binary strings here because it's easier b = '{:0{}b}'.format(i, 2*mt.ceil(mt.log2(i+1)/2)) x = int(b[1::2], 2) if b[1::2] else 0 y = int(b[0::2], 2) if b[0::2] else 0 if x < width and y < height: - curve.append((x, y)) - - return curve + yield x, y -class Pixel(int): - __slots__ = () - def __new__(cls, state=0, *, - wear=0, - readed=False, - proged=False, - erased=False): - return super().__new__(cls, - state - | (wear << 3) - | (1 if readed else 0) - | (2 if proged else 0) - | (4 if erased else 0)) +# a mergable range set type +class RangeSet: + def __init__(self, ranges=None): + self._ranges = [] - @property - def wear(self): - return self >> 3 + if ranges is not None: + # using add here makes sure all ranges are merged/sorted + # correctly + for r in ranges: + self.add(r) - @property - def readed(self): - return (self & 1) != 0 + def __repr__(self): + return 'RangeSet(%r)' % self._ranges - @property - def proged(self): - return (self & 2) != 0 - - @property - def erased(self): - return (self & 4) != 0 - - def read(self): - return Pixel(int(self) | 1) - - def prog(self): - return Pixel(int(self) | 2) - - def erase(self): - return Pixel((int(self) | 4) + 8) - - def clear(self): - return Pixel(int(self) & ~7) - - def __or__(self, other): - return Pixel( - (int(self) | int(other)) & 7, - wear=max(self.wear, other.wear)) - - def worn(self, max_wear, *, - block_cycles=None, - wear_chars=None, - **_): - if wear_chars is None: - wear_chars = WEAR_CHARS - - if block_cycles: - return self.wear / block_cycles + def __contains__(self, k): + i = bisect.bisect(self._ranges, k, + key=lambda r: r.start) - 1 + if i > -1: + return k in self._ranges[i] else: - return self.wear / max(max_wear, len(wear_chars)) + return False - def draw(self, max_wear, char=None, *, - reads=True, - progs=True, - erases=True, - wear=False, - block_cycles=None, - color=True, - dots=False, - braille=False, - chars=None, - wear_chars=None, - colors=None, - wear_colors=None, - **_): - # fallback to default chars/colors - if chars is None: - chars = CHARS - if len(chars) < len(CHARS): - chars = chars + CHARS[len(chars):] + def __bool__(self): + return bool(self._ranges) - if colors is None: - colors = COLORS - if len(colors) < len(COLORS): - colors = colors + COLORS[len(colors):] + def ranges(self): + yield from self._ranges - if wear_chars is None: - wear_chars = WEAR_CHARS + def __iter__(self): + for r in self._ranges: + yield from r - if wear_colors is None: - wear_colors = WEAR_COLORS - - # compute char/color - c = chars[3] - f = [colors[3]] - - if wear: - w = min(self.worn( - max_wear, - block_cycles=block_cycles, - wear_chars=wear_chars), - 1) - - c = wear_chars[int(w * (len(wear_chars)-1))] - f.append(wear_colors[int(w * (len(wear_colors)-1))]) - - if progs and self.proged: - c = chars[1] - f.append(colors[1]) - elif erases and self.erased: - c = chars[2] - f.append(colors[2]) - elif reads and self.readed: - c = chars[0] - f.append(colors[0]) - - # override char? - if char: - c = char - - # apply colors - if f and color: - c = '%s%s\x1b[m' % ( - ''.join('\x1b[%sm' % f_ for f_ in f), - c) - - return c - - -class Bmap: - def __init__(self, *, - block_size=1, - block_count=1, - block_window=None, - off_window=None, - width=1, - height=1, - pixels=None): - # default width to block_window or block_size - if width is None: - if block_window is not None: - width = len(block_window) - else: - width = block_count - - # allocate pixels if not provided - if pixels is None: - pixels = [Pixel() for _ in range(width*height)] - - self.pixels = pixels - self.block_size = block_size - self.block_count = block_count - self.block_window = block_window - self.off_window = off_window - self.width = width - self.height = height - - @property - def _block_window(self): - if self.block_window is None: - return range(0, self.block_count) - else: - return self.block_window - - @property - def _off_window(self): - if self.off_window is None: - return range(0, self.block_size) - else: - return self.off_window - - @property - def _window(self): - return len(self._off_window)*len(self._block_window) - - def _op(self, f, block=None, off=None, size=None): - if block is None: - range_ = range(len(self.pixels)) - else: - if off is None: - off, size = 0, self.block_size - elif size is None: - off, size = 0, off - - # map into our window - if block not in self._block_window: - return - block -= self._block_window.start - - size = (max(self._off_window.start, - min(self._off_window.stop, off+size)) - - max(self._off_window.start, - min(self._off_window.stop, off))) - off = (max(self._off_window.start, - min(self._off_window.stop, off)) - - self._off_window.start) - if size == 0: - return - - # map to our block space - range_ = range( - block*len(self._off_window) + off, - block*len(self._off_window) + off+size) - range_ = range( - (range_.start*len(self.pixels)) // self._window, - (range_.stop*len(self.pixels)) // self._window) - range_ = range( - range_.start, - max(range_.stop, range_.start+1)) - - # apply the op - for i in range_: - self.pixels[i] = f(self.pixels[i]) - - def read(self, block=None, off=None, size=None): - self._op(Pixel.read, block, off, size) - - def prog(self, block=None, off=None, size=None): - self._op(Pixel.prog, block, off, size) - - def erase(self, block=None, off=None, size=None): - self._op(Pixel.erase, block, off, size) - - def clear(self, block=None, off=None, size=None): - self._op(Pixel.clear, block, off, size) - - def resize(self, *, - block_size=None, - block_count=None, - width=None, - height=None): - block_size = (block_size if block_size is not None - else self.block_size) - block_count = (block_count if block_count is not None - else self.block_count) - width = width if width is not None else self.width - height = height if height is not None else self.height - - if (block_size == self.block_size - and block_count == self.block_count - and width == self.width - and height == self.height): + def add(self, r): + assert isinstance(r, range) + # trivial range? + if not r: return - # transform our pixels - self.block_size = block_size - self.block_count = block_count + # find earliest possible merge point + ranges = self._ranges + i = bisect.bisect_left(ranges, r.start, + key=lambda r: r.stop) - pixels = [] - for x in range(width*height): - # map into our old bd space - range_ = range( - (x*self._window) // (width*height), - ((x+1)*self._window) // (width*height)) - range_ = range( - range_.start, - max(range_.stop, range_.start+1)) + # copy ranges < merge + merged = ranges[:i] - # aggregate state - pixels.append(ft.reduce( - Pixel.__or__, - self.pixels[range_.start:range_.stop], - Pixel())) - + # merge ranges and append + while i < len(ranges) and ranges[i].start <= r.stop: + r = range( + min(ranges[i].start, r.start), + max(ranges[i].stop, r.stop)) + i += 1 + merged.append(r) + + # copy ranges > merge + merged.extend(ranges[i:]) + + self._ranges = merged + + def remove(self, r): + assert isinstance(r, range) + # trivial range? + if not r: + return + + # find earliest possible carve point + ranges = self._ranges + i = bisect.bisect_left(ranges, r.start, + key=lambda r: r.stop) + + # copy ranges < carve + carved = ranges[:i] + + # carve overlapping ranges, note this can split ranges + while i < len(ranges) and ranges[i].start <= r.stop: + if ranges[i].start < r.start: + carved.append(range(ranges[i].start, r.start)) + if ranges[i].stop > r.stop: + carved.append(range(r.stop, ranges[i].stop)) + i += 1 + + # copy ranges > carve + carved.extend(ranges[i:]) + + self._ranges = carved + + @property + def start(self): + if not self._ranges: + return 0 + else: + return self._ranges[0].start + + @property + def stop(self): + if not self._ranges: + return 0 + else: + return self._ranges[-1].stop + + def __len__(self): + return self.stop + + def copy(self): + # create a shallow copy + ranges = RangeSet() + ranges._ranges = self._ranges.copy() + return ranges + + def __getitem__(self, slice_): + assert isinstance(slice_, slice) + + # create a copy + ranges = self.copy() + + # just use remove to do the carving, it's good enough probably + if slice_.stop is not None: + ranges.remove(range(slice_.stop, len(self))) + if slice_.start is not None: + ranges.remove(range(0, slice_.start)) + ranges._ranges = [range( + r.start - slice_.start, + r.stop - slice_.start) + for r in ranges._ranges] + + return ranges + + def __ior__(self, other): + for r in other.ranges(): + self.add(r) + return self + + def __or__(self, other): + ranges = self.copy() + ranges |= other + return ranges + + +# an abstract block representation +class TraceBlock: + def __init__(self, block, *, + readed=None, proged=None, erased=None, wear=0, + x=None, y=None, width=None, height=None): + self.block = block + self.readed = readed if readed is not None else RangeSet() + self.proged = proged if proged is not None else RangeSet() + self.erased = erased if erased is not None else RangeSet() + self.wear = wear + self.x = x + self.y = y self.width = width self.height = height - self.pixels = pixels - def draw(self, row, *, - reads=False, - progs=False, - erases=False, - wear=False, - hilbert=False, - lebesgue=False, - dots=False, - braille=False, - **args): - # find max wear? - max_wear = None - if wear: - max_wear = max(p.wear for p in self.pixels) + def __repr__(self): + return 'TraceBlock(0x%x, x=%s, y=%s, width=%s, height=%s)' % ( + self.block, + self.x, self.y, self.width, self.height) - # fold via a curve? - if hilbert: - grid = [None]*(self.width*self.height) - for (x,y), p in zip( - hilbert_curve(self.width, self.height), - self.pixels): - grid[x + y*self.width] = p - elif lebesgue: - grid = [None]*(self.width*self.height) - for (x,y), p in zip( - lebesgue_curve(self.width, self.height), - self.pixels): - grid[x + y*self.width] = p - else: - grid = self.pixels + def __eq__(self, other): + return self.block == other.block - # need to wait for more trace output before rendering - # - # this is sort of a hack that knows the output is going to a terminal - if (braille and self.height < 4) or (dots and self.height < 2): - needed_height = 4 if braille else 2 + def __ne__(self, other): + return self.block != other.block - self.history = getattr(self, 'history', []) - self.history.append(grid.copy()) + def __hash__(self): + return hash(self.block) - if len(self.history)*self.height < needed_height: - # skip for now - return None + def __lt__(self, other): + return self.block < other.block - grid = list(it.chain.from_iterable( - # did we resize? - it.islice(it.chain(h, it.repeat(Pixel())), - self.width*self.height) - for h in self.history)) - self.history = [] + def __le__(self, other): + return self.block <= other.block - line = [] - if braille: - # encode into a byte - for x in range(0, self.width, 2): - byte_p = 0 - best_p = Pixel() - for i in range(2*4): - p = grid[x+(2-1-(i%2)) + ((row*4)+(4-1-(i//2)))*self.width] - best_p |= p - if ((reads and p.readed) - or (progs and p.proged) - or (erases and p.erased) - or (not reads and not progs and not erases - and wear and p.worn(max_wear, **args) >= 0.7)): - byte_p |= 1 << i + def __gt__(self, other): + return self.block > other.block - line.append(best_p.draw( - max_wear, - CHARS_BRAILLE[byte_p], - braille=True, - reads=reads, - progs=progs, - erases=erases, - wear=wear, - **args)) - elif dots: - # encode into a byte - for x in range(self.width): - byte_p = 0 - best_p = Pixel() - for i in range(2): - p = grid[x + ((row*2)+(2-1-i))*self.width] - best_p |= p - if ((reads and p.readed) - or (progs and p.proged) - or (erases and p.erased) - or (not reads and not progs and not erases - and wear and p.worn(max_wear, **args) >= 0.7)): - byte_p |= 1 << i + def __ge__(self, other): + return self.block >= other.block - line.append(best_p.draw( - max_wear, - CHARS_DOTS[byte_p], - dots=True, - reads=reads, - progs=progs, - erases=erases, - wear=wear, - **args)) - else: - for x in range(self.width): - line.append(grid[x + row*self.width].draw( - max_wear, - reads=reads, - progs=progs, - erases=erases, - wear=wear, - **args)) + # align to pixel boundaries + def align(self): + # this extra +0.1 and using points instead of width/height is + # to help minimize rounding errors + x0 = int(self.x+0.1) + y0 = int(self.y+0.1) + x1 = int(self.x+self.width+0.1) + y1 = int(self.y+self.height+0.1) + self.x = x0 + self.y = y0 + self.width = x1 - x0 + self.height = y1 - y0 - return ''.join(line) + # generate attrs for punescaping + @ft.cached_property + def attrs(self): + # really the only reasonable attrs are block and wear + return { + 'block': self.block, + 'wear': self.wear, + } + + # some simulated bd operations + def read(self, off, size): + self.readed.add(range(off, off+size)) + + def prog(self, off, size): + self.proged.add(range(off, off+size)) + + def erase(self, off, size, *, + wear=0): + self.erased.add(range(off, off+size)) + self.wear += wear + + def clear(self): + self.readed = RangeSet() + self.proged = RangeSet() + self.erased = RangeSet() + + + + + + + +## space filling Hilbert-curve +## +## note we memoize the last curve since this is a bit expensive +## +#@ft.lru_cache(1) +#def hilbert_curve(width, height): +# # based on generalized Hilbert curves: +# # https://github.com/jakubcerveny/gilbert +# # +# def hilbert_(x, y, a_x, a_y, b_x, b_y): +# w = abs(a_x+a_y) +# h = abs(b_x+b_y) +# a_dx = -1 if a_x < 0 else +1 if a_x > 0 else 0 +# a_dy = -1 if a_y < 0 else +1 if a_y > 0 else 0 +# b_dx = -1 if b_x < 0 else +1 if b_x > 0 else 0 +# b_dy = -1 if b_y < 0 else +1 if b_y > 0 else 0 +# +# # trivial row +# if h == 1: +# for _ in range(w): +# yield (x,y) +# x, y = x+a_dx, y+a_dy +# return +# +# # trivial column +# if w == 1: +# for _ in range(h): +# yield (x,y) +# x, y = x+b_dx, y+b_dy +# return +# +# a_x_, a_y_ = a_x//2, a_y//2 +# b_x_, b_y_ = b_x//2, b_y//2 +# w_ = abs(a_x_+a_y_) +# h_ = abs(b_x_+b_y_) +# +# if 2*w > 3*h: +# # prefer even steps +# if w_ % 2 != 0 and w > 2: +# a_x_, a_y_ = a_x_+a_dx, a_y_+a_dy +# +# # split in two +# yield from hilbert_(x, y, a_x_, a_y_, b_x, b_y) +# yield from hilbert_(x+a_x_, y+a_y_, a_x-a_x_, a_y-a_y_, b_x, b_y) +# else: +# # prefer even steps +# if h_ % 2 != 0 and h > 2: +# b_x_, b_y_ = b_x_+b_dx, b_y_+b_dy +# +# # split in three +# yield from hilbert_(x, y, b_x_, b_y_, a_x_, a_y_) +# yield from hilbert_(x+b_x_, y+b_y_, a_x, a_y, b_x-b_x_, b_y-b_y_) +# yield from hilbert_( +# x+(a_x-a_dx)+(b_x_-b_dx), y+(a_y-a_dy)+(b_y_-b_dy), +# -b_x_, -b_y_, -(a_x-a_x_), -(a_y-a_y_)) +# +# if width >= height: +# curve = hilbert_(0, 0, +width, 0, 0, +height) +# else: +# curve = hilbert_(0, 0, 0, +height, +width, 0) +# +# return list(curve) +# +## space filling Z-curve/Lebesgue-curve +## +## note we memoize the last curve since this is a bit expensive +## +#@ft.lru_cache(1) +#def lebesgue_curve(width, height): +# # we create a truncated Z-curve by simply filtering out the points +# # that are outside our region +# curve = [] +# for i in range(2**(2*mt.ceil(mt.log2(max(width, height))))): +# # we just operate on binary strings here because it's easier +# b = '{:0{}b}'.format(i, 2*mt.ceil(mt.log2(i+1)/2)) +# x = int(b[1::2], 2) if b[1::2] else 0 +# y = int(b[0::2], 2) if b[0::2] else 0 +# if x < width and y < height: +# curve.append((x, y)) +# +# return curve +# +# +#class Pixel(int): +# __slots__ = () +# def __new__(cls, state=0, *, +# wear=0, +# readed=False, +# proged=False, +# erased=False): +# return super().__new__(cls, +# state +# | (wear << 3) +# | (1 if readed else 0) +# | (2 if proged else 0) +# | (4 if erased else 0)) +# +# @property +# def wear(self): +# return self >> 3 +# +# @property +# def readed(self): +# return (self & 1) != 0 +# +# @property +# def proged(self): +# return (self & 2) != 0 +# +# @property +# def erased(self): +# return (self & 4) != 0 +# +# def read(self): +# return Pixel(int(self) | 1) +# +# def prog(self): +# return Pixel(int(self) | 2) +# +# def erase(self): +# return Pixel((int(self) | 4) + 8) +# +# def clear(self): +# return Pixel(int(self) & ~7) +# +# def __or__(self, other): +# return Pixel( +# (int(self) | int(other)) & 7, +# wear=max(self.wear, other.wear)) +# +# def worn(self, max_wear, *, +# block_cycles=None, +# wear_chars=None, +# **_): +# if wear_chars is None: +# wear_chars = WEAR_CHARS +# +# if block_cycles: +# return self.wear / block_cycles +# else: +# return self.wear / max(max_wear, len(wear_chars)) +# +# def draw(self, max_wear, char=None, *, +# reads=True, +# progs=True, +# erases=True, +# wear=False, +# block_cycles=None, +# color=True, +# dots=False, +# braille=False, +# chars=None, +# wear_chars=None, +# colors=None, +# wear_colors=None, +# **_): +# # fallback to default chars/colors +# if chars is None: +# chars = CHARS +# if len(chars) < len(CHARS): +# chars = chars + CHARS[len(chars):] +# +# if colors is None: +# colors = COLORS +# if len(colors) < len(COLORS): +# colors = colors + COLORS[len(colors):] +# +# if wear_chars is None: +# wear_chars = WEAR_CHARS +# +# if wear_colors is None: +# wear_colors = WEAR_COLORS +# +# # compute char/color +# c = chars[3] +# f = [colors[3]] +# +# if wear: +# w = min(self.worn( +# max_wear, +# block_cycles=block_cycles, +# wear_chars=wear_chars), +# 1) +# +# c = wear_chars[int(w * (len(wear_chars)-1))] +# f.append(wear_colors[int(w * (len(wear_colors)-1))]) +# +# if progs and self.proged: +# c = chars[1] +# f.append(colors[1]) +# elif erases and self.erased: +# c = chars[2] +# f.append(colors[2]) +# elif reads and self.readed: +# c = chars[0] +# f.append(colors[0]) +# +# # override char? +# if char: +# c = char +# +# # apply colors +# if f and color: +# c = '%s%s\x1b[m' % ( +# ''.join('\x1b[%sm' % f_ for f_ in f), +# c) +# +# return c +# +# +#class Bmap: +# def __init__(self, *, +# block_size=1, +# block_count=1, +# block_window=None, +# off_window=None, +# width=1, +# height=1, +# pixels=None): +# # default width to block_window or block_size +# if width is None: +# if block_window is not None: +# width = len(block_window) +# else: +# width = block_count +# +# # allocate pixels if not provided +# if pixels is None: +# pixels = [Pixel() for _ in range(width*height)] +# +# self.pixels = pixels +# self.block_size = block_size +# self.block_count = block_count +# self.block_window = block_window +# self.off_window = off_window +# self.width = width +# self.height = height +# +# @property +# def _block_window(self): +# if self.block_window is None: +# return range(0, self.block_count) +# else: +# return self.block_window +# +# @property +# def _off_window(self): +# if self.off_window is None: +# return range(0, self.block_size) +# else: +# return self.off_window +# +# @property +# def _window(self): +# return len(self._off_window)*len(self._block_window) +# +# def _op(self, f, block=None, off=None, size=None): +# if block is None: +# range_ = range(len(self.pixels)) +# else: +# if off is None: +# off, size = 0, self.block_size +# elif size is None: +# off, size = 0, off +# +# # map into our window +# if block not in self._block_window: +# return +# block -= self._block_window.start +# +# size = (max(self._off_window.start, +# min(self._off_window.stop, off+size)) +# - max(self._off_window.start, +# min(self._off_window.stop, off))) +# off = (max(self._off_window.start, +# min(self._off_window.stop, off)) +# - self._off_window.start) +# if size == 0: +# return +# +# # map to our block space +# range_ = range( +# block*len(self._off_window) + off, +# block*len(self._off_window) + off+size) +# range_ = range( +# (range_.start*len(self.pixels)) // self._window, +# (range_.stop*len(self.pixels)) // self._window) +# range_ = range( +# range_.start, +# max(range_.stop, range_.start+1)) +# +# # apply the op +# for i in range_: +# self.pixels[i] = f(self.pixels[i]) +# +# def read(self, block=None, off=None, size=None): +# self._op(Pixel.read, block, off, size) +# +# def prog(self, block=None, off=None, size=None): +# self._op(Pixel.prog, block, off, size) +# +# def erase(self, block=None, off=None, size=None): +# self._op(Pixel.erase, block, off, size) +# +# def clear(self, block=None, off=None, size=None): +# self._op(Pixel.clear, block, off, size) +# +# def resize(self, *, +# block_size=None, +# block_count=None, +# width=None, +# height=None): +# block_size = (block_size if block_size is not None +# else self.block_size) +# block_count = (block_count if block_count is not None +# else self.block_count) +# width = width if width is not None else self.width +# height = height if height is not None else self.height +# +# if (block_size == self.block_size +# and block_count == self.block_count +# and width == self.width +# and height == self.height): +# return +# +# # transform our pixels +# self.block_size = block_size +# self.block_count = block_count +# +# pixels = [] +# for x in range(width*height): +# # map into our old bd space +# range_ = range( +# (x*self._window) // (width*height), +# ((x+1)*self._window) // (width*height)) +# range_ = range( +# range_.start, +# max(range_.stop, range_.start+1)) +# +# # aggregate state +# pixels.append(ft.reduce( +# Pixel.__or__, +# self.pixels[range_.start:range_.stop], +# Pixel())) +# +# self.width = width +# self.height = height +# self.pixels = pixels +# +# def draw(self, row, *, +# reads=False, +# progs=False, +# erases=False, +# wear=False, +# hilbert=False, +# lebesgue=False, +# dots=False, +# braille=False, +# **args): +# # find max wear? +# max_wear = None +# if wear: +# max_wear = max(p.wear for p in self.pixels) +# +# # fold via a curve? +# if hilbert: +# grid = [None]*(self.width*self.height) +# for (x,y), p in zip( +# hilbert_curve(self.width, self.height), +# self.pixels): +# grid[x + y*self.width] = p +# elif lebesgue: +# grid = [None]*(self.width*self.height) +# for (x,y), p in zip( +# lebesgue_curve(self.width, self.height), +# self.pixels): +# grid[x + y*self.width] = p +# else: +# grid = self.pixels +# +# # need to wait for more trace output before rendering +# # +# # this is sort of a hack that knows the output is going to a terminal +# if (braille and self.height < 4) or (dots and self.height < 2): +# needed_height = 4 if braille else 2 +# +# self.history = getattr(self, 'history', []) +# self.history.append(grid.copy()) +# +# if len(self.history)*self.height < needed_height: +# # skip for now +# return None +# +# grid = list(it.chain.from_iterable( +# # did we resize? +# it.islice(it.chain(h, it.repeat(Pixel())), +# self.width*self.height) +# for h in self.history)) +# self.history = [] +# +# line = [] +# if braille: +# # encode into a byte +# for x in range(0, self.width, 2): +# byte_p = 0 +# best_p = Pixel() +# for i in range(2*4): +# p = grid[x+(2-1-(i%2)) + ((row*4)+(4-1-(i//2)))*self.width] +# best_p |= p +# if ((reads and p.readed) +# or (progs and p.proged) +# or (erases and p.erased) +# or (not reads and not progs and not erases +# and wear and p.worn(max_wear, **args) >= 0.7)): +# byte_p |= 1 << i +# +# line.append(best_p.draw( +# max_wear, +# CHARS_BRAILLE[byte_p], +# braille=True, +# reads=reads, +# progs=progs, +# erases=erases, +# wear=wear, +# **args)) +# elif dots: +# # encode into a byte +# for x in range(self.width): +# byte_p = 0 +# best_p = Pixel() +# for i in range(2): +# p = grid[x + ((row*2)+(2-1-i))*self.width] +# best_p |= p +# if ((reads and p.readed) +# or (progs and p.proged) +# or (erases and p.erased) +# or (not reads and not progs and not erases +# and wear and p.worn(max_wear, **args) >= 0.7)): +# byte_p |= 1 << i +# +# line.append(best_p.draw( +# max_wear, +# CHARS_DOTS[byte_p], +# dots=True, +# reads=reads, +# progs=progs, +# erases=erases, +# wear=wear, +# **args)) +# else: +# for x in range(self.width): +# line.append(grid[x + row*self.width].draw( +# max_wear, +# reads=reads, +# progs=progs, +# erases=erases, +# wear=wear, +# **args)) +# +# return ''.join(line) def main(path='-', *, block_size=None, block_count=None, - block_cycles=None, - block=None, + blocks=None, reads=False, progs=False, erases=False, wear=False, - reset=False, - no_header=False, + wear_only=False, + block_cycles=None, + volatile=False, + chars=[], + wear_chars=[], + colors=[], + wear_colors=[], + # TODO this should be color='auto' in all scripts color='auto', dots=False, braille=False, width=None, height=None, + block_cols=None, + block_rows=None, + block_ratio=None, + no_header=False, + hilbert=False, + lebesgue=False, + contiguous=False, + to_scale=None, + aspect_ratio=(1,1), + tiny=False, + title=None, + padding=0, lines=None, head=False, cat=False, - hilbert=False, - lebesgue=False, coalesce=None, sleep=None, keep_open=False, @@ -688,28 +1420,56 @@ def main(path='-', *, else: color = False - # exclusive wear or reads/progs/erases by default - if not reads and not progs and not erases and not wear: - reads = True + # if not specified default to all ops + if not reads and not progs and not erases and not wear_only: progs = True + reads = True erases = True - # assume a reasonable lines/height if not specified - # - # note that we let height = None if neither hilbert or lebesgue - # are specified, this is a bit special as the default may be less - # than one character in height. - if height is None and (hilbert or lebesgue): - if lines is not None: - height = lines - else: - height = 5 + # wear_only implies only wear + if wear_only: + wear = True - if lines is None: - if height is not None: - lines = height + # block_cycles implies wear + if block_cycles is not None: + wear = True + + # tiny mode? + if tiny: + if block_ratio is None: + block_ratio = 1 + if to_scale is None: + to_scale = 1 + no_header = True + + if block_ratio is None: + # try to align block_ratio to chars, even in braille/dots + # mode (we can't color sub-chars) + if braille or dots: + block_ratio = 1/2 else: - lines = 5 + block_ratio = 1 + + # what chars/colors/labels to use? + chars_ = [] + for char in chars: + if isinstance(char, tuple): + chars_.extend((char[0], c) for c in psplit(char[1])) + else: + chars_.extend(psplit(char)) + chars_ = Attr(chars_, defaults=CHARS) + + wear_chars_ = [] + for char in wear_chars: + if isinstance(char, tuple): + wear_chars_.extend((char[0], c) for c in psplit(char[1])) + else: + wear_chars_.extend(psplit(char)) + wear_chars_ = Attr(wear_chars_, defaults=WEAR_CHARS) + + colors_ = Attr(colors, defaults=COLORS) + + wear_colors_ = Attr(wear_colors, defaults=WEAR_COLORS) # is bd geometry specified? if isinstance(block_size, tuple): @@ -717,356 +1477,1080 @@ def main(path='-', *, if block_count is None: block_count = block_count_ - # try to simplify the block/off/size arguments a bit - if not isinstance(block, dict): - block = {'block': block} - block, off, size = ( - block.get('block'), - block.get('off'), - block.get('size')) - - if not isinstance(block, tuple): - block = block, - if isinstance(off, tuple) and len(off) == 1: - off, = off - if isinstance(size, tuple) and len(size) == 1: - if off is None: - off, = size - size = None - - if any(isinstance(b, tuple) and len(b) > 1 for b in block): - print("error: more than one block address?", - file=sys.stderr) - sys.exit(-1) - if isinstance(block[0], tuple): - block = (block[0][0], *block[1:]) - if len(block) > 1 and isinstance(block[1], tuple): - block = (block[0], block[1][0]) - if isinstance(block[0], tuple): - block, off_ = (block[0][0], *block[1:]), block[0][1] - if off is None: - off = off_ - if len(block) > 1 and isinstance(block[1], tuple): - block = (block[0], block[1][0]) - if len(block) == 1: - block, = block - - if isinstance(off, tuple): - off, size_ = off[0], off[1] - off[0] - if size is None: - size = size_ - if isinstance(size, tuple): - off_, size = off[0], off[1] - off[0] - if off is None: - off = off_ - - # is a block window specified? - block_window = None - if block is not None: - if isinstance(block, tuple): - block_window = range(*block) - else: - block_window = range(block, block+1) - - off_window = None - if off is not None or size is not None: - off_ = off if off is not None else 0 - size_ = size if size is not None else 1 - off_window = range(off_, off_+size_) - - # create our block device representation - bmap = Bmap( - block_size=block_size if block_size is not None else 1, - block_count=block_count if block_count is not None else 1, - block_window=block_window, - off_window=off_window) - - def resize(): - nonlocal bmap - - # figure out best width/height - if width is None: - width_ = min(80, shutil.get_terminal_size((80, 5))[0]) - elif width: - width_ = width - else: - width_ = shutil.get_terminal_size((80, 5))[0] - - if height is None: - height_ = 0 - elif height: - height_ = height - else: - height_ = shutil.get_terminal_size((80, 5))[1] - - # terminal size changed? - if width_ != bmap.width or height_ != bmap.height: - bmap.resize( - # scale if we're printing with dots or braille - width=2*width_ if braille else width_, - height=max( - 1, - 4*height_ if braille - else 2*height_ if dots - else height_)) - resize() - + # keep track of block_size/block_count and block map state + block_size_ = block_size + block_count_ = block_count + bmap = None # keep track of some extra info readed = 0 proged = 0 erased = 0 - # parse a line of trace output - pattern = re.compile( - '^(?P[^:]*):(?P[0-9]+):trace:.*?bd_(?:' - '(?Pcreate\w*)\(' + def bmap_init(block_size__, block_count__): + nonlocal block_size_ + nonlocal block_count_ + nonlocal bmap + nonlocal readed + nonlocal proged + nonlocal erased + + # keep track of block_size/block_count + if block_size is None: + block_size_ = block_size__ + if block_count is None: + block_count_ = block_count__ + + # flatten blocks, default to all blocks + blocks_ = list( + range(blocks.start or 0, blocks.stop or block_count_) + if isinstance(blocks, slice) + else range(blocks, blocks+1) + if blocks + else range(block_count_)) + + # create a new block map? + if bmap is None or volatile: + bmap = {b: TraceBlock(b) for b in blocks_} + readed = 0 + proged = 0 + erased = 0 + + # just resize block map + else: + bmap = {b: bmap[b] if b in bmap else TraceBlock(b) + for b in blocks_} + + # if we know block_count, go ahead and flatten blocks + create + # a block map, otherwise we need to wait for first bd init + if block_size is not None and block_count is not None: + bmap_init(block_size, block_count) + + + ## trace parser + + # precompute trace regexes + + # TODO rm me +# trace_pattern = re.compile( +# '^(?P[^:]*):(?P[0-9]+):trace:.*?bd_(?:' +# '(?Pcreate\w*)\(' +# '(?:' +# 'block_size=(?P\w+)' +# '|' 'block_count=(?P\w+)' +# '|' '.*?' ')*' +# '\)' +# '|' '(?Pread)\(' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*\)' +# '|' '(?Pprog)\(' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*\)' +# '|' '(?Perase)\(' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' +# '\s*\(\s*(?P\w+)\s*\)' '\s*\)' +# '|' '(?Psync)\(' +# '\s*(?P\w+)' '\s*\)' +# ')\s*$') + + init_pattern = re.compile( + '^(?P[^ :]*):(?P[0-9]+):trace:' + '.*?bd_createcfg\(' + '\s*(?P\w+)' '(?:' 'block_size=(?P\w+)' '|' 'block_count=(?P\w+)' - '|' '.*?' ')*' - '\)' - '|' '(?Pread)\(' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' '\s*\)' - '|' '(?Pprog)\(' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' '\s*\)' - '|' '(?Perase)\(' - '\s*(?P\w+)' '\s*,' - '\s*(?P\w+)' - '\s*\(\s*(?P\w+)\s*\)' '\s*\)' - '|' '(?Psync)\(' - '\s*(?P\w+)' '\s*\)' - ')\s*$') - def parse(line): - nonlocal bmap + '|' '.*?' ')*' '\)') + read_pattern = re.compile( + '^(?P[^ :]*):(?P[0-9]+):trace:' + '.*?bd_read\(' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' '\s*\)') + prog_pattern = re.compile( + '^(?P[^ :]*):(?P[0-9]+):trace:' + '.*?bd_prog\(' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' '\s*\)') + erase_pattern = re.compile( + '^(?P[^ :]*):(?P[0-9]+):trace:' + '.*?bd_erase\(' + '\s*(?P\w+)' '\s*,' + '\s*(?P\w+)' + '(?:\s*\(\s*(?P\w+)\s*\))?' '\s*\)') + sync_pattern = re.compile( + '^(?P[^ :]*):(?P[0-9]+):trace:' + '.*?bd_sync\(' + '\s*(?P\w+)' '\s*\)') + + def trace__(line): nonlocal readed nonlocal proged nonlocal erased - # string searching is much faster than the regex here, and this - # actually has a big impact given how much trace output comes - # through here - if 'trace' not in line or 'bd' not in line: - return False - m = pattern.match(line) - if not m: + # string searching is much faster than the regex here, this + # actually has a big impact given the sheer quantity of how much + # trace output we have to deal with + if ('trace' not in line + # ignore return trace statements + or '->' in line): return False - if m.group('create'): - # update our block size/count - block_size_ = int(m.group('block_size'), 0) - block_count_ = int(m.group('block_count'), 0) + # note we can't do most ops until we know block_count/block_size - if reset: - bmap = Bmap( - block_size=block_size_, - block_count=block_count_, - block_window=bmap.block_window, - off_window=bmap.off_window, - width=bmap.width, - height=bmap.height) - elif ((block_size is None - and block_size_ != bmap.block_size) - or (block_count is None - and block_count_ != bmap.block_count)): - bmap.resize( - block_size=block_size if block_size is not None - else block_size_, - block_count=block_count if block_count is not None - else block_count_) + # bd init? + if 'bd_createcfg(' in line: + m = init_pattern.match(line) + if not m: + return False + # block_size/block_count missing? + if not m.group('block_size') or not m.group('block_count'): + return False + block_size__ = int(m.group('block_size'), 0) + block_count__ = int(m.group('block_count'), 0) + + bmap_init(block_size__, block_count__) return True - elif m.group('read') and reads: - block = int(m.group('read_block'), 0) - off = int(m.group('read_off'), 0) - size = int(m.group('read_size'), 0) + # bd read? + elif reads and bmap is not None and 'bd_read(' in line: + m = read_pattern.match(line) + if not m: + return False + block = int(m.group('block'), 0) + off = int(m.group('off'), 0) + size = int(m.group('size'), 0) - if ((block_size is None and off+size > bmap.block_size) - or (block_count is None and block >= bmap.block_count)): - bmap.resize( - block_size=block_size if block_size is not None - else max(off+size, bmap.block_size), - block_count=block_count if block_count is not None - else max(block+1, bmap.block_count)) + if block not in bmap: + return False + else: + bmap[block].read(off, size) + readed += size + return True - bmap.read(block, off, size) - readed += size - return True + # bd prog? + elif progs and bmap is not None and 'bd_prog(' in line: + m = prog_pattern.match(line) + if not m: + return False + block = int(m.group('block'), 0) + off = int(m.group('off'), 0) + size = int(m.group('size'), 0) - elif m.group('prog') and progs: - block = int(m.group('prog_block'), 0) - off = int(m.group('prog_off'), 0) - size = int(m.group('prog_size'), 0) + if block not in bmap: + return False + else: + bmap[block].prog(off, size) + proged += size + return True - if ((block_size is None and off+size > bmap.block_size) - or (block_count is None and block >= bmap.block_count)): - bmap.resize( - block_size=block_size if block_size is not None - else max(off+size, bmap.block_size), - block_count=block_count if block_count is not None - else max(block+1, bmap.block_count)) + # bd erase? + elif (erases or wear) and bmap is not None and 'bd_erase(' in line: + m = erase_pattern.match(line) + if not m: + return False - bmap.prog(block, off, size) - proged += size - return True - - elif m.group('erase') and (erases or wear): - block = int(m.group('erase_block'), 0) - size = int(m.group('erase_size'), 0) - - if ((block_size is None and size > bmap.block_size) - or (block_count is None and block >= bmap.block_count)): - bmap.resize( - block_size=block_size if block_size is not None - else max(size, bmap.block_size), - block_count=block_count if block_count is not None - else max(block+1, bmap.block_count)) - - bmap.erase(block, size) - erased += size - return True + block = int(m.group('block'), 0) + if block not in bmap: + return False + else: + bmap[block].erase(0, block_count_, + wear=+1 if wear else 0) + erased += block_count_ + return True else: return False - # print trace output - def draw(f): - nonlocal readed - nonlocal proged - nonlocal erased - def writeln(s=''): - f.write(s) - f.write('\n') - f.writeln = writeln + ## bmap renderer - # don't forget we've scaled this for braille/dots! - for row in range( - mt.ceil(bmap.height/4) if braille - else mt.ceil(bmap.height/2) if dots - else bmap.height): - line = bmap.draw(row, - reads=reads, - progs=progs, - erases=erases, - wear=wear, - block_cycles=block_cycles, - color=color, - dots=dots, - braille=braille, - hilbert=hilbert, - lebesgue=lebesgue, - **args) - if line: - f.writeln(line) - - # print some information about reads/progs/erases - # - # cat implies no-header, because a header wouldn't really make sense - if not no_header and not cat: - # compute total ops - total = readed+proged+erased - - # compute stddev of wear using our bmap, this is a bit different - # from reads/progs/erases which ignores any bmap window, but it's - # what we have - if wear: - mean = (sum(p.wear for p in bmap.pixels) - / max(len(bmap.pixels), 1)) - stddev = mt.sqrt(sum((p.wear - mean)**2 for p in bmap.pixels) - / max(len(bmap.pixels), 1)) - worst = max((p.wear for p in bmap.pixels), default=0) - - # a bit of a hack here, but this forces our header to always be - # at row zero - if len(f.lines) == 0: - f.lines.append('') - f.lines[0] = 'bd %dx%d%s%s%s%s' % ( - bmap.block_size, bmap.block_count, - ', %6s read' % ('%.1f%%' % (100*readed / max(total, 1))) - if reads else '', - ', %6s prog' % ('%.1f%%' % (100*proged / max(total, 1))) - if progs else '', - ', %6s erase' % ('%.1f%%' % (100*erased / max(total, 1))) - if erases else '', - ', %13s wear' % ('%.1fσ (%.1f%%)' % ( - worst / max(stddev, 1), - 100*stddev / max(worst, 1))) - if wear else '') - - bmap.clear() - readed = 0 - proged = 0 - erased = 0 - resize() - - - # read/parse/coalesce operations - if cat: - ring = sys.stdout + # these curves are expensive to calculate, so memoize these + if hilbert: + curve = ft.lru_cache(16)(lambda w, h: list(hilbert_curve(w, h))) + elif lebesgue: + curve = ft.lru_cache(16)(lambda w, h: list(lebesgue_curve(w, h))) else: - ring = RingIO(lines + (1 if not no_header else 0), head) + curve = ft.lru_cache(16)(lambda w, h: list(naive_curve(w, h))) - # if sleep print in background thread to avoid getting stuck in a read call - event = th.Event() + # TODO adopt f -> ring name in all scripts? + def draw__(ring, width, height): + nonlocal bmap + # still waiting on bd init + if bmap is None: + return + + # give ring a writeln function + def writeln(s=''): + ring.write(s) + ring.write('\n') + ring.writeln = writeln + + # figure out width/height + if width is None: + width_ = min(80, shutil.get_terminal_size((80, 5))[0]) + elif width > 0: + width_ = width + else: + width_ = max(0, shutil.get_terminal_size((80, 5))[0] + width) + + if height is None: + height_ = 2 if not no_header else 1 + elif height > 0: + height_ = height + else: + height_ = max(0, shutil.get_terminal_size((80, 5))[1] + height) + + # scale width/height if requested + if (to_scale is not None + and (width is None or height is None)): + # scale width only + if height is not None: + width_ = mt.ceil((len(bmap) * to_scale) / height_) + # scale height only + elif width is not None: + height_ = mt.ceil((len(bmap) * to_scale) / width_) + # scale based on aspect-ratio + else: + width_ = mt.ceil(mt.sqrt(len(bmap) * to_scale) + * (aspect_ratio[0] / aspect_ratio[1])) + height_ = mt.ceil((len(bmap) * to_scale) / width_) + + # create a canvas + canvas = Canvas( + width_, + height_ - (1 if not no_header else 0), + color=color, + dots=dots, + braille=braille) + + # if contiguous, compute the global curve + if contiguous: + global_block = min(bmap.keys(), default=0) + global_curve = list(curve(canvas.width, canvas.height)) + + # if blocky, figure out block sizes/locations + else: + # figure out block_cols_/block_rows_ + if block_cols is not None and block_rows is not None: + block_cols_ = block_cols + block_rows_ = block_rows + elif block_rows is not None: + block_cols_ = mt.ceil(len(bmap) / block_rows) + block_rows_ = block_rows + elif block_cols is not None: + block_cols_ = block_cols + block_rows_ = mt.ceil(len(bmap) / block_cols) + else: + # divide by 2 until we hit our target ratio, this works + # well for things that are often powers-of-two + block_cols_ = 1 + block_rows_ = len(bmap) + while (abs(((canvas.width/(block_cols_ * 2)) + / (canvas.height/mt.ceil(block_rows_ / 2))) + - block_ratio) + < abs(((canvas.width/block_cols_) + / (canvas.height/block_rows_))) + - block_ratio): + block_cols_ *= 2 + block_rows_ = mt.ceil(block_rows_ / 2) + + block_width_ = canvas.width / block_cols_ + block_height_ = canvas.height / block_rows_ + + # assign block locations based on block_rows_/block_cols_ and + # the requested space filling curve + for (x, y), b in zip( + curve(block_cols_, block_rows_), + sorted(bmap.values())): + b.x = x * block_width_ + b.y = y * block_height_ + b.width = block_width_ + b.height = block_height_ + + # apply top padding + if x == 0: + b.x += padding + b.width -= min(padding, b.width) + if y == 0: + b.y += padding + b.height -= min(padding, b.height) + # apply bottom padding + b.width -= min(padding, b.width) + b.height -= min(padding, b.height) + + # align to pixel boundaries + b.align() + + # bump up to at least one pixel for every block, dont't + # worry about out-of-bounds, Canvas handles this for us + b.width = max(b.width, 1) + b.height = max(b.height, 1) + + # TODO chars should probably take priority over braille/dots + # assign chars based on op + block + for b in bmap.values(): + b.chars = {} + for op in ['read', 'prog', 'erase', 'noop']: + char__ = chars_.get((b.block, (op, '0x%x' % b.block))) + if char__ is not None: + # don't punescape unless we have to + if '%' in char__: + char__ = punescape(char__, b.attrs) + b.chars[op] = char__[0] # limit to 1 char + + # assign colors based on op + block + for b in bmap.values(): + b.colors = {} + for op in ['read', 'prog', 'erase', 'noop']: + color__ = colors_.get((b.block, (op, '0x%x' % b.block))) + if color__ is not None: + # don't punescape unless we have to + if '%' in color__: + color__ = punescape(color__, b.attrs) + b.colors[op] = color__ + + # assign wear chars based on block + for b in bmap.values(): + b.wear_chars = [] + for char__ in wear_chars_.getall((b.block, '0x%x' % b.block)): + # don't punescape unless we have to + if '%' in char__: + char__ = punescape(char__, b.attrs) + b.wear_chars.append(char__[0]) # limit to 1 char + + # assign wear colors based on block + for b in bmap.values(): + b.wear_colors = [] + for color__ in wear_colors_.getall((b.block, '0x%x' % b.block)): + # don't punescape unless we have to + if '%' in color__: + color__ = punescape(color__, b.attrs) + b.wear_colors.append(color__) + + # compute total ops + total = readed + proged + erased + + # if we're showing wear, find min/max/avg/etc + if wear: + wear_min = min(b.wear for b in bmap.values()) + wear_max = max(b.wear for b in bmap.values()) + wear_avg = (sum(b.wear for b in bmap.values()) + / max(len(bmap), 1)) + wear_stddev = mt.sqrt( + sum((b.wear - wear_avg)**2 for b in bmap.values()) + / max(len(bmap), 1)) + + # if block_cycles isn't provide, scale based on max wear + if block_cycles is not None: + block_cycles_ = block_cycles + else: + block_cycles_ = wear_max + + # render to canvas in a specific z-order that prioritizes + # interesting ops + for op in reversed(Z_ORDER): + # TODO if we're rendering noops as dashes here, should we render + # unused as dashes even in usage mode in dbgbmap.py? + # don't render noops in braille/dots mode + if (braille or dots) and op == 'noop': + continue + # skip ops we're not interested in + if ((not reads and op == 'read') + or (not progs and op == 'prog') + or (not erases and op == 'erase') + or (not wear and op == 'wear')): + continue + + for b in bmap.values(): + if op == 'read': + ranges__ = b.readed + char__ = b.chars['read'] + color__ = b.colors['read'] + elif op == 'prog': + ranges__ = b.proged + char__ = b.chars['prog'] + color__ = b.colors['prog'] + elif op == 'erase': + ranges__ = b.erased + char__ = b.chars['erase'] + color__ = b.colors['erase'] + elif op == 'wear': + # _no_ wear? + if b.wear == 0: + continue + ranges__ = RangeSet([range(block_size_)]) + # scale char/color based on max_wear/block_cycles + wear__ = b.wear / max(block_cycles_, 1) + char__ = b.wear_chars[int(wear__*(len(b.wear_chars)-1))] + color__ = b.wear_colors[int(wear__*(len(b.wear_colors)-1))] + else: + ranges__ = RangeSet([range(block_size_)]) + char__ = b.chars['noop'] + color__ = b.colors['noop'] + + if not ranges__: + continue + + # contiguous? + if contiguous: + for range__ in ranges__.ranges(): + # where are we in the curve? + block__ = b.block - global_block + range__ = range( + mt.floor(((block__*block_size_ + range__.start) + / (block_size_ * len(bmap))) + * len(global_curve)), + mt.ceil(((block__*block_size_ + range__.stop) + / (block_size_ * len(bmap))) + * len(global_curve))) + + # map to global curve + for i in range__: + if i >= len(global_curve): + continue + x__, y__ = global_curve[i] + + # flip y + y__ = canvas.height - (y__+1) + + canvas.point(x__, y__, + char=True if braille or dots else char__, + color=color__) + + # blocky? + else: + x__ = b.x + y__ = b.y + width__ = b.width + height__ = b.height + + # flip y + y__ = canvas.height - (y__+height__) + + for range__ in ranges__.ranges(): + # scale from bytes -> pixels + range__ = range( + mt.floor((range__.start/block_size_) + * (width__*height__)), + mt.ceil((range__.stop/block_size_) + * (width__*height__))) + # map to in-block curve + for i, (dx, dy) in enumerate(curve(width__, height__)): + if i in range__: + # flip y + canvas.point(x__+dx, y__+(height__-(dy+1)), + char=True if braille or dots + else char__, + color=color__) + + # print some summary info + if not no_header: + +# TODO +# # compute stddev of wear using our bmap, this is a bit different +# # from reads/progs/erases which ignores any bmap window, but it's +# # what we have +# if wear: +# mean = (sum(p.wear for p in bmap.pixels) +# / max(len(bmap.pixels), 1)) +# stddev = mt.sqrt(sum((p.wear - mean)**2 for p in bmap.pixels) +# / max(len(bmap.pixels), 1)) +# worst = max((p.wear for p in bmap.pixels), default=0) + + if title: + ring.writeln(punescape(title, { + # TODO simplify this in other scripts + # geometry -> just block_size/block_count, don't eagerly + # format percents, etc + 'block_size': block_size_, + 'block_count': block_count_, + 'total': total, + 'read': readed, + # TODO if we're showing percentage of total ops here, + # should we should percentage of in-use blocks in + # dbgbmap.py? + 'read_percent': 100*readed / max(total, 1), + 'prog': proged, + 'prog_percent': 100*proged / max(total, 1), + 'erase': erased, + 'erase_percent': 100*erased / max(total, 1), + 'wear_min': wear_min if wear else '?', + 'wear_min_percent': 100*wear_min / max(wear_max, 1) + if wear else '?', +# 'wear_min_stddev': (wear_min-wear_avg) / max(wear_stddev, 1) +# if wear else '?', + 'wear_max': wear_max if wear else '?', + 'wear_max_percent': 100*wear_max / max(wear_max, 1) + if wear else '?', +# 'wear_max_stddev': (wear_max-wear_avg) / max(wear_stddev, 1) +# if wear else '?', + 'wear_avg': wear_avg if wear else '?', + 'wear_avg_percent': 100*wear_avg / max(wear_max, 1) + if wear else '?', + 'wear_stddev': wear_stddev if wear else '?', + 'wear_stddev_percent': 100*wear_stddev / max(wear_max, 1) + if wear else '?', +# 'wear_stddev_percent': 100*(wear_stddev) / max(wear_max, 1) +# if wear else '?', + })) + else: +# ring.writeln('curve: %s' % (curve.cache_info(),)) + ring.writeln('bd %dx%d%s%s%s%s' % ( + block_size_, block_count_, + ', %6s read' % ( + '%.1f%%' % (100*readed / max(total, 1))) + if reads else '', + ', %6s prog' % ( + '%.1f%%' % (100*proged / max(total, 1))) + if progs else '', + ', %6s erase' % ( + '%.1f%%' % (100*erased / max(total, 1))) + if erases else '', + ', %15s wear' % ( + '%.1f%% +-%.1fσ' % ( + 100*wear_avg / max(wear_max, 1), + 100*wear_stddev / max(wear_max, 1))) + if wear else '')) + + # draw canvas + for row in range(canvas.height//canvas.yscale): + line = canvas.draw(row) + ring.writeln(line) + + # clear bmap + for b in bmap.values(): + b.clear() + + + +# TODO rm me +# +# +# +# +# +# +# +# +# +# # exclusive wear or reads/progs/erases by default +# if not reads and not progs and not erases and not wear: +# reads = True +# progs = True +# erases = True +# +# # assume a reasonable lines/height if not specified +# # +# # note that we let height = None if neither hilbert or lebesgue +# # are specified, this is a bit special as the default may be less +# # than one character in height. +# if height is None and (hilbert or lebesgue): +# if lines is not None: +# height = lines +# else: +# height = 5 +# +# if lines is None: +# if height is not None: +# lines = height +# else: +# lines = 5 +# +# # is bd geometry specified? +# if isinstance(block_size, tuple): +# block_size, block_count_ = block_size +# if block_count is None: +# block_count = block_count_ +# +# # try to simplify the block/off/size arguments a bit +# if not isinstance(block, dict): +# block = {'block': block} +# block, off, size = ( +# block.get('block'), +# block.get('off'), +# block.get('size')) +# +# if not isinstance(block, tuple): +# block = block, +# if isinstance(off, tuple) and len(off) == 1: +# off, = off +# if isinstance(size, tuple) and len(size) == 1: +# if off is None: +# off, = size +# size = None +# +# if any(isinstance(b, tuple) and len(b) > 1 for b in block): +# print("error: more than one block address?", +# file=sys.stderr) +# sys.exit(-1) +# if isinstance(block[0], tuple): +# block = (block[0][0], *block[1:]) +# if len(block) > 1 and isinstance(block[1], tuple): +# block = (block[0], block[1][0]) +# if isinstance(block[0], tuple): +# block, off_ = (block[0][0], *block[1:]), block[0][1] +# if off is None: +# off = off_ +# if len(block) > 1 and isinstance(block[1], tuple): +# block = (block[0], block[1][0]) +# if len(block) == 1: +# block, = block +# +# if isinstance(off, tuple): +# off, size_ = off[0], off[1] - off[0] +# if size is None: +# size = size_ +# if isinstance(size, tuple): +# off_, size = off[0], off[1] - off[0] +# if off is None: +# off = off_ +# +# # is a block window specified? +# block_window = None +# if block is not None: +# if isinstance(block, tuple): +# block_window = range(*block) +# else: +# block_window = range(block, block+1) +# +# off_window = None +# if off is not None or size is not None: +# off_ = off if off is not None else 0 +# size_ = size if size is not None else 1 +# off_window = range(off_, off_+size_) +# +# # create our block device representation +# bmap = Bmap( +# block_size=block_size if block_size is not None else 1, +# block_count=block_count if block_count is not None else 1, +# block_window=block_window, +# off_window=off_window) +# +# def resize(): +# nonlocal bmap +# +# # figure out best width/height +# if width is None: +# width_ = min(80, shutil.get_terminal_size((80, 5))[0]) +# elif width: +# width_ = width +# else: +# width_ = shutil.get_terminal_size((80, 5))[0] +# +# if height is None: +# height_ = 0 +# elif height: +# height_ = height +# else: +# height_ = shutil.get_terminal_size((80, 5))[1] +# +# # terminal size changed? +# if width_ != bmap.width or height_ != bmap.height: +# bmap.resize( +# # scale if we're printing with dots or braille +# width=2*width_ if braille else width_, +# height=max( +# 1, +# 4*height_ if braille +# else 2*height_ if dots +# else height_)) +# resize() +# +# # keep track of some extra info +# readed = 0 +# proged = 0 +# erased = 0 +# +# # parse a line of trace output +# pattern = re.compile( +# '^(?P[^:]*):(?P[0-9]+):trace:.*?bd_(?:' +# '(?Pcreate\w*)\(' +# '(?:' +# 'block_size=(?P\w+)' +# '|' 'block_count=(?P\w+)' +# '|' '.*?' ')*' +# '\)' +# '|' '(?Pread)\(' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*\)' +# '|' '(?Pprog)\(' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' '\s*\)' +# '|' '(?Perase)\(' +# '\s*(?P\w+)' '\s*,' +# '\s*(?P\w+)' +# '\s*\(\s*(?P\w+)\s*\)' '\s*\)' +# '|' '(?Psync)\(' +# '\s*(?P\w+)' '\s*\)' +# ')\s*$') +# def parse(line): +# nonlocal bmap +# nonlocal readed +# nonlocal proged +# nonlocal erased +# +# # string searching is much faster than the regex here, and this +# # actually has a big impact given the sheer quantity of how much +# # trace output we have to deal with +# if 'trace' not in line or 'bd' not in line: +# return False +# m = pattern.match(line) +# if not m: +# return False +# +# if m.group('create'): +# # update our block size/count +# block_size_ = int(m.group('block_size'), 0) +# block_count_ = int(m.group('block_count'), 0) +# +# if reset: +# bmap = Bmap( +# block_size=block_size_, +# block_count=block_count_, +# block_window=bmap.block_window, +# off_window=bmap.off_window, +# width=bmap.width, +# height=bmap.height) +# elif ((block_size is None +# and block_size_ != bmap.block_size) +# or (block_count is None +# and block_count_ != bmap.block_count)): +# bmap.resize( +# block_size=block_size if block_size is not None +# else block_size_, +# block_count=block_count if block_count is not None +# else block_count_) +# return True +# +# elif m.group('read') and reads: +# block = int(m.group('read_block'), 0) +# off = int(m.group('read_off'), 0) +# size = int(m.group('read_size'), 0) +# +# if ((block_size is None and off+size > bmap.block_size) +# or (block_count is None and block >= bmap.block_count)): +# bmap.resize( +# block_size=block_size if block_size is not None +# else max(off+size, bmap.block_size), +# block_count=block_count if block_count is not None +# else max(block+1, bmap.block_count)) +# +# bmap.read(block, off, size) +# readed += size +# return True +# +# elif m.group('prog') and progs: +# block = int(m.group('prog_block'), 0) +# off = int(m.group('prog_off'), 0) +# size = int(m.group('prog_size'), 0) +# +# if ((block_size is None and off+size > bmap.block_size) +# or (block_count is None and block >= bmap.block_count)): +# bmap.resize( +# block_size=block_size if block_size is not None +# else max(off+size, bmap.block_size), +# block_count=block_count if block_count is not None +# else max(block+1, bmap.block_count)) +# +# bmap.prog(block, off, size) +# proged += size +# return True +# +# elif m.group('erase') and (erases or wear): +# block = int(m.group('erase_block'), 0) +# size = int(m.group('erase_size'), 0) +# +# if ((block_size is None and size > bmap.block_size) +# or (block_count is None and block >= bmap.block_count)): +# bmap.resize( +# block_size=block_size if block_size is not None +# else max(size, bmap.block_size), +# block_count=block_count if block_count is not None +# else max(block+1, bmap.block_count)) +# +# bmap.erase(block, size) +# erased += size +# return True +# +# else: +# return False +# +# # print trace output +# def draw(f): +# nonlocal readed +# nonlocal proged +# nonlocal erased +# +# def writeln(s=''): +# f.write(s) +# f.write('\n') +# f.writeln = writeln +# +# # don't forget we've scaled this for braille/dots! +# for row in range( +# mt.ceil(bmap.height/4) if braille +# else mt.ceil(bmap.height/2) if dots +# else bmap.height): +# line = bmap.draw(row, +# reads=reads, +# progs=progs, +# erases=erases, +# wear=wear, +# block_cycles=block_cycles, +# color=color, +# dots=dots, +# braille=braille, +# hilbert=hilbert, +# lebesgue=lebesgue, +# **args) +# if line: +# f.writeln(line) +# +# # print some information about reads/progs/erases +# # +# # cat implies no-header, because a header wouldn't really make sense +# if not no_header and not cat: +# # compute total ops +# total = readed+proged+erased +# +# # compute stddev of wear using our bmap, this is a bit different +# # from reads/progs/erases which ignores any bmap window, but it's +# # what we have +# if wear: +# mean = (sum(p.wear for p in bmap.pixels) +# / max(len(bmap.pixels), 1)) +# stddev = mt.sqrt(sum((p.wear - mean)**2 for p in bmap.pixels) +# / max(len(bmap.pixels), 1)) +# worst = max((p.wear for p in bmap.pixels), default=0) +# +# # a bit of a hack here, but this forces our header to always be +# # at row zero +# if len(f.lines) == 0: +# f.lines.append('') +# f.lines[0] = 'bd %dx%d%s%s%s%s' % ( +# bmap.block_size, bmap.block_count, +# ', %6s read' % ('%.1f%%' % (100*readed / max(total, 1))) +# if reads else '', +# ', %6s prog' % ('%.1f%%' % (100*proged / max(total, 1))) +# if progs else '', +# ', %6s erase' % ('%.1f%%' % (100*erased / max(total, 1))) +# if erases else '', +# ', %13s wear' % ('%.1fσ (%.1f%%)' % ( +# worst / max(stddev, 1), +# 100*stddev / max(worst, 1))) +# if wear else '') +# +# bmap.clear() +# readed = 0 +# proged = 0 +# erased = 0 +# resize() +# +# +# # read/parse/coalesce operations +# if cat: +# ring = sys.stdout +# else: +# 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() +# lock = th.Lock() +# if sleep: +# done = False +# def background(): +# while not done: +# event.wait() +# event.clear() +# with lock: +# draw(ring) +# if not cat: +# ring.draw() +# # sleep a minimum amount of time to avoid flickering +# time.sleep(sleep or 0.01) +# th.Thread(target=background, daemon=True).start() +# +# try: +# while True: +# with openio(path) as f: +# changed = 0 +# for line in f: +# with lock: +# changed += parse(line) +# +# # need to redraw? +# if changed and (not coalesce or changed >= coalesce): +# if sleep: +# event.set() +# else: +# draw(ring) +# if not cat: +# ring.draw() +# changed = 0 +# +# if not keep_open: +# break +# # don't just flood open calls +# time.sleep(sleep or 2) +# except FileNotFoundError as e: +# print("error: file not found %r" % path, +# file=sys.stderr) +# sys.exit(-1) +# except KeyboardInterrupt: +# pass +# +# if sleep: +# done = True +# lock.acquire() # avoids https://bugs.python.org/issue42717 +# if not cat: +# sys.stdout.write('\n') +# +# +# +# +# +# +# +# def trace_(line): +# # TODO +# +# # TODO adopt f -> ring name in all scripts? +# def draw_(ring, *, +# width=None, +# height=None): +# # TODO + + + + + + ## main loop lock = th.Lock() - if sleep: - done = False - def background(): - while not done: - event.wait() - event.clear() - with lock: - draw(ring) - if not cat: - ring.draw() - # sleep a minimum amount of time to avoid flickering - time.sleep(sleep or 0.01) - th.Thread(target=background, daemon=True).start() + event = th.Event() + def main_(): + try: + while True: + with openio(path) as f: + count = 0 + for line in f: + with lock: + count += trace__(line) - try: - while True: - with openio(path) as f: - changed = 0 - for line in f: - with lock: - changed += parse(line) - - # need to redraw? - if changed and (not coalesce or changed >= coalesce): - if sleep: + # always redraw if we're sleeping, otherwise + # wait for coalesce number of operations + if sleep is not None or count >= (coalesce or 1): event.set() - else: - draw(ring) - if not cat: - ring.draw() - changed = 0 + count = 0 - if not keep_open: - break - # don't just flood open calls - time.sleep(sleep or 2) - except FileNotFoundError as e: - print("error: file not found %r" % path, - file=sys.stderr) - sys.exit(-1) - except KeyboardInterrupt: - pass + if not keep_open: + break + # don't just flood open calls + time.sleep(sleep or 2) - if sleep: - done = True - lock.acquire() # avoids https://bugs.python.org/issue42717 + except FileNotFoundError as e: + print("error: file not found %r" % path, + file=sys.stderr) + sys.exit(-1) + + except KeyboardInterrupt: + pass + + # keep track of history if lines specified + if lines is not None: + ring = RingIO(lines+1 + # TODO expose no_header in other scripts for consistency + if not no_header and lines > 0 + else lines) + def draw_(): + # cat? write directly to stdout + if cat: + draw__(sys.stdout, + # TODO copy this pattern (width+height) for + # other scripts? + width=width, + # make space for shell prompt + height=height if height is not False else -1) + # not cat? write to a bounded ring + else: + ring_ = RingIO(head=head) + draw__(ring_, + width=width, + height=height if height is not False else 0) + # no history? draw immediately + if lines is None: + ring_.draw() + # history? merge with previous lines + else: + # write header separately? + if not no_header: + if not ring.lines: + ring.lines.append('') + ring.lines.extend(it.islice(ring_.lines, 1, None)) + ring.lines[0] = ring_.lines[0] + else: + ring.lines.extend(ring_.lines) + ring.draw() + + # print in a background thread + done = False + def background(): + while not done: + event.wait() + event.clear() + with lock: + draw_() + # sleep a minimum amount of time to avoid flickering + time.sleep(sleep or 0.01) + th.Thread(target=background, daemon=True).start() + + main_() + + done = True + lock.acquire() # avoids https://bugs.python.org/issue42717 if not cat: + # give ourselves one last draw, helps if background is + # never triggered + draw_() sys.stdout.write('\n') + + + + +# # TODO WIP +# +# # cat? let main_ write directly to stdout +# else: +# main_(sys.stdout) +# +# # not cat? write to a bounded ring +# if not cat: +# ring = RingIO(lines, head=head) +# main_(ring) +# sys.stdout.write('\n') + + + + if __name__ == "__main__": import sys import argparse @@ -1086,44 +2570,14 @@ if __name__ == "__main__": '--block-count', type=lambda x: int(x, 0), help="Block count in blocks.") - # TODO implies --wear? parser.add_argument( - '-W', '--block-cycles', - type=lambda x: int(x, 0), - help="Assumed maximum number of erase cycles when measuring " - "wear.") - # subparser for block arguments - blockparser = argparse.ArgumentParser( - prog="%s -@/--block" % parser.prog, - allow_abbrev=False) - blockparser.add_argument( - 'block', - nargs='?', - type=lambda x: tuple( - rbydaddr(x) if x.strip() else None - for x in x.split(',')), - help="Block address, may be a range.") - blockparser.add_argument( - '--off', - type=lambda x: tuple( - int(x, 0) if x.strip() else None - for x in x.split(',')), - help="Show a specific offset, may be a range.") - blockparser.add_argument( - '-n', '--size', - type=lambda x: tuple( - int(x, 0) if x.strip() else None - for x in x.split(',')), - help="Show this many bytes, may be a range.") - parser.add_argument( - '-@', '--block', - type=lambda x: {k: v - for k, v in vars(blockparser.parse_intermixed_args( - shlex.split(x))).items() - if v is not None}, - help="Optional block to show, may be a range. Can also include " - "--off and -n/--size flags to indicate a range inside the " - "block, both which may also be ranges.") + '-@', '--blocks', + type=lambda x: ( + slice(*(int(x, 0) if x.strip() else None + for x in x.split(',', 1))) + if ',' in x + else int(x, 0)), + help="Show a specific block, may be a range.") parser.add_argument( '--reads', action='store_true', @@ -1137,17 +2591,73 @@ if __name__ == "__main__": action='store_true', help="Render erases.") parser.add_argument( - '-w', '--wear', + '--wear', action='store_true', help="Render wear.") parser.add_argument( - '-R', '--reset', + '--wear-only', + action='store_true', + help="Only render wear, don't render bd ops. Implies --wear.") + # TODO implies --wear? + parser.add_argument( + '-w', '--block-cycles', + nargs='?', + type=lambda x: int(x, 0), + const=0, + help="Assumed maximum number of erase cycles when measuring " + "wear. Implies --wear.") + parser.add_argument( + '--volatile', action='store_true', help="Reset wear on block device initialization.") parser.add_argument( - '--no-header', - action='store_true', - help="Don't show the header.") + '-.', '--add-char', '--chars', + dest='chars', + action='append', + type=lambda x: ( + lambda ks, v: ( + tuple(k.strip() for k in ks.split(',')), + v.strip()) + )(*x.split('=', 1)) + if '=' in x else x.strip(), + help="Add characters to use. Can be assigned to a specific " + "operation/block. Accepts %% modifiers.") + parser.add_argument( + '-,', '--add-wear-char', '--wear-chars', + dest='wear_chars', + action='append', + type=lambda x: ( + lambda ks, v: ( + tuple(k.strip() for k in ks.split(',')), + v.strip()) + )(*x.split('=', 1)) + if '=' in x else x.strip(), + help="Add wear characters to use. Can be assigned to a specific " + "operation/block. Accepts %% modifiers.") + parser.add_argument( + '-C', '--add-color', + dest='colors', + action='append', + type=lambda x: ( + lambda ks, v: ( + tuple(k.strip() for k in ks.split(',')), + v.strip()) + )(*x.split('=', 1)) + if '=' in x else x.strip(), + help="Add a color to use. Can be assigned to a specific " + "block type/block. Accepts %% modifiers.") + parser.add_argument( + '-G', '--add-wear-color', + dest='wear_colors', + action='append', + type=lambda x: ( + lambda ks, v: ( + tuple(k.strip() for k in ks.split(',')), + v.strip()) + )(*x.split('=', 1)) + if '=' in x else x.strip(), + help="Add a wear color to use. Can be assigned to a specific " + "operation/block. Accepts %% modifiers.") parser.add_argument( '--color', choices=['never', 'always', 'auto'], @@ -1162,48 +2672,42 @@ if __name__ == "__main__": action='store_true', help="Use 2x4 unicode braille characters. Note that braille " "characters sometimes suffer from inconsistent widths.") - parser.add_argument( - '--chars', - help="Characters to use for read, prog, erase, noop operations.") - parser.add_argument( - '--wear-chars', - help="Characters to use for showing wear.") - parser.add_argument( - '--colors', - type=lambda x: [x.strip() for x in x.split(',')], - help="Colors to use for read, prog, erase, noop operations.") - parser.add_argument( - '--wear-colors', - type=lambda x: [x.strip() for x in x.split(',')], - help="Colors to use for showing wear.") parser.add_argument( '-W', '--width', nargs='?', type=lambda x: int(x, 0), const=0, - help="Width in columns. 0 uses the terminal width. Defaults to " - "min(terminal, 80).") + help="Width in columns. <=0 uses the terminal width. Defaults " + "to min(terminal, 80).") parser.add_argument( '-H', '--height', nargs='?', type=lambda x: int(x, 0), - const=0, - help="Height in rows. 0 uses the terminal height. Defaults to 1.") + const=False, + help="Height in rows. <=0 uses the terminal height. Defaults " + "to 1.") parser.add_argument( - '-n', '--lines', - nargs='?', + '-X', '--block-cols', type=lambda x: int(x, 0), - const=0, - help="Show this many lines of history. 0 uses the terminal " - "height. Defaults to 5.") + help="Number of blocks on the x-axis. Guesses from --block-count " + "and --block-ratio by default.") parser.add_argument( - '-^', '--head', - action='store_true', - help="Show the first n lines.") + '-Y', '--block-rows', + type=lambda x: int(x, 0), + help="Number of blocks on the y-axis. Guesses from --block-count " + "and --block-ratio by default.") parser.add_argument( - '-c', '--cat', + '--block-ratio', + dest='block_ratio', + type=lambda x: ( + (lambda a, b: a / b)(*(float(v) for v in x.split(':', 1))) + if ':' in x else float(x)), + help="Target ratio for block sizes. Defaults to 1:1 or 1:2 " + "for -:/--dots and -⣿/--braille.") + parser.add_argument( + '--no-header', action='store_true', - help="Pipe directly to stdout.") + help="Don't show the header.") parser.add_argument( '-U', '--hilbert', action='store_true', @@ -1212,6 +2716,55 @@ if __name__ == "__main__": '-Z', '--lebesgue', action='store_true', help="Render as a space-filling Z-curve.") + parser.add_argument( + '-u', '--contiguous', + action='store_true', + help="Render as one contiguous curve instead of organizing by " + "blocks first.") + parser.add_argument( + '--to-scale', + nargs='?', + type=lambda x: ( + (lambda a, b: a / b)(*(float(v) for v in x.split(':', 1))) + if ':' in x else float(x)), + const=1, + help="Scale the resulting map such that 1 pixel ~= 1/scale " + "blocks. Defaults to scale=1. ") + parser.add_argument( + '-R', '--aspect-ratio', + type=lambda x: ( + tuple(float(v) for v in x.split(':', 1)) + if ':' in x else (float(x), 1)), + help="Aspect ratio to use with --to-scale. Defaults to 1:1.") + parser.add_argument( + '-t', '--tiny', + action='store_true', + help="Tiny mode, alias for --block-ratio=1, --to-scale=1, " + "and --no-header.") + parser.add_argument( + '--title', + help="Add a title. Accepts %% modifiers.") + # TODO drop padding in ascii scripts, no one is ever going to use this + parser.add_argument( + '--padding', + type=float, + help="Padding to add to each block. Defaults to 0.") + # TODO lines in dbgbmap.py? + parser.add_argument( + '-n', '--lines', + nargs='?', + type=lambda x: int(x, 0), + const=0, + help="Show this many lines of history. 0 uses the terminal " + "height. Defaults to 1.") + parser.add_argument( + '-^', '--head', + action='store_true', + help="Show the first n lines.") + parser.add_argument( + '-c', '--cat', + action='store_true', + help="Pipe directly to stdout.") parser.add_argument( '-S', '--coalesce', type=lambda x: int(x, 0), @@ -1219,8 +2772,8 @@ if __name__ == "__main__": parser.add_argument( '-s', '--sleep', type=float, - help="Time in seconds to sleep between reads, coalescing " - "operations.") + help="Seconds to sleep between draws, coalescing operations " + "in between.") parser.add_argument( '-k', '--keep-open', action='store_true',