masq/i3pystatus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #154 from facetoe/master

Browse Source 
Centralize network related functionality under one module.
This commit is contained in:
enkore 2015-01-08 00:35:40 +01:00
commit a786736ee2
5 changed files with 245 additions and 291 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/i3pystatus.rst
View File

@ -7,7 +7,7 @@ Module reference
:System: `clock`_ - `disk`_ - `load`_ - `mem`_ - `cpu_usage`_
:Audio: `alsa`_ - `pulseaudio`_
:Hardware: `battery`_ - `backlight`_ - `temp`_
:Network: `network`_ - `wireless`_
:Network: `network`_
:Music: `now_playing`_ - `mpd`_
:Websites & stuff: `weather`_ - `bitcoin`_ - `reddit`_ - `parcel`_
:Other: `mail`_ - `pyload`_ - `text`_

323
i3pystatus/network.py
View File

@ -1,9 +1,10 @@
from itertools import zip_longest
# -*- coding: utf-8 -*-
import netifaces
import basiciw
import psutil
from i3pystatus import IntervalModule
# Reminder: if we raise minimum Python version to 3.3, use ipaddress module
from i3pystatus.core.color import ColorRangeModule
from i3pystatus.core.util import make_graph, round_dict, make_bar
def count_bits(integer):
@ -62,20 +63,166 @@ def sysfs_interface_up(interface, unknown_up=False):
with open("/sys/class/net/{}/operstate".format(interface)) as f:
status = f.read().strip()
except FileNotFoundError:
raise RuntimeError("Unknown interface {iface}!".format(iface=interface))
# Interface doesn't exist
return False
return status == "up" or unknown_up and status == "unknown"
class Network(IntervalModule):
class NetworkInfo():
"""
Retrieve network information.
"""
Display network information about a interface.
Requires the PyPI package `netifaces`.
def __init__(self, interface, ignore_interfaces, detached_down, unknown_up):
if interface not in netifaces.interfaces() and not detached_down:
raise RuntimeError(
"Unknown interface {iface}!".format(iface=interface))
self.ignore_interfaces = ignore_interfaces
self.detached_down = detached_down
self.unknown_up = unknown_up
def get_info(self, interface):
format_dict = dict(v4="", v4mask="", v4cidr="", v6="", v6mask="", v6cidr="")
iface_up = sysfs_interface_up(interface, self.unknown_up)
if not iface_up:
return format_dict
network_info = netifaces.ifaddresses(interface)
slaves = get_bonded_slaves()
try:
master = slaves[interface]
except KeyError:
pass
else:
if sysfs_interface_up(interface, self.unknown_up):
master_info = netifaces.ifaddresses(master)
for af in (netifaces.AF_INET, netifaces.AF_INET6):
try:
network_info[af] = master_info[af]
except KeyError:
pass
try:
mac = network_info[netifaces.AF_PACKET][0]["addr"]
except KeyError:
mac = "NONE"
format_dict['mac'] = mac
if iface_up:
format_dict.update(self.extract_network_info(network_info))
format_dict.update(self.extract_wireless_info(interface))
return format_dict
@staticmethod
def extract_network_info(network_info):
info = dict()
if netifaces.AF_INET in network_info:
v4 = network_info[netifaces.AF_INET][0]
info["v4"] = v4["addr"]
info["v4mask"] = v4["netmask"]
info["v4cidr"] = cidr4(v4["addr"], v4["netmask"])
if netifaces.AF_INET6 in network_info:
for v6 in network_info[netifaces.AF_INET6]:
info["v6"] = v6["addr"]
info["v6mask"] = v6["netmask"]
info["v6cidr"] = cidr6(v6["addr"], v6["netmask"])
if not v6["addr"].startswith("fe80::"): # prefer non link-local addresses
break
return info
@staticmethod
def extract_wireless_info(interface):
info = dict(essid="", freq="", quality=0.0, quality_bar="")
try:
iwi = basiciw.iwinfo(interface)
except Exception:
# Not a wireless interface
return info
info["essid"] = iwi["essid"]
info["freq"] = iwi["freq"]
quality = iwi["quality"]
if quality["quality_max"] > 0:
info["quality"] = quality["quality"] / quality["quality_max"]
else:
info["quality"] = quality["quality"]
info["quality"] *= 100
info["quality_bar"] = make_bar(info["quality"])
return info
class NetworkTraffic():
"""
Retrieve network traffic information
"""
pnic = None
pnic_before = None
def __init__(self, unknown_up, divisor, round_size):
self.unknown_up = unknown_up
self.divisor = divisor
self.round_size = round_size
def update_counters(self, interface):
self.pnic_before = self.pnic
counters = psutil.net_io_counters(pernic=True)
self.pnic = counters[interface] if interface in counters else None
def clear_counters(self):
self.pnic_before = None
self.pnic = None
def get_bytes_sent(self):
return (self.pnic.bytes_sent - self.pnic_before.bytes_sent) / self.divisor
def get_bytes_received(self):
return (self.pnic.bytes_recv - self.pnic_before.bytes_recv) / self.divisor
def get_packets_sent(self):
return self.pnic.packets_sent - self.pnic_before.packets_sent
def get_packets_received(self):
return self.pnic.packets_recv - self.pnic_before.packets_recv
def get_usage(self, interface):
self.update_counters(interface)
usage = dict(bytes_sent=0, bytes_recv=0, packets_sent=0, packets_recv=0)
if not sysfs_interface_up(interface, self.unknown_up) or not self.pnic_before:
return usage
else:
usage["bytes_sent"] = self.get_bytes_sent()
usage["bytes_recv"] = self.get_bytes_received()
usage["packets_sent"] = self.get_packets_sent()
usage["packets_recv"] = self.get_packets_received()
round_dict(usage, self.round_size)
return usage
class Network(IntervalModule, ColorRangeModule):
"""
Displays network information for an interface.
Requires the PyPI packages `psutil`, `colour`, `netifaces` and `basiciw`
.. rubric:: Available formatters
Network Traffic Formatters:
* `{interface}` — the configured network interface
* `{kbs}` – Float representing kb\s
* `{network_graph}` – Unicode graph representing network usage
* `{bytes_sent}` — bytes sent per second (divided by divisor)
* `{bytes_recv}` — bytes received per second (divided by divisor)
* `{packets_sent}` — bytes sent per second (divided by divisor)
* `{packets_recv}` — bytes received per second (divided by divisor)
Network Information Formatters:
* `{interface}` — same as setting
* `{name}` — same as setting
* `{v4}` — IPv4 address
* `{v4mask}` — subnet mask
* `{v4cidr}` — IPv4 address in cidr notation (i.e. 192.168.2.204/24)
@ -84,91 +231,64 @@ class Network(IntervalModule):
* `{v6cidr}` — IPv6 address in cidr notation
* `{mac}` — MAC of interface
Not available addresses (i.e. no IPv6 connectivity) are replaced with empty strings.
Wireless Information Formatters:
* `{essid}` — ESSID of currently connected wifi
* `{freq}` — Current frequency
* `{quality}` — Link quality in percent
* `{quality_bar}` —Bar graphically representing link quality
"""
settings = (
("interface", "Interface to obtain information for"),
("format_up", "format string"),
("format_down", "format string"),
"color_up",
"color_down",
("interface", "Interface to watch, eg 'eth0'"),
("dynamic_color", "Set color dynamically based on network traffic. Note: this overrides color_up"),
("start_color", "Hex or English name for start of color range, eg '#00FF00' or 'green'"),
("end_color", "Hex or English name for end of color range, eg '#FF0000' or 'red'"),
("graph_width", "Width of the network traffic graph"),
("upper_limit",
"Expected max kb/s. This value controls how the network traffic graph is drawn and in what color"),
("graph_type", "Whether to draw the network traffic graph for input or output. "
"Allowed values 'input' or 'output'"),
("divisor", "divide all byte values by this value"),
("ignore_interfaces", "Array of interfaces to ignore when cycling through "
"with right click. Eg, 'lo'"),
"format_up", "color_up",
"format_down", "color_down",
"on click, eg, ['lo']"),
("round_size", "defines number of digits in round"),
("detached_down", "If the interface doesn't exist, display it as if it were down"),
("unknown_up", "If the interface is in unknown state, display it as if it were up"),
"name",
)
name = interface = "eth0"
ignore_interfaces = ["lo"]
format_up = "{interface}: {v4}"
format_down = "{interface}"
interval = 1
interface = 'eth0'
format_up = "{interface} {network_graph}{kbs}KB/s"
format_down = "{interface}: DOWN"
color_up = "#00FF00"
color_down = "#FF0000"
dynamic_color = True
graph_type = 'input'
graph_width = 15
upper_limit = 150.0
# Network traffic settings
divisor = 1024
round_size = None
# Network info settings
detached_down = True
unknown_up = False
ignore_interfaces = ["lo"]
on_leftclick = "nm-connection-editor"
on_rightclick = "cycle_interface"
interval = 1
def init(self):
if self.interface not in netifaces.interfaces() and not self.detached_down:
raise RuntimeError(
"Unknown interface {iface}!".format(iface=self.interface))
self.network_traffic = NetworkTraffic(self.unknown_up, self.divisor, self.round_size)
self.network_info = NetworkInfo(self.interface, self.ignore_interfaces, self.detached_down, self.unknown_up)
def collect(self):
if self.interface not in netifaces.interfaces() and self.detached_down:
self.format = self.format_down
color = self.color_down
return self.color_down, self.format_down, {"interface": self.interface, "name": self.name}, False
info = netifaces.ifaddresses(self.interface)
slaves = get_bonded_slaves()
try:
master = slaves[self.interface]
except KeyError:
pass
else:
if sysfs_interface_up(self.interface, self.unknown_up):
master_info = netifaces.ifaddresses(master)
for af in (netifaces.AF_INET, netifaces.AF_INET6):
try:
info[af] = master_info[af]
except KeyError:
pass
up = sysfs_interface_up(self.interface, self.unknown_up)
fdict = dict(
zip_longest(["v4", "v4mask", "v4cidr", "v6", "v6mask", "v6cidr"], [], fillvalue=""))
try:
mac = info[netifaces.AF_PACKET][0]["addr"]
except KeyError:
mac = "NONE"
fdict.update({
"interface": self.interface,
"name": self.name,
"mac": mac,
})
if up:
format = self.format_up
color = self.color_up
if netifaces.AF_INET in info:
v4 = info[netifaces.AF_INET][0]
fdict["v4"] = v4["addr"]
fdict["v4mask"] = v4["netmask"]
fdict["v4cidr"] = cidr4(v4["addr"], v4["netmask"])
if netifaces.AF_INET6 in info:
for v6 in info[netifaces.AF_INET6]:
fdict["v6"] = v6["addr"]
fdict["v6mask"] = v6["netmask"]
fdict["v6cidr"] = cidr6(v6["addr"], v6["netmask"])
if not v6["addr"].startswith("fe80::"): # prefer non link-local addresses
break
else:
format = self.format_down
color = self.color_down
return color, format, fdict, up
self.colors = self.get_hex_color_range(self.start_color, self.end_color, int(self.upper_limit))
self.kbs_arr = [0.0] * self.graph_width
def cycle_interface(self):
interfaces = [i for i in netifaces.interfaces() if i not in self.ignore_interfaces]
@ -178,11 +298,50 @@ class Network(IntervalModule):
elif len(interfaces) > 0:
self.interface = interfaces[0]
self.network_traffic.clear_counters()
self.kbs_arr = [0.0] * self.graph_width
def get_network_graph(self, kbs):
# Cycle array by inserting at the start and chopping off the last element
self.kbs_arr.insert(0, kbs)
self.kbs_arr = self.kbs_arr[:self.graph_width]
return make_graph(self.kbs_arr, self.upper_limit)
def run(self):
color, format, fdict, up = self.collect()
format_values = dict(kbs="", network_graph="", bytes_sent="", bytes_recv="", packets_sent="", packets_recv="",
interface="", v4="", v4mask="", v4cidr="", v6="", v6mask="", v6cidr="", mac="",
essid="", freq="", quality="", quality_bar="")
network_usage = self.network_traffic.get_usage(self.interface)
format_values.update(network_usage)
network_info = self.network_info.get_info(self.interface)
format_values.update(network_info)
if self.graph_type == 'input':
kbs = network_usage['bytes_recv']
elif self.graph_type == 'output':
kbs = network_usage['bytes_sent']
else:
raise Exception("graph_type must be either 'input' or 'output'!")
format_values['network_graph'] = self.get_network_graph(kbs)
format_values['kbs'] = "{0:.1f}".format(round(kbs, 2)).rjust(6)
format_values['interface'] = self.interface
if sysfs_interface_up(self.interface, self.unknown_up):
if self.dynamic_color:
color = self.get_gradient(kbs, self.colors, self.upper_limit)
else:
color = self.color_up
self.output = {
"full_text": format.format(**fdict),
"color": color,
"instance": self.interface
"full_text": self.format_up.format(**format_values),
'color': color,
}
else:
color = self.color_down
self.output = {
"full_text": self.format_down.format(**format_values),
'color': color,
}

69
i3pystatus/network_graph.py
View File

@ -1,69 +0,0 @@
# -*- coding: utf-8 -*-
from i3pystatus.core.color import ColorRangeModule
from i3pystatus.network_traffic import NetworkTraffic
from i3pystatus.core.util import make_graph
class NetworkGraph(NetworkTraffic, ColorRangeModule):
"""
Shows Network activity as a Unicode graph
Linux only
Requires the PyPI packages `psutil` and `colour`.
.. rubric:: Available formatters
* {kbs} – Float representing kb\s
* {network_graph} – Unicode network graph
"""
settings = (
("format", "format string"),
("graph_width", "Width of the graph"),
("upper_limit", "Expected max kb/s. This value controls how the graph is drawn and in what color"),
("graph_type", "Whether to draw the graph for input or output. "
"Allowed values 'input' or 'output'"),
("divisor", "divide all byte values by this value"),
("interface", "Interface to watch, eg 'eth0'"),
("start_color", "Hex or English name for start of color range, eg '#00FF00' or 'green'"),
("end_color", "Hex or English name for end of color range, eg '#FF0000' or 'red'")
)
format = "{network_graph}{kbs}KB/s"
graph_type = 'input'
interval = 1
graph_width = 15
upper_limit = 150.0
def init(self):
self.colors = self.get_hex_color_range(self.start_color, self.end_color, int(self.upper_limit))
self.kbs_arr = [0.0] * self.graph_width
def run(self):
self.update_counters()
if not self.pnic_before:
return
if self.graph_type == 'input':
kbs = self.get_bytes_received()
elif self.graph_type == 'output':
kbs = self.get_bytes_sent()
else:
raise Exception("graph_type must be either 'input' or 'output'!")
# Cycle array by inserting at the start and chopping off the last element
self.kbs_arr.insert(0, kbs)
self.kbs_arr = self.kbs_arr[:self.graph_width]
color = self.get_gradient(kbs, self.colors, self.upper_limit)
network_graph = make_graph(self.kbs_arr, self.upper_limit)
self.output = {
"full_text": self.format.format(
network_graph=network_graph,
kbs="{0:.1f}".format(round(kbs, 2)).rjust(6)
),
'color': color,
}

94
i3pystatus/network_traffic.py
View File

@ -1,94 +0,0 @@
from . import IntervalModule
from .core.util import round_dict
import psutil
class NetworkTraffic(IntervalModule):
"""
Network traffic per interface, i.e., packets/bytes sent/received per second.
Requires the PyPI packages `psutil`.
.. rubric:: Available formatters
* `{interface}` — the configured network interface
* `{bytes_sent}` — bytes sent per second (divided by divisor)
* `{bytes_recv}` — bytes received per second (divided by divisor)
* `{packets_sent}` — bytes sent per second (divided by divisor)
* `{packets_recv}` — bytes received per second (divided by divisor)
"""
interval = 1
settings = (
("format", "format string"),
("format_down", "format string if the interface is down (unless hide_down is set)"),
("hide_down", "whether to not display a interface which is down"),
("interface", "network interface"),
("divisor", "divide all byte values by this value"),
("round_size", "defines number of digits in round"),
)
format = "{interface} \u2197{bytes_sent}kB/s \u2198{bytes_recv}kB/s"
format_down = "{interface} \u2013"
hide_down = False
interface = "eth0"
divisor = 1024
round_size = None
pnic = None
pnic_before = None
def update_counters(self):
self.pnic_before = self.pnic
counters = psutil.net_io_counters(pernic=True)
self.pnic = counters[self.interface] if self.interface in counters else None
def get_bytes_sent(self):
return (self.pnic.bytes_sent - self.pnic_before.bytes_sent) / self.divisor
def get_bytes_received(self):
return (self.pnic.bytes_recv - self.pnic_before.bytes_recv) / self.divisor
def get_packets_sent(self):
return self.pnic.packets_sent - self.pnic_before.packets_sent
def get_packets_received(self):
return self.pnic.packets_recv - self.pnic_before.packets_recv
def sysfs_interface_up(self):
try:
sysfs = "/sys/class/net/{}/operstate".format(self.interface)
with open(sysfs) as operstate:
status = operstate.read().strip()
return status == "up" or status == "unknown"
except FileNotFoundError:
return False
def run(self):
self.update_counters()
if self.sysfs_interface_up():
if not self.pnic_before:
return
cdict = {
"bytes_sent": self.get_bytes_sent(),
"bytes_recv": self.get_bytes_received(),
"packets_sent": self.get_packets_sent(),
"packets_recv": self.get_packets_received(),
}
round_dict(cdict, self.round_size)
cdict["interface"] = self.interface
self.output = {
"full_text": self.format.format(**cdict),
"instance": self.interface,
}
elif self.hide_down:
self.output = None
return
else:
cdict = {
"interface": self.interface,
}
self.output = {
"full_text": self.format_down.format(**cdict),
"instance": self.interface,
}

42
i3pystatus/wireless.py
View File

@ -1,42 +0,0 @@
import basiciw
from i3pystatus.core.util import make_bar
from i3pystatus.network import Network
class Wireless(Network):
"""
Display network information about a interface.
Requires the PyPI packages `netifaces` and `basiciw`.
This is based on the network module, so all options and formatters are
the same, except for these additional formatters and that detached_down doesn't work.
* `{essid}` — ESSID of currently connected wifi
* `{freq}` — Current frequency
* `{quality}` — Link quality in percent
* `{quality_bar}` —Bar graphically representing link quality
"""
interface = "wlan0"
def collect(self):
color, format, fdict, up = super().collect()
if up:
iwi = basiciw.iwinfo(self.interface)
fdict["essid"] = iwi["essid"]
fdict["freq"] = iwi["freq"]
quality = iwi["quality"]
if quality["quality_max"] > 0:
fdict["quality"] = quality["quality"] / quality["quality_max"]
else:
fdict["quality"] = quality["quality"]
fdict["quality"] *= 100
fdict["quality_bar"] = make_bar(fdict["quality"])
else:
fdict["essid"] = ""
fdict["freq"] = fdict["quality"] = 0.0
return color, format, fdict, up