mirror of
https://gitee.com/openharmony/third_party_littlefs
synced 2024-11-23 06:50:37 +00:00
1003 lines
30 KiB
Python
1003 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# Display operations on block devices based on trace output
|
|
#
|
|
# Example:
|
|
# ./scripts/tracebd.py trace
|
|
#
|
|
# Copyright (c) 2022, The littlefs authors.
|
|
# SPDX-License-Identifier: BSD-3-Clause
|
|
#
|
|
|
|
import collections as co
|
|
import functools as ft
|
|
import io
|
|
import itertools as it
|
|
import math as m
|
|
import os
|
|
import re
|
|
import shutil
|
|
import threading as th
|
|
import time
|
|
|
|
|
|
CHARS = 'rpe.'
|
|
COLORS = ['42', '45', '44', '']
|
|
|
|
WEAR_CHARS = '0123456789'
|
|
WEAR_CHARS_SUBSCRIPTS = '.₁₂₃₄₅₆789'
|
|
WEAR_COLORS = ['', '', '', '', '', '', '', '35', '35', '1;31']
|
|
|
|
CHARS_DOTS = " .':"
|
|
COLORS_DOTS = ['32', '35', '34', '']
|
|
CHARS_BRAILLE = (
|
|
'⠀⢀⡀⣀⠠⢠⡠⣠⠄⢄⡄⣄⠤⢤⡤⣤' '⠐⢐⡐⣐⠰⢰⡰⣰⠔⢔⡔⣔⠴⢴⡴⣴'
|
|
'⠂⢂⡂⣂⠢⢢⡢⣢⠆⢆⡆⣆⠦⢦⡦⣦' '⠒⢒⡒⣒⠲⢲⡲⣲⠖⢖⡖⣖⠶⢶⡶⣶'
|
|
'⠈⢈⡈⣈⠨⢨⡨⣨⠌⢌⡌⣌⠬⢬⡬⣬' '⠘⢘⡘⣘⠸⢸⡸⣸⠜⢜⡜⣜⠼⢼⡼⣼'
|
|
'⠊⢊⡊⣊⠪⢪⡪⣪⠎⢎⡎⣎⠮⢮⡮⣮' '⠚⢚⡚⣚⠺⢺⡺⣺⠞⢞⡞⣞⠾⢾⡾⣾'
|
|
'⠁⢁⡁⣁⠡⢡⡡⣡⠅⢅⡅⣅⠥⢥⡥⣥' '⠑⢑⡑⣑⠱⢱⡱⣱⠕⢕⡕⣕⠵⢵⡵⣵'
|
|
'⠃⢃⡃⣃⠣⢣⡣⣣⠇⢇⡇⣇⠧⢧⡧⣧' '⠓⢓⡓⣓⠳⢳⡳⣳⠗⢗⡗⣗⠷⢷⡷⣷'
|
|
'⠉⢉⡉⣉⠩⢩⡩⣩⠍⢍⡍⣍⠭⢭⡭⣭' '⠙⢙⡙⣙⠹⢹⡹⣹⠝⢝⡝⣝⠽⢽⡽⣽'
|
|
'⠋⢋⡋⣋⠫⢫⡫⣫⠏⢏⡏⣏⠯⢯⡯⣯' '⠛⢛⡛⣛⠻⢻⡻⣻⠟⢟⡟⣟⠿⢿⡿⣿')
|
|
|
|
|
|
def openio(path, mode='r', buffering=-1):
|
|
# allow '-' for stdin/stdout
|
|
if path == '-':
|
|
if mode == 'r':
|
|
return os.fdopen(os.dup(sys.stdin.fileno()), mode, buffering)
|
|
else:
|
|
return os.fdopen(os.dup(sys.stdout.fileno()), mode, buffering)
|
|
else:
|
|
return open(path, mode, buffering)
|
|
|
|
class LinesIO:
|
|
def __init__(self, maxlen=None):
|
|
self.maxlen = maxlen
|
|
self.lines = co.deque(maxlen=maxlen)
|
|
self.tail = io.StringIO()
|
|
|
|
# trigger automatic sizing
|
|
if maxlen == 0:
|
|
self.resize(0)
|
|
|
|
def write(self, s):
|
|
# note using split here ensures the trailing string has no newline
|
|
lines = s.split('\n')
|
|
|
|
if len(lines) > 1 and self.tail.getvalue():
|
|
self.tail.write(lines[0])
|
|
lines[0] = self.tail.getvalue()
|
|
self.tail = io.StringIO()
|
|
|
|
self.lines.extend(lines[:-1])
|
|
|
|
if lines[-1]:
|
|
self.tail.write(lines[-1])
|
|
|
|
def resize(self, maxlen):
|
|
self.maxlen = maxlen
|
|
if maxlen == 0:
|
|
maxlen = shutil.get_terminal_size((80, 5))[1]
|
|
if maxlen != self.lines.maxlen:
|
|
self.lines = co.deque(self.lines, maxlen=maxlen)
|
|
|
|
canvas_lines = 1
|
|
def draw(self):
|
|
# did terminal size change?
|
|
if self.maxlen == 0:
|
|
self.resize(0)
|
|
|
|
# first thing first, give ourself a canvas
|
|
while LinesIO.canvas_lines < len(self.lines):
|
|
sys.stdout.write('\n')
|
|
LinesIO.canvas_lines += 1
|
|
|
|
# clear the bottom of the canvas if we shrink
|
|
shrink = LinesIO.canvas_lines - len(self.lines)
|
|
if shrink > 0:
|
|
for i in range(shrink):
|
|
sys.stdout.write('\r')
|
|
if shrink-1-i > 0:
|
|
sys.stdout.write('\x1b[%dA' % (shrink-1-i))
|
|
sys.stdout.write('\x1b[K')
|
|
if shrink-1-i > 0:
|
|
sys.stdout.write('\x1b[%dB' % (shrink-1-i))
|
|
sys.stdout.write('\x1b[%dA' % shrink)
|
|
LinesIO.canvas_lines = len(self.lines)
|
|
|
|
for i, line in enumerate(self.lines):
|
|
# move cursor, clear line, disable/reenable line wrapping
|
|
sys.stdout.write('\r')
|
|
if len(self.lines)-1-i > 0:
|
|
sys.stdout.write('\x1b[%dA' % (len(self.lines)-1-i))
|
|
sys.stdout.write('\x1b[K')
|
|
sys.stdout.write('\x1b[?7l')
|
|
sys.stdout.write(line)
|
|
sys.stdout.write('\x1b[?7h')
|
|
if len(self.lines)-1-i > 0:
|
|
sys.stdout.write('\x1b[%dB' % (len(self.lines)-1-i))
|
|
sys.stdout.flush()
|
|
|
|
|
|
# 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*m.ceil(m.log2(max(width, height))))):
|
|
# we just operate on binary strings here because it's easier
|
|
b = '{:0{}b}'.format(i, 2*m.ceil(m.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 Block(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 Block(int(self) | 1)
|
|
|
|
def prog(self):
|
|
return Block(int(self) | 2)
|
|
|
|
def erase(self):
|
|
return Block((int(self) | 4) + 8)
|
|
|
|
def clear(self):
|
|
return Block(int(self) & ~7)
|
|
|
|
def __or__(self, other):
|
|
return Block(
|
|
(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, *,
|
|
read=True,
|
|
prog=True,
|
|
erase=True,
|
|
wear=False,
|
|
block_cycles=None,
|
|
color=True,
|
|
subscripts=False,
|
|
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:
|
|
if braille or dots:
|
|
colors = COLORS_DOTS
|
|
else:
|
|
colors = COLORS
|
|
if len(colors) < len(COLORS):
|
|
colors = colors + COLORS[len(colors):]
|
|
|
|
if wear_chars is None:
|
|
if subscripts:
|
|
wear_chars = WEAR_CHARS_SUBSCRIPTS
|
|
else:
|
|
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 erase and self.erased:
|
|
c = chars[2]
|
|
f.append(colors[2])
|
|
elif prog and self.proged:
|
|
c = chars[1]
|
|
f.append(colors[1])
|
|
elif read 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 Bd:
|
|
def __init__(self, *,
|
|
size=1,
|
|
count=1,
|
|
width=None,
|
|
height=1,
|
|
blocks=None):
|
|
if width is None:
|
|
width = count
|
|
|
|
if blocks is None:
|
|
self.blocks = [Block() for _ in range(width*height)]
|
|
else:
|
|
self.blocks = blocks
|
|
self.size = size
|
|
self.count = count
|
|
self.width = width
|
|
self.height = height
|
|
|
|
def _op(self, f, block=None, off=None, size=None):
|
|
if block is None:
|
|
range_ = range(len(self.blocks))
|
|
else:
|
|
if off is None:
|
|
off, size = 0, self.size
|
|
elif size is None:
|
|
off, size = 0, off
|
|
|
|
# update our geometry? this will do nothing if we haven't changed
|
|
self.resize(
|
|
size=max(self.size, off+size),
|
|
count=max(self.count, block+1))
|
|
|
|
# map to our block space
|
|
start = (block*self.size + off) / (self.size*self.count)
|
|
stop = (block*self.size + off+size) / (self.size*self.count)
|
|
|
|
range_ = range(
|
|
m.floor(start*len(self.blocks)),
|
|
m.ceil(stop*len(self.blocks)))
|
|
|
|
# apply the op
|
|
for i in range_:
|
|
self.blocks[i] = f(self.blocks[i])
|
|
|
|
def read(self, block=None, off=None, size=None):
|
|
self._op(Block.read, block, off, size)
|
|
|
|
def prog(self, block=None, off=None, size=None):
|
|
self._op(Block.prog, block, off, size)
|
|
|
|
def erase(self, block=None, off=None, size=None):
|
|
self._op(Block.erase, block, off, size)
|
|
|
|
def clear(self, block=None, off=None, size=None):
|
|
self._op(Block.clear, block, off, size)
|
|
|
|
def copy(self):
|
|
return Bd(
|
|
blocks=self.blocks.copy(),
|
|
size=self.size,
|
|
count=self.count,
|
|
width=self.width,
|
|
height=self.height)
|
|
|
|
def resize(self, *,
|
|
size=None,
|
|
count=None,
|
|
width=None,
|
|
height=None):
|
|
size = size if size is not None else self.size
|
|
count = count if count is not None else self.count
|
|
width = width if width is not None else self.width
|
|
height = height if height is not None else self.height
|
|
|
|
if (size == self.size
|
|
and count == self.count
|
|
and width == self.width
|
|
and height == self.height):
|
|
return
|
|
|
|
# transform our blocks
|
|
blocks = []
|
|
for x in range(width*height):
|
|
# map from new bd space
|
|
start = m.floor(x * (size*count)/(width*height))
|
|
stop = m.ceil((x+1) * (size*count)/(width*height))
|
|
start_block = start // size
|
|
start_off = start % size
|
|
stop_block = stop // size
|
|
stop_off = stop % size
|
|
# map to old bd space
|
|
start = start_block*self.size + start_off
|
|
stop = stop_block*self.size + stop_off
|
|
start = m.floor(start * len(self.blocks)/(self.size*self.count))
|
|
stop = m.ceil(stop * len(self.blocks)/(self.size*self.count))
|
|
|
|
# aggregate state
|
|
blocks.append(ft.reduce(
|
|
Block.__or__,
|
|
self.blocks[start:stop],
|
|
Block()))
|
|
|
|
self.size = size
|
|
self.count = count
|
|
self.width = width
|
|
self.height = height
|
|
self.blocks = blocks
|
|
|
|
def draw(self, row, *,
|
|
read=False,
|
|
prog=False,
|
|
erase=False,
|
|
wear=False,
|
|
hilbert=False,
|
|
lebesgue=False,
|
|
dots=False,
|
|
braille=False,
|
|
**args):
|
|
# find max wear?
|
|
max_wear = None
|
|
if wear:
|
|
max_wear = max(b.wear for b in self.blocks)
|
|
|
|
# fold via a curve?
|
|
if hilbert:
|
|
grid = [None]*(self.width*self.height)
|
|
for (x,y), b in zip(
|
|
hilbert_curve(self.width, self.height),
|
|
self.blocks):
|
|
grid[x + y*self.width] = b
|
|
elif lebesgue:
|
|
grid = [None]*(self.width*self.height)
|
|
for (x,y), b in zip(
|
|
lebesgue_curve(self.width, self.height),
|
|
self.blocks):
|
|
grid[x + y*self.width] = b
|
|
else:
|
|
grid = self.blocks
|
|
|
|
# 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)
|
|
|
|
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(Block())),
|
|
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_b = 0
|
|
best_b = Block()
|
|
for i in range(2*4):
|
|
b = grid[x+(2-1-(i%2)) + ((row*4)+(4-1-(i//2)))*self.width]
|
|
best_b |= b
|
|
if ((read and b.readed)
|
|
or (prog and b.proged)
|
|
or (erase and b.erased)
|
|
or (not read and not prog and not erase
|
|
and wear and b.worn(max_wear, **args) >= 0.7)):
|
|
byte_b |= 1 << i
|
|
|
|
line.append(best_b.draw(
|
|
max_wear,
|
|
CHARS_BRAILLE[byte_b],
|
|
braille=True,
|
|
read=read,
|
|
prog=prog,
|
|
erase=erase,
|
|
wear=wear,
|
|
**args))
|
|
elif dots:
|
|
# encode into a byte
|
|
for x in range(self.width):
|
|
byte_b = 0
|
|
best_b = Block()
|
|
for i in range(2):
|
|
b = grid[x + ((row*2)+(2-1-i))*self.width]
|
|
best_b |= b
|
|
if ((read and b.readed)
|
|
or (prog and b.proged)
|
|
or (erase and b.erased)
|
|
or (not read and not prog and not erase
|
|
and wear and b.worn(max_wear, **args) >= 0.7)):
|
|
byte_b |= 1 << i
|
|
|
|
line.append(best_b.draw(
|
|
max_wear,
|
|
CHARS_DOTS[byte_b],
|
|
dots=True,
|
|
read=read,
|
|
prog=prog,
|
|
erase=erase,
|
|
wear=wear,
|
|
**args))
|
|
else:
|
|
for x in range(self.width):
|
|
line.append(grid[x + row*self.width].draw(
|
|
max_wear,
|
|
read=read,
|
|
prog=prog,
|
|
erase=erase,
|
|
wear=wear,
|
|
**args))
|
|
|
|
return ''.join(line)
|
|
|
|
|
|
|
|
def main(path='-', *,
|
|
read=False,
|
|
prog=False,
|
|
erase=False,
|
|
wear=False,
|
|
block=(None,None),
|
|
off=(None,None),
|
|
block_size=None,
|
|
block_count=None,
|
|
block_cycles=None,
|
|
reset=False,
|
|
color='auto',
|
|
dots=False,
|
|
braille=False,
|
|
width=None,
|
|
height=None,
|
|
lines=None,
|
|
cat=False,
|
|
hilbert=False,
|
|
lebesgue=False,
|
|
coalesce=None,
|
|
sleep=None,
|
|
keep_open=False,
|
|
**args):
|
|
# figure out what color should be
|
|
if color == 'auto':
|
|
color = sys.stdout.isatty()
|
|
elif color == 'always':
|
|
color = True
|
|
else:
|
|
color = False
|
|
|
|
# exclusive wear or read/prog/erase by default
|
|
if not read and not prog and not erase and not wear:
|
|
read = True
|
|
prog = True
|
|
erase = 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
|
|
|
|
# allow ranges for blocks/offs
|
|
block_start = block[0]
|
|
block_stop = block[1] if len(block) > 1 else block[0]+1
|
|
off_start = off[0]
|
|
off_stop = off[1] if len(off) > 1 else off[0]+1
|
|
|
|
if block_start is None:
|
|
block_start = 0
|
|
if block_stop is None and block_count is not None:
|
|
block_stop = block_count
|
|
if off_start is None:
|
|
off_start = 0
|
|
if off_stop is None and block_size is not None:
|
|
off_stop = block_size
|
|
|
|
# create a block device representation
|
|
bd = Bd()
|
|
|
|
def resize(*, size=None, count=None):
|
|
nonlocal bd
|
|
|
|
# size may be overriden by cli args
|
|
if block_size is not None:
|
|
size = block_size
|
|
elif off_stop is not None:
|
|
size = off_stop-off_start
|
|
|
|
if block_count is not None:
|
|
count = block_count
|
|
elif block_stop is not None:
|
|
count = block_stop-block_start
|
|
|
|
# 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]
|
|
|
|
bd.resize(
|
|
size=size,
|
|
count=count,
|
|
# 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()
|
|
|
|
# parse a line of trace output
|
|
pattern = re.compile(
|
|
'^(?P<file>[^:]*):(?P<line>[0-9]+):trace:.*?bd_(?:'
|
|
'(?P<create>create\w*)\('
|
|
'(?:'
|
|
'block_size=(?P<block_size>\w+)'
|
|
'|' 'block_count=(?P<block_count>\w+)'
|
|
'|' '.*?' ')*' '\)'
|
|
'|' '(?P<read>read)\('
|
|
'\s*(?P<read_ctx>\w+)' '\s*,'
|
|
'\s*(?P<read_block>\w+)' '\s*,'
|
|
'\s*(?P<read_off>\w+)' '\s*,'
|
|
'\s*(?P<read_buffer>\w+)' '\s*,'
|
|
'\s*(?P<read_size>\w+)' '\s*\)'
|
|
'|' '(?P<prog>prog)\('
|
|
'\s*(?P<prog_ctx>\w+)' '\s*,'
|
|
'\s*(?P<prog_block>\w+)' '\s*,'
|
|
'\s*(?P<prog_off>\w+)' '\s*,'
|
|
'\s*(?P<prog_buffer>\w+)' '\s*,'
|
|
'\s*(?P<prog_size>\w+)' '\s*\)'
|
|
'|' '(?P<erase>erase)\('
|
|
'\s*(?P<erase_ctx>\w+)' '\s*,'
|
|
'\s*(?P<erase_block>\w+)'
|
|
'\s*\(\s*(?P<erase_size>\w+)\s*\)' '\s*\)'
|
|
'|' '(?P<sync>sync)\('
|
|
'\s*(?P<sync_ctx>\w+)' '\s*\)' ')\s*$')
|
|
def parse(line):
|
|
nonlocal bd
|
|
|
|
# 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:
|
|
return False
|
|
|
|
if m.group('create'):
|
|
# update our block size/count
|
|
size = int(m.group('block_size'), 0)
|
|
count = int(m.group('block_count'), 0)
|
|
|
|
resize(size=size, count=count)
|
|
if reset:
|
|
bd = Bd(
|
|
size=bd.size,
|
|
count=bd.count,
|
|
width=bd.width,
|
|
height=bd.height)
|
|
return True
|
|
|
|
elif m.group('read') and read:
|
|
block = int(m.group('read_block'), 0)
|
|
off = int(m.group('read_off'), 0)
|
|
size = int(m.group('read_size'), 0)
|
|
|
|
if block_stop is not None and block >= block_stop:
|
|
return False
|
|
block -= block_start
|
|
if off_stop is not None:
|
|
if off >= off_stop:
|
|
return False
|
|
size = min(size, off_stop-off)
|
|
off -= off_start
|
|
|
|
bd.read(block, off, size)
|
|
return True
|
|
|
|
elif m.group('prog') and prog:
|
|
block = int(m.group('prog_block'), 0)
|
|
off = int(m.group('prog_off'), 0)
|
|
size = int(m.group('prog_size'), 0)
|
|
|
|
if block_stop is not None and block >= block_stop:
|
|
return False
|
|
block -= block_start
|
|
if off_stop is not None:
|
|
if off >= off_stop:
|
|
return False
|
|
size = min(size, off_stop-off)
|
|
off -= off_start
|
|
|
|
bd.prog(block, off, size)
|
|
return True
|
|
|
|
elif m.group('erase') and (erase or wear):
|
|
block = int(m.group('erase_block'), 0)
|
|
size = int(m.group('erase_size'), 0)
|
|
|
|
if block_stop is not None and block >= block_stop:
|
|
return False
|
|
block -= block_start
|
|
if off_stop is not None:
|
|
size = min(size, off_stop)
|
|
off = -off_start
|
|
|
|
bd.erase(block, off, size)
|
|
return True
|
|
|
|
else:
|
|
return False
|
|
|
|
# print trace output
|
|
def draw(f):
|
|
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(
|
|
m.ceil(bd.height/4) if braille
|
|
else m.ceil(bd.height/2) if dots
|
|
else bd.height):
|
|
line = bd.draw(row,
|
|
read=read,
|
|
prog=prog,
|
|
erase=erase,
|
|
wear=wear,
|
|
block_cycles=block_cycles,
|
|
color=color,
|
|
dots=dots,
|
|
braille=braille,
|
|
hilbert=hilbert,
|
|
lebesgue=lebesgue,
|
|
**args)
|
|
if line:
|
|
f.writeln(line)
|
|
|
|
bd.clear()
|
|
resize()
|
|
|
|
|
|
# read/parse/coalesce operations
|
|
if cat:
|
|
ring = sys.stdout
|
|
else:
|
|
ring = LinesIO(lines)
|
|
|
|
# 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()
|
|
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 0.1)
|
|
except FileNotFoundError as e:
|
|
print("error: file not found %r" % path)
|
|
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')
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
import argparse
|
|
parser = argparse.ArgumentParser(
|
|
description="Display operations on block devices based on "
|
|
"trace output.",
|
|
allow_abbrev=False)
|
|
parser.add_argument(
|
|
'path',
|
|
nargs='?',
|
|
help="Path to read from.")
|
|
parser.add_argument(
|
|
'-r', '--read',
|
|
action='store_true',
|
|
help="Render reads.")
|
|
parser.add_argument(
|
|
'-p', '--prog',
|
|
action='store_true',
|
|
help="Render progs.")
|
|
parser.add_argument(
|
|
'-e', '--erase',
|
|
action='store_true',
|
|
help="Render erases.")
|
|
parser.add_argument(
|
|
'-w', '--wear',
|
|
action='store_true',
|
|
help="Render wear.")
|
|
parser.add_argument(
|
|
'-b', '--block',
|
|
type=lambda x: tuple(
|
|
int(x, 0) if x.strip() else None
|
|
for x in x.split(',')),
|
|
help="Show a specific block or range of blocks.")
|
|
parser.add_argument(
|
|
'-i', '--off',
|
|
type=lambda x: tuple(
|
|
int(x, 0) if x.strip() else None
|
|
for x in x.split(',')),
|
|
help="Show a specific offset or range of offsets.")
|
|
parser.add_argument(
|
|
'-B', '--block-size',
|
|
type=lambda x: int(x, 0),
|
|
help="Assume a specific block size.")
|
|
parser.add_argument(
|
|
'--block-count',
|
|
type=lambda x: int(x, 0),
|
|
help="Assume a specific block count.")
|
|
parser.add_argument(
|
|
'-C', '--block-cycles',
|
|
type=lambda x: int(x, 0),
|
|
help="Assumed maximum number of erase cycles when measuring wear.")
|
|
parser.add_argument(
|
|
'-R', '--reset',
|
|
action='store_true',
|
|
help="Reset wear on block device initialization.")
|
|
parser.add_argument(
|
|
'--color',
|
|
choices=['never', 'always', 'auto'],
|
|
default='auto',
|
|
help="When to use terminal colors. Defaults to 'auto'.")
|
|
parser.add_argument(
|
|
'--subscripts',
|
|
action='store_true',
|
|
help="Use unicode subscripts for showing wear.")
|
|
parser.add_argument(
|
|
'-:', '--dots',
|
|
action='store_true',
|
|
help="Use 1x2 ascii dot characters.")
|
|
parser.add_argument(
|
|
'-⣿', '--braille',
|
|
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).")
|
|
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.")
|
|
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 5.")
|
|
parser.add_argument(
|
|
'-z', '--cat',
|
|
action='store_true',
|
|
help="Pipe directly to stdout.")
|
|
parser.add_argument(
|
|
'-U', '--hilbert',
|
|
action='store_true',
|
|
help="Render as a space-filling Hilbert curve.")
|
|
parser.add_argument(
|
|
'-Z', '--lebesgue',
|
|
action='store_true',
|
|
help="Render as a space-filling Z-curve.")
|
|
parser.add_argument(
|
|
'-c', '--coalesce',
|
|
type=lambda x: int(x, 0),
|
|
help="Number of operations to coalesce together.")
|
|
parser.add_argument(
|
|
'-s', '--sleep',
|
|
type=float,
|
|
help="Time in seconds to sleep between reads, coalescing operations.")
|
|
parser.add_argument(
|
|
'-k', '--keep-open',
|
|
action='store_true',
|
|
help="Reopen the pipe on EOF, useful when multiple "
|
|
"processes are writing.")
|
|
sys.exit(main(**{k: v
|
|
for k, v in vars(parser.parse_intermixed_args()).items()
|
|
if v is not None}))
|