bumblebee-status/bumblebee_status/modules/contrib/traffic.py

188 lines
6.3 KiB
Python
Raw Normal View History

2020-04-11 09:02:11 +02:00
# pylint: disable=C0111,R0903
"""Displays network IO for interfaces.
Parameters:
2020-04-11 09:02:47 +02:00
* 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)
2020-04-11 09:02:11 +02:00
* traffic.showname: If set to False, hide network interface name (defaults to True)
* traffic.format: Format string for download/upload speeds.
2020-05-06 12:57:38 +02:00
Defaults to '{:.2f}'
2020-04-11 09:02:11 +02:00
* traffic.graphlen: Graph lenth in seconds. Positive even integer. Each
2020-05-06 12:57:38 +02:00
char shows 2 seconds. If set, enables up/down traffic
graphs
contributed by `meain <https://github.com/meain>`_ - many thanks!
2020-04-11 09:02:11 +02:00
"""
import re
import time
import psutil
import netifaces
2020-04-11 09:11:24 +02:00
import core.module
2020-04-11 09:02:11 +02:00
2020-04-11 09:11:24 +02:00
import util.format
import util.graph
2020-04-11 09:11:24 +02:00
class Module(core.module.Module):
def __init__(self, config, theme):
super().__init__(config, theme, [])
2020-04-11 09:11:24 +02:00
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}")
2020-04-11 09:02:11 +02:00
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:])
2020-04-11 09:02:11 +02:00
else:
self._states["include"].append(state)
self._graphlen = int(self.parameter("graphlen", 0))
2020-04-11 09:02:11 +02:00
if self._graphlen > 0:
self._graphdata = {}
self._first_run = True
self._update_widgets()
2020-04-11 09:02:11 +02:00
def state(self, widget):
if "traffic.rx" in widget.name:
return "rx"
if "traffic.tx" in widget.name:
return "tx"
2020-04-11 09:02:11 +02:00
return self._status
2020-04-11 09:11:24 +02:00
def update(self):
self._update_widgets()
2020-04-11 09:02:11 +02:00
def create_widget(self, name, txt=None, attributes={}):
widget = self.add_widget(name=name, full_text=txt)
2020-04-11 09:02:11 +02:00
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"))
2020-04-11 09:02:11 +02:00
except Exception:
return []
return retval
def get_minwidth_str(self):
"""
computes theme.minwidth string
based on traffic.format and traffic.graphlen parameters
"""
minwidth_str = ""
2020-04-11 09:02:11 +02:00
if self._graphlen > 0:
graph_len = int(self._graphlen / 2)
graph_prefix = "0" * graph_len
2020-04-11 09:02:11 +02:00
minwidth_str += graph_prefix
minwidth_str += "1000"
2020-04-11 09:02:11 +02:00
try:
length = int(re.match("{:\.(\d+)f}", self._format).group(1))
2020-04-11 09:02:11 +02:00
if length > 0:
minwidth_str += "." + "0" * length
2020-04-11 09:02:11 +02:00
except AttributeError:
# return default value
return "1000.00KiB/s"
2020-04-11 09:02:11 +02:00
finally:
minwidth_str += "KiB/s"
2020-04-11 09:02:11 +02:00
return minwidth_str
def _update_widgets(self):
interfaces = [
i for i in netifaces.interfaces() if not i.startswith(self._exclude)
]
2020-04-11 09:02:11 +02:00
self.clear_widgets()
2020-04-11 09:02:11 +02:00
counters = psutil.net_io_counters(pernic=True)
now = time.time()
timediff = now - (self._lastcheck if self._lastcheck else now)
if timediff <= 0:
timediff = 1
2020-04-11 09:02:11 +02:00
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"
2020-04-11 09:02:11 +02:00
if len(self.get_addresses(interface)) > 0:
state = "up"
elif util.format.asbool(self.parameter("hide_down", True)):
2020-04-11 09:02:11 +02:00
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
2020-04-11 09:02:11 +02:00
data = {
"rx": counters[interface].bytes_recv,
"tx": counters[interface].bytes_sent,
2020-04-11 09:02:11 +02:00
}
name = "traffic-{}".format(interface)
2020-04-11 09:02:11 +02:00
if self._showname:
self.create_widget(name, interface)
for direction in ["rx", "tx"]:
name = "traffic.{}-{}".format(direction, interface)
widget = self.create_widget(
2020-05-09 21:24:28 +02:00
name, attributes={"theme.minwidth": self.get_minwidth_str()},
)
2020-04-11 09:02:11 +02:00
prev = self._prev.get(name, 0)
bspeed = (int(data[direction]) - int(prev)) / timediff
2020-04-11 09:11:24 +02:00
speed = util.format.byte(bspeed, self._format)
txtspeed = "{0}/s".format(speed)
2020-04-11 09:02:11 +02:00
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:]
2020-04-11 09:02:11 +02:00
self._graphdata[interface][direction].append(bspeed)
txtspeed = "{}{}".format(
util.graph.braille(self._graphdata[interface][direction]),
txtspeed,
)
widget.full_text(txtspeed)
2020-04-11 09:02:11 +02:00
self._prev[name] = data[direction]
2020-04-11 09:02:11 +02:00
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4