9cd9ff626d
first, this fixes #607 also, i think it slightly simplifies code to make "simple" stuff like names, etc. attributes instead of methods all the time. so, expect this to be extended to other components, as well.
161 lines
6 KiB
Python
161 lines
6 KiB
Python
# pylint: disable=C0111,R0903
|
|
|
|
"""Displays network IO for interfaces.
|
|
|
|
Parameters:
|
|
* traffic.exclude: Comma-separated list of interface prefixes to exclude (defaults to 'lo,virbr,docker,vboxnet,veth')
|
|
* traffic.states: Comma-separated list of states to show (prefix with '^' to invert - i.e. ^down -> show all devices that are not in state down)
|
|
* traffic.showname: If set to False, hide network interface name (defaults to True)
|
|
* traffic.format: Format string for download/upload speeds.
|
|
Defaults to '{:.2f}'
|
|
* traffic.graphlen: Graph lenth in seconds. Positive even integer. Each
|
|
char shows 2 seconds. If set, enables up/down traffic
|
|
graphs
|
|
"""
|
|
|
|
import re
|
|
import time
|
|
import psutil
|
|
import netifaces
|
|
|
|
import core.module
|
|
import core.widget
|
|
|
|
import util.format
|
|
import util.graph
|
|
|
|
class Module(core.module.Module):
|
|
def __init__(self, config, theme):
|
|
widgets = []
|
|
super().__init__(config, theme, widgets)
|
|
|
|
self._exclude = tuple(filter(len, util.format.aslist(self.parameter('exclude', 'lo,virbr,docker,vboxnet,veth'))))
|
|
self._status = ''
|
|
|
|
self._showname = util.format.asbool(self.parameter('showname', True))
|
|
self._format = self.parameter('format', '{:.2f}')
|
|
self._prev = {}
|
|
self._states = {}
|
|
self._lastcheck = 0
|
|
self._states['include'] = []
|
|
self._states['exclude'] = []
|
|
for state in tuple(filter(len, util.format.aslist(self.parameter('states', '')))):
|
|
if state[0] == '^':
|
|
self._states['exclude'].append(state[1:])
|
|
else:
|
|
self._states['include'].append(state)
|
|
self._graphlen = int(self.parameter('graphlen', 0))
|
|
if self._graphlen > 0:
|
|
self._graphdata = {}
|
|
self._first_run = True
|
|
self._update_widgets(widgets)
|
|
|
|
def state(self, widget):
|
|
if 'traffic.rx' in widget.name:
|
|
return 'rx'
|
|
if 'traffic.tx' in widget.name:
|
|
return 'tx'
|
|
return self._status
|
|
|
|
def update(self):
|
|
self._update_widgets(self.widgets())
|
|
|
|
def create_widget(self, widgets, name, txt=None, attributes={}):
|
|
widget = core.widget.Widget(name=name, module=self)
|
|
widget.full_text(txt)
|
|
widgets.append(widget)
|
|
|
|
for key in attributes:
|
|
widget.set(key, attributes[key])
|
|
|
|
return widget
|
|
|
|
def get_addresses(self, intf):
|
|
retval = []
|
|
try:
|
|
for ip in netifaces.ifaddresses(intf).get(netifaces.AF_INET, []):
|
|
if ip.get('addr', '') != '':
|
|
retval.append(ip.get('addr'))
|
|
except Exception:
|
|
return []
|
|
return retval
|
|
|
|
def get_minwidth_str(self):
|
|
"""
|
|
computes theme.minwidth string
|
|
based on traffic.format and traffic.graphlen parameters
|
|
"""
|
|
minwidth_str = ''
|
|
if self._graphlen > 0:
|
|
graph_len = int(self._graphlen / 2)
|
|
graph_prefix = '0' * graph_len
|
|
minwidth_str += graph_prefix
|
|
minwidth_str += '1000'
|
|
try:
|
|
length = int(re.match('{:\.(\d+)f}', self._format).group(1))
|
|
if length > 0:
|
|
minwidth_str += '.' + '0' * length
|
|
except AttributeError:
|
|
# return default value
|
|
return '1000.00KiB/s'
|
|
finally:
|
|
minwidth_str += 'KiB/s'
|
|
return minwidth_str
|
|
|
|
def _update_widgets(self, widgets):
|
|
interfaces = [i for i in netifaces.interfaces() if not i.startswith(self._exclude)]
|
|
|
|
del widgets[:]
|
|
|
|
counters = psutil.net_io_counters(pernic=True)
|
|
now = time.time()
|
|
timediff = now - (self._lastcheck if self._lastcheck else now)
|
|
if timediff <= 0: timediff = 1
|
|
self._lastcheck = now
|
|
for interface in interfaces:
|
|
if self._graphlen > 0:
|
|
if interface not in self._graphdata:
|
|
self._graphdata[interface] = {
|
|
'rx': [0] * self._graphlen,
|
|
'tx': [0] * self._graphlen}
|
|
if not interface: interface = 'lo'
|
|
state = 'down'
|
|
if len(self.get_addresses(interface)) > 0:
|
|
state = 'up'
|
|
elif util.format.asbool(self.parameter('hide_down', True)):
|
|
continue
|
|
|
|
if len(self._states['exclude']) > 0 and state in self._states['exclude']: continue
|
|
if len(self._states['include']) > 0 and state not in self._states['include']: continue
|
|
|
|
data = {
|
|
'rx': counters[interface].bytes_recv,
|
|
'tx': counters[interface].bytes_sent,
|
|
}
|
|
|
|
name = 'traffic-{}'.format(interface)
|
|
|
|
if self._showname:
|
|
self.create_widget(widgets, name, interface)
|
|
|
|
for direction in ['rx', 'tx']:
|
|
name = 'traffic.{}-{}'.format(direction, interface)
|
|
widget = self.create_widget(widgets, name, attributes={'theme.minwidth': self.get_minwidth_str()})
|
|
prev = self._prev.get(name, 0)
|
|
bspeed = (int(data[direction]) - int(prev))/timediff
|
|
speed = util.format.byte(bspeed, self._format)
|
|
txtspeed = '{0}/s'.format(speed)
|
|
if self._graphlen > 0:
|
|
# skip first value returned by psutil, because it is
|
|
# giant and ruins the grapth ratio until it gets pushed
|
|
# out of saved list
|
|
if self._first_run is True:
|
|
self._first_run = False
|
|
else:
|
|
self._graphdata[interface][direction] = self._graphdata[interface][direction][1:]
|
|
self._graphdata[interface][direction].append(bspeed)
|
|
txtspeed = '{}{}'.format(util.graph.braille(self._graphdata[interface][direction]), txtspeed)
|
|
widget.full_text(txtspeed)
|
|
self._prev[name] = data[direction]
|
|
|
|
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|