from prettytable import PrettyTable, HEADER
from signal import signal, SIGWINCH
from termios import TIOCGWINSZ
-from typing import Optional
+from typing import Any, Dict, Optional, Sequence, TextIO, Tuple
from ceph_argparse import parse_json_funcsigs, validate_command
class Termsize(object):
DEFAULT_SIZE = (25, 80)
- def __init__(self):
+
+ def __init__(self) -> None:
self.rows, self.cols = self._gettermsize()
self.changed = False
- def _gettermsize(self):
+ def _gettermsize(self) -> Tuple[int, int]:
try:
fd = sys.stdin.fileno()
sz = struct.pack('hhhh', 0, 0, 0, 0)
except IOError:
return self.DEFAULT_SIZE
- def update(self):
+ def update(self) -> None:
rows, cols = self._gettermsize()
if not self.changed:
self.changed = (self.rows, self.cols) != (rows, cols)
self.rows, self.cols = rows, cols
- def reset_changed(self):
+ def reset_changed(self) -> None:
self.changed = False
- def __str__(self):
+ def __str__(self) -> str:
return '%s(%dx%d, changed %s)' % (self.__class__,
self.rows, self.cols, self.changed)
- def __repr__(self):
+ def __repr__(self) -> str:
return 'Termsize(%d,%d,%s)' % (self.__class__,
self.rows, self.cols, self.changed)
BOLD_SEQ = "\033[1m"
UNDERLINE_SEQ = "\033[4m"
- def __init__(self, asok, statpats=None, min_prio=0):
+ def __init__(self,
+ asok: str,
+ statpats: Optional[Sequence[str]] = None,
+ min_prio: int = 0) -> None:
self.asok_path = asok
self._colored = False
- self._stats = None
+ self._stats: Optional[Dict[str, dict]] = None
self._schema = None
self._statpats = statpats
- self._stats_that_fit = dict()
+ self._stats_that_fit: Dict[str, dict] = OrderedDict()
self._min_prio = min_prio
self.termsize = Termsize()
- def supports_color(self, ostr):
+ def supports_color(self, ostr: TextIO) -> bool:
"""
Returns True if the running system's terminal supports color, and False
otherwise.
return False
return True
- def colorize(self, msg, color, dark=False):
+ def colorize(self,
+ msg: str,
+ color: int,
+ dark: bool = False) -> str:
"""
Decorate `msg` with escape sequences to give the requested color
"""
return (self.COLOR_DARK_SEQ if dark else self.COLOR_SEQ) % (30 + color) \
+ msg + self.RESET_SEQ
- def bold(self, msg):
+ def bold(self, msg) -> str:
"""
Decorate `msg` with escape sequences to make it appear bold
"""
return self.BOLD_SEQ + msg + self.RESET_SEQ
- def format_dimless(self, n, width):
+ def format_dimless(self, n: int, width: int) -> str:
"""
Format a number without units, so as to fit into `width` characters, substituting
an appropriate unit suffix.
else:
return formatted
- def col_width(self, nick):
+ def col_width(self, nick: str) -> int:
"""
Given the short name `nick` for a column, how many characters
of width should the column be allocated? Does not include spacing
"""
return max(len(nick), 4)
- def get_stats_that_fit(self):
+ def get_stats_that_fit(self) -> Tuple[Dict[str, dict], bool]:
'''
Get a possibly-truncated list of stats to display based on
current terminal width. Allow breaking mid-section.
'''
- current_fit = OrderedDict()
+ current_fit: Dict[str, dict] = OrderedDict()
if self.termsize.changed or not self._stats_that_fit:
width = 0
+ assert self._stats is not None
for section_name, names in self._stats.items():
for name, stat_data in names.items():
width += self.col_width(stat_data) + 1
break
self.termsize.reset_changed()
- changed = current_fit and (current_fit != self._stats_that_fit)
+ changed = bool(current_fit) and (current_fit != self._stats_that_fit)
if changed:
self._stats_that_fit = current_fit
return self._stats_that_fit, changed
- def _print_headers(self, ostr):
+ def _print_headers(self, ostr: TextIO) -> None:
"""
Print a header row to `ostr`
"""
sub_header += "\n"
ostr.write(sub_header)
- def _print_vals(self, ostr, dump, last_dump):
+ def _print_vals(self,
+ ostr: TextIO,
+ dump: Dict[str, Any],
+ last_dump: Dict[str, Any]) -> None:
"""
Print a single row of values to `ostr`, based on deltas between `dump` and
`last_dump`.
self._print_headers(ostr)
for section_name, names in fit.items():
for stat_name, stat_nick in names.items():
+ assert self._schema is not None
stat_type = self._schema[section_name][stat_name]['type']
if bool(stat_type & COUNTER):
n = max(dump[section_name][stat_name] -
else:
n = dump[section_name][stat_name]
- val_row += self.format_dimless(n, self.col_width(stat_nick))
+ val_row += self.format_dimless(int(n),
+ self.col_width(stat_nick))
val_row += " "
val_row = val_row[0:-1]
val_row += self.colorize("|", self.BLUE)
val_row = val_row[0:-len(self.colorize("|", self.BLUE))]
ostr.write("{0}\n".format(val_row))
- def _should_include(self, sect, name, prio):
+ def _should_include(self, sect: str, name: str, prio: int) -> bool:
'''
boolean: should we output this stat?
return True
- def _load_schema(self):
+ def _load_schema(self) -> None:
"""
Populate our instance-local copy of the daemon's performance counter
schema, and work out which stats we will display.
# Build list of which stats we will display
self._stats = OrderedDict()
+ assert self._schema is not None
for section_name, section_stats in self._schema.items():
for name, schema_data in section_stats.items():
prio = schema_data.get('priority', 0)
if not len(self._stats):
raise RuntimeError("no stats selected by filters")
- def _handle_sigwinch(self, signo, frame):
+ def _handle_sigwinch(self, signo, frame) -> None:
self.termsize.update()
- def run(self, interval, count=None, ostr=sys.stdout):
+ def run(self,
+ interval: int,
+ count: Optional[int] = None,
+ ostr: TextIO = sys.stdout) -> None:
"""
Print output at regular intervals until interrupted.
except KeyboardInterrupt:
return
- def list(self, ostr=sys.stdout):
+ def list(self, ostr: TextIO = sys.stdout) -> None:
"""
Show all selected stats with section, full name, nick, and prio
"""
table.align['nick'] = 'l'
table.align['prio'] = 'r'
self._load_schema()
+ assert self._stats is not None
+ assert self._schema is not None
for section_name, section_stats in self._stats.items():
for name, nick in section_stats.items():
prio = self._schema[section_name][name].get('priority') or 0
projects / ceph.git / commitdiff