Fixed: --no-color option didn't work (#198)

* Refactoring(color usage): new class Color makes it stateful and able to use args.
* Refactoring(top-utils): default init and post_optparse defined in basetop.
* Refactoring(color usage): "null-object" for --no-color and .enabled field

netutils_linux_monitoring/base_top.py

@@ -7,18 +7,21 @@
 
 from six import print_
 
-from netutils_linux_monitoring.colors import wrap, GREY
+from netutils_linux_monitoring.colors import Color
+from netutils_linux_monitoring.topology import Topology
 
 
 class BaseTop(object):
  """ Base class for all these top-like utils. """
  current = None
  previous = None
  diff = None
- header = wrap('Press CTRL-C to exit...\n', GREY)
+ header = Color.wrap('Press CTRL-C to exit...\n', Color.GREY)
  options = None
  file_arg = None
  file_value = None
+ topology = None
+ color = None
 
  @staticmethod
  def make_base_parser(parser=None):
@@ -113,6 +116,16 @@ def __repr_table__(self, table):
  return BaseTop.header + str(table)
  return str(table)
 
+ def default_init(self, topology=None):
+ BaseTop.__init__(self)
+ self.topology = topology
+ self.color = Color(self.topology)
+
+ def default_post_optparse(self):
+ if not self.topology:
+ self.topology = Topology(fake=self.options.random)
+ self.color = Color(self.topology, self.options.color)
+
  @abstractmethod
  def parse(self):
  """ Should read some file(s) into python structure (dict/list) """

netutils_linux_monitoring/colors.py

@@ -1,59 +1,73 @@
-from colorama import Fore, Style
-
-try:
- YELLOW = Fore.LIGHTYELLOW_EX
- GREY = Fore.LIGHTBLACK_EX
-except AttributeError:
- YELLOW = Fore.YELLOW
- GREY = Fore.CYAN
-
-COLORS_NODE = {
- 0: Fore.GREEN,
- 1: Fore.RED,
- 2: YELLOW,
- 3: Fore.BLUE,
- -1: Style.RESET_ALL,
-}
-
-COLORS_SOCKET = {
- 0: Fore.BLUE,
- 1: YELLOW,
- 2: Fore.RED,
- 3: Fore.GREEN,
- -1: Style.RESET_ALL,
-}
-
-
-def bright(string):
- return wrap(string, Style.BRIGHT)
-
-
-def wrap_header(string):
- return wrap("# {0}\n".format(string), Style.BRIGHT)
+# coding: utf-8
 
-
-def colorize(value, warning, error):
- return wrap(value, Fore.RED if value >= error else YELLOW if value >= warning else Fore.RESET)
-
-
-def wrap(word, color):
- """ wrap string in given color """
- return '{0}{1}{2}'.format(color, word, Style.RESET_ALL)
-
-
-def __choose_color_scheme(topology):
- return COLORS_NODE if topology.layout_kind == 'NUMA' else COLORS_SOCKET
-
-
-def cpu_color(cpu, topology, color_scheme=None):
- if not color_scheme:
- color_scheme = __choose_color_scheme(topology)
- if isinstance(cpu, str):
- cpu = int(cpu[3:])
- return color_scheme.get(topology.layout.get(cpu))
+from colorama import Fore, Style
 
 
-def colorize_cpu_list(cpu_list, topology):
- """ return list of highlighted strings with CPU names regarding to NUMA """
- color_scheme = __choose_color_scheme(topology)
- return [wrap(cpu, cpu_color(cpu, topology, color_scheme)) for cpu in cpu_list]
+class Color(object):
+ """ Helper for highlighting a console tools """
+ try:
+ YELLOW = Fore.LIGHTYELLOW_EX
+ GREY = Fore.LIGHTBLACK_EX
+ except AttributeError:
+ YELLOW = Fore.YELLOW
+ GREY = Fore.CYAN
+
+ COLORS_NODE = {
+ 0: Fore.GREEN,
+ 1: Fore.RED,
+ 2: YELLOW,
+ 3: Fore.BLUE,
+ -1: Style.RESET_ALL,
+ }
+
+ COLORS_SOCKET = {
+ 0: Fore.BLUE,
+ 1: YELLOW,
+ 2: Fore.RED,
+ 3: Fore.GREEN,
+ -1: Style.RESET_ALL,
+ }
+
+ COLOR_NONE = dict((key, "") for key in range(-1, 4))
+
+ def __init__(self, topology, enabled=True):
+ self.enabled = enabled
+ self.topology = topology
+ if topology is not None:
+ self.color_scheme = self.__choose_color_scheme()
+
+ def __choose_color_scheme(self):
+ if not self.enabled:
+ Style.BRIGHT = Style.RESET_ALL = Fore.RED = Fore.RESET = self.GREY = self.YELLOW = ""
+ return self.COLOR_NONE
+ if self.topology.layout_kind == 'NUMA':
+ return self.COLORS_NODE
+ return self.COLORS_SOCKET
+
+ @staticmethod
+ def wrap(word, color):
+ """ wrap string in given color """
+ return '{0}{1}{2}'.format(color, word, Style.RESET_ALL)
+
+ @staticmethod
+ def colorize(value, warning, error):
+ return Color.wrap(value, Fore.RED if value >= error else Color.YELLOW if value >= warning else Fore.RESET)
+
+ @staticmethod
+ def bright(string):
+ return Color.wrap(string, Style.BRIGHT)
+
+ @staticmethod
+ def wrap_header(string):
+ return Color.wrap("# {0}\n".format(string), Style.BRIGHT)
+
+ def colorize_cpu(self, cpu):
+ if not self.color_scheme:
+ self.color_scheme = self.__choose_color_scheme()
+ if isinstance(cpu, str):
+ cpu = int(cpu[3:])
+ return self.color_scheme.get(self.topology.layout.get(cpu))
+
+ def colorize_cpu_list(self, cpu_list):
+ """ return list of highlighted strings with CPU names regarding to NUMA """
+ return [self.wrap(cpu, self.colorize_cpu(cpu)) for cpu in cpu_list]

netutils_linux_monitoring/irqtop.py

@@ -6,9 +6,8 @@
 from six.moves import xrange
 
 from netutils_linux_monitoring.base_top import BaseTop
-from netutils_linux_monitoring.colors import colorize_cpu_list, colorize
+from netutils_linux_monitoring.colors import Color
 from netutils_linux_monitoring.layout import make_table
-from netutils_linux_monitoring.topology import Topology
 
 
 class IrqTop(BaseTop):
@@ -17,12 +16,10 @@ class IrqTop(BaseTop):
  file_arg, file_value = '--interrupts-file', '/proc/interrupts'
 
  def __init__(self, topology=None):
- BaseTop.__init__(self)
- self.topology = topology
+ BaseTop.default_init(self, topology)
 
  def post_optparse(self):
- if not self.topology:
- self.topology = Topology(fake=self.options.random)
+ BaseTop.default_post_optparse(self)
 
  def __int(self, line):
  return [self.int(item) for item in line.strip().split()]
@@ -54,7 +51,7 @@ def make_rows(self):
  for line in self.repr_source():
  if line[0] == 'CPU0':
  cpu_count = len(line)
- line = colorize_cpu_list(line, self.topology) + ['']
+ line = self.color.colorize_cpu_list(line) + ['']
  elif self.skip_zero_line(line): # hiding useless data such a kind of interrupt etc
  continue
  else: # make line with irq counters as compact as we can, it can be very long!
@@ -65,7 +62,7 @@ def make_rows(self):
  @staticmethod
  def colorize_irq_per_cpu(irq_per_cpu):
  """ :returns: highlighted by warning/error irq string """
- return colorize(irq_per_cpu, 40000, 80000)
+ return Color.colorize(irq_per_cpu, 40000, 80000)
 
  def __repr__(self):
  output_lines, cpu_count = self.make_rows()

netutils_linux_monitoring/link_rate.py

@@ -8,7 +8,7 @@
 from six import iteritems
 
 from netutils_linux_monitoring.base_top import BaseTop
-from netutils_linux_monitoring.colors import wrap, COLORS_NODE, colorize
+from netutils_linux_monitoring.colors import Color
 from netutils_linux_monitoring.layout import make_table
 from netutils_linux_monitoring.pci import PCI
 
@@ -75,7 +75,7 @@ def make_header(self):
 
  @staticmethod
  def colorize_stat(stat, value):
- return colorize(value, 1, 1) if 'errors' in stat.filename or 'dropped' in stat.filename else value
+ return Color.colorize(value, 1, 1) if 'errors' in stat.filename or 'dropped' in stat.filename else value
 
  def colorize_stats(self, dev, repr_source):
  return [self.colorize_stat(stat, repr_source[dev][stat]) for stat in self.stats]
@@ -86,8 +86,8 @@ def make_rows(self):
  repr_source = self.repr_source()
  for dev in self.options.devices:
  dev_node = self.pci.devices.get(dev)
- dev_color = COLORS_NODE.get(dev_node)
- _dev = wrap(dev, dev_color)
+ dev_color = self.color.COLORS_NODE.get(dev_node)
+ _dev = self.color.wrap(dev, dev_color)
  yield [_dev] + self.colorize_stats(dev, repr_source)
 
  def __repr__(self):
@@ -155,6 +155,7 @@ def post_optparse(self):
  if not self.pci:
  self.pci = PCI()
  self.pci.devices = self.pci.node_dev_dict(self.options.devices, self.options.random)
+ self.color = Color(topology=None, enabled=self.options.color)
 
  def unit_change(self):
  if not any([self.options.bits, self.options.kbits, self.options.mbits]):

netutils_linux_monitoring/network_top.py

@@ -5,7 +5,7 @@
 
 from netutils_linux_monitoring import IrqTop, Softirqs, SoftnetStatTop, LinkRateTop
 from netutils_linux_monitoring.base_top import BaseTop
-from netutils_linux_monitoring.colors import cpu_color, wrap, wrap_header, bright
+from netutils_linux_monitoring.colors import Color
 from netutils_linux_monitoring.layout import make_table
 from netutils_linux_monitoring.topology import Topology
 
@@ -23,8 +23,10 @@ def __init__(self):
  }
  self.parse_options()
  self.topology = Topology(fake=self.options.random, lscpu_output=self.options.lscpu_output)
+ self.color = Color(self.topology)
  for top in self.tops.values():
  top.topology = self.topology
+ top.color = self.color
 
  def parse(self):
  """
@@ -73,14 +75,14 @@ def parse_options(self):
  def __repr_dev(self):
  top = self.tops.get('link-rate')
  table = make_table(top.make_header(), top.align_map, top.make_rows())
- return wrap_header('Network devices') + str(table)
+ return self.color.wrap_header('Network devices') + str(table)
 
  def __repr_irq(self):
  top = self.tops.get('irqtop')
  output_lines, cpu_count = top.make_rows()
  align_map = top.make_align_map(cpu_count)
  table = make_table(output_lines[0], align_map, output_lines[1:])
- return wrap_header('/proc/interrupts') + str(table)
+ return self.color.wrap_header('/proc/interrupts') + str(table)
 
  def __repr_cpu(self):
  irqtop = self.tops.get('irqtop')
@@ -98,15 +100,15 @@ def __repr_cpu(self):
  'CPU', 'Interrupts', 'NET RX', 'NET TX',
  'total', 'dropped', 'time_squeeze', 'cpu_collision', 'received_rps',
  ]
- fields = [bright(word) for word in fields]
+ fields = [self.color.bright(word) for word in fields]
  rows = self.__repr_cpu_make_rows(irqtop, network_output, softirq_top, softnet_stat_top)
  table = make_table(fields, ['l'] + ['r'] * (len(fields) - 1), rows)
- return wrap_header('Load per cpu:') + str(table)
+ return self.color.wrap_header('Load per cpu:') + str(table)
 
  def __repr_cpu_make_rows(self, irqtop, network_output, softirq_top, softnet_stat_top):
  return [
  [
- wrap('CPU{0}'.format(stat.cpu), cpu_color(stat.cpu, self.topology)),
+ self.color.wrap('CPU{0}'.format(stat.cpu), self.color.colorize_cpu(stat.cpu)),
  irqtop.colorize_irq_per_cpu(irq),
  softirq_top.colorize_net_rx(net_rx),
  softirq_top.colorize_net_tx(net_tx),

netutils_linux_monitoring/softirqs.py

@@ -1,22 +1,20 @@
+# coding: utf-8
 from six import iteritems
 
 from netutils_linux_monitoring.base_top import BaseTop
-from netutils_linux_monitoring.colors import wrap, cpu_color, colorize
+from netutils_linux_monitoring.colors import Color
 from netutils_linux_monitoring.layout import make_table
-from netutils_linux_monitoring.topology import Topology
 
 
 class Softirqs(BaseTop):
  """ Utility for monitoring software interrupts distribution """
  file_arg, file_value = '--softirqs-file', '/proc/softirqs'
 
  def __init__(self, topology=None):
- BaseTop.__init__(self)
- self.topology = topology
+ BaseTop.default_init(self, topology)
 
  def post_optparse(self):
- if not self.topology:
- self.topology = Topology(fake=self.options.random)
+ BaseTop.default_post_optparse(self)
 
  def parse(self):
  with open(self.options.softirqs_file) as softirq_file:
@@ -34,7 +32,7 @@ def __repr__(self):
  net_tx = self.repr_source().get('NET_TX')[:active_cpu_count]
  rows = [
  [
- wrap('CPU{0}'.format(n), cpu_color(n, self.topology)),
+ self.color.wrap('CPU{0}'.format(n), self.color.colorize_cpu(n)),
  self.colorize_net_rx(v[0]),
  self.colorize_net_tx(v[1])
  ]
@@ -46,12 +44,12 @@ def __repr__(self):
  @staticmethod
  def colorize_net_rx(net_rx):
  """ :returns: highlighted by warning/error net_rx string """
- return colorize(net_rx, 40000, 80000)
+ return Color.colorize(net_rx, 40000, 80000)
 
  @staticmethod
  def colorize_net_tx(net_tx):
  """ :returns: highlighted by warning/error net_tx string """
- return colorize(net_tx, 20000, 30000)
+ return Color.colorize(net_tx, 20000, 30000)
 
  @staticmethod
  def __active_cpu_count__(data):