[modules/nic] Re-enable NIC module
Re-add the NIC module with all its functionality (hopefully...). This introduces a new concept: Instead of having separate queries for critical and warning (which really are just another set of states), a module can now return a list of states for each widget. All the state information is then merged together into a single theme. So, for instance, the NIC module can return a state saying "critical - wlan-down", which applies the theme information for both "critical" and "wlan-down". see #23
This commit is contained in:
parent
c820223d0c
commit
a045962d00
8 changed files with 104 additions and 19 deletions
|
@ -80,3 +80,5 @@ class Module(bumblebee.engine.Module):
|
|||
self._repeat = False if "false" in line else True
|
||||
if line.startswith("set shuffle "):
|
||||
self._shuffle = False if "false" in line else True
|
||||
|
||||
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|
||||
|
|
67
bumblebee/modules/nic.py
Normal file
67
bumblebee/modules/nic.py
Normal file
|
@ -0,0 +1,67 @@
|
|||
#pylint: disable=C0111,R0903
|
||||
|
||||
import netifaces
|
||||
|
||||
import bumblebee.util
|
||||
import bumblebee.input
|
||||
import bumblebee.output
|
||||
import bumblebee.engine
|
||||
|
||||
"""Displays the name, IP address(es) and status of each available network interface."""
|
||||
|
||||
class Module(bumblebee.engine.Module):
|
||||
def __init__(self, engine, config):
|
||||
widgets = []
|
||||
super(Module, self).__init__(engine, config, widgets)
|
||||
self._exclude = tuple(filter(len, self.parameter("exclude", "lo,virbr,docker,vboxnet,veth").split(",")))
|
||||
self._update_widgets(widgets)
|
||||
|
||||
def update(self, widgets):
|
||||
self._update_widgets(widgets)
|
||||
|
||||
def state(self, widget):
|
||||
states = []
|
||||
|
||||
if widget.get("state") == "down":
|
||||
states.append("critical")
|
||||
elif widget.get("state") != "up":
|
||||
states.append("warning")
|
||||
|
||||
intf = widget.get("intf")
|
||||
iftype = "wireless" if self._iswlan(intf) else "wired"
|
||||
iftype = "tunnel" if self._istunnel(intf) else iftype
|
||||
|
||||
states.append("{}-{}".format(iftype, widget.get("state")))
|
||||
|
||||
return states
|
||||
|
||||
def _iswlan(self, intf):
|
||||
# wifi, wlan, wlp, seems to work for me
|
||||
if intf.startswith("w"): return True
|
||||
return False
|
||||
|
||||
def _istunnel(self, intf):
|
||||
return intf.startswith("tun")
|
||||
|
||||
def _update_widgets(self, widgets):
|
||||
interfaces = [ i for i in netifaces.interfaces() if not i.startswith(self._exclude) ]
|
||||
for intf in interfaces:
|
||||
addr = []
|
||||
state = "down"
|
||||
try:
|
||||
if netifaces.AF_INET in netifaces.ifaddresses(intf):
|
||||
for ip in netifaces.ifaddresses(intf)[netifaces.AF_INET]:
|
||||
if "addr" in ip and ip["addr"] != "":
|
||||
addr.append(ip["addr"])
|
||||
state = "up"
|
||||
except Exception as e:
|
||||
addr = []
|
||||
widget = self.widget(intf)
|
||||
if not widget:
|
||||
widget = bumblebee.output.Widget(name=intf)
|
||||
widgets.append(widget)
|
||||
widget.full_text("{} {} {}".format(intf, state, ", ".join(addr)))
|
||||
widget.set("intf", intf)
|
||||
widget.set("state", state)
|
||||
|
||||
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|
|
@ -6,9 +6,12 @@ import sys
|
|||
import json
|
||||
import uuid
|
||||
|
||||
class Widget(object):
|
||||
import bumblebee.store
|
||||
|
||||
class Widget(bumblebee.store.Store):
|
||||
"""Represents a single visible block in the status bar"""
|
||||
def __init__(self, full_text="", name=""):
|
||||
super(Widget, self).__init__()
|
||||
self._full_text = full_text
|
||||
self.module = None
|
||||
self._module = None
|
||||
|
@ -26,11 +29,17 @@ class Widget(object):
|
|||
def state(self):
|
||||
"""Return the widget's state"""
|
||||
if self._module and hasattr(self._module, "state"):
|
||||
return self._module.state(self)
|
||||
return None
|
||||
states = self._module.state(self)
|
||||
if not isinstance(states, list):
|
||||
return [states]
|
||||
return states
|
||||
return []
|
||||
|
||||
def full_text(self):
|
||||
"""Retrieve the full text to display in the widget"""
|
||||
def full_text(self, value=None):
|
||||
"""Set or retrieve the full text to display in the widget"""
|
||||
if value:
|
||||
self._full_text = value
|
||||
else:
|
||||
if callable(self._full_text):
|
||||
return self._full_text()
|
||||
else:
|
||||
|
|
|
@ -118,16 +118,19 @@ class Theme(object):
|
|||
self._cycle = self._cycles[self._cycle_idx]
|
||||
|
||||
module_theme = self._theme.get(widget.module, {})
|
||||
if name != widget.state():
|
||||
|
||||
state_themes = []
|
||||
# avoid infinite recursion
|
||||
state_theme = self._get(widget, widget.state(), {})
|
||||
else:
|
||||
state_theme = {}
|
||||
if name not in widget.state():
|
||||
for state in widget.state():
|
||||
state_themes.append(self._get(widget, state, {}))
|
||||
|
||||
value = self._defaults.get(name, default)
|
||||
value = self._cycle.get(name, value)
|
||||
value = module_theme.get(name, value)
|
||||
value = state_theme.get(name, value)
|
||||
|
||||
for theme in state_themes:
|
||||
value = theme.get(name, value)
|
||||
|
||||
return value
|
||||
|
||||
|
|
|
@ -43,7 +43,7 @@ class TestCPUModule(unittest.TestCase):
|
|||
self.config.set("cpu.warning", "18")
|
||||
mock_psutil.return_value = 19.0
|
||||
self.module.update(self.module.widgets())
|
||||
self.assertEquals(self.module.widgets()[0].state(), "warning")
|
||||
self.assertEquals(self.module.widgets()[0].state(), ["warning"])
|
||||
|
||||
@mock.patch("psutil.cpu_percent")
|
||||
def test_critical(self, mock_psutil):
|
||||
|
@ -51,6 +51,6 @@ class TestCPUModule(unittest.TestCase):
|
|||
self.config.set("cpu.warning", "19")
|
||||
mock_psutil.return_value = 21.0
|
||||
self.module.update(self.module.widgets())
|
||||
self.assertEquals(self.module.widgets()[0].state(), "critical")
|
||||
self.assertEquals(self.module.widgets()[0].state(), ["critical"])
|
||||
|
||||
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|
||||
|
|
|
@ -15,6 +15,8 @@ class TestGenericModules(unittest.TestCase):
|
|||
for mod in all_modules():
|
||||
cls = importlib.import_module("bumblebee.modules.{}".format(mod["name"]))
|
||||
self.objects[mod["name"]] = getattr(cls, "Module")(engine, {"config": config})
|
||||
for widget in self.objects[mod["name"]].widgets():
|
||||
self.assertEquals(widget.get("variable", None), None)
|
||||
|
||||
def test_widgets(self):
|
||||
for mod in self.objects:
|
||||
|
@ -23,6 +25,8 @@ class TestGenericModules(unittest.TestCase):
|
|||
widget.link_module(self.objects[mod])
|
||||
self.assertEquals(widget.module, mod)
|
||||
assertWidgetAttributes(self, widget)
|
||||
widget.set("variable", "value")
|
||||
self.assertEquals(widget.get("variable", None), "value")
|
||||
|
||||
def test_update(self):
|
||||
for mod in self.objects:
|
||||
|
|
|
@ -102,11 +102,11 @@ class TestTheme(unittest.TestCase):
|
|||
self.assertEquals(theme.fg(self.anyWidget), data["defaults"]["fg"])
|
||||
self.assertEquals(theme.bg(self.anyWidget), data["defaults"]["bg"])
|
||||
|
||||
self.anyWidget.attr_state = "critical"
|
||||
self.anyWidget.attr_state = ["critical"]
|
||||
self.assertEquals(theme.fg(self.anyWidget), data["defaults"]["critical"]["fg"])
|
||||
self.assertEquals(theme.bg(self.anyWidget), data["defaults"]["critical"]["bg"])
|
||||
|
||||
self.themedWidget.attr_state = "critical"
|
||||
self.themedWidget.attr_state = ["critical"]
|
||||
self.assertEquals(theme.fg(self.themedWidget), data[self.widgetTheme]["critical"]["fg"])
|
||||
# if elements are missing in the state theme, they are taken from the
|
||||
# widget theme instead (i.e. no fallback to a more general state theme)
|
||||
|
|
|
@ -70,7 +70,7 @@ class MockWidget(Widget):
|
|||
super(MockWidget, self).__init__(text)
|
||||
self._text = text
|
||||
self.module = None
|
||||
self.attr_state = "state-default"
|
||||
self.attr_state = ["state-default"]
|
||||
self.id = "none"
|
||||
|
||||
def state(self):
|
||||
|
|
Loading…
Reference in a new issue