1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
|
import os
import importlib
import importlib.util
import logging
import threading
import core.config
import core.input
import core.widget
import core.decorators
import util.format
try:
error = ModuleNotFoundError("")
except Exception as e:
ModuleNotFoundError = Exception
log = logging.getLogger(__name__)
def import_user(module_short, config, theme):
usermod = os.path.expanduser("~/.config/bumblebee-status/modules/{}.py".format(module_short))
if os.path.exists(usermod):
if hasattr(importlib, "machinery"):
log.debug("importing {} from user via machinery".format(module_short))
mod = importlib.machinery.SourceFileLoader("modules.{}".format(module_short),
os.path.expanduser(usermod)).load_module()
return getattr(mod, "Module")(config, theme)
else:
log.debug("importing {} from user via importlib.util".format(module_short))
try:
spec = importlib.util.spec_from_file_location("modules.{}".format(module_short), usermod)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod.Module(config, theme)
except Exception as e:
spec = importlib.util.find_spec("modules.{}".format(module_short), usermod)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod.Module(config, theme)
raise ImportError("not found")
"""Loads a module by name
:param module_name: Name of the module to load
:param config: Configuration to apply to the module (defaults to an empty configuration)
:param theme: Theme for this module, defaults to None, which means whatever is configured in "config"
:return: A module object representing the module, or an Error module if loading failed
:rtype: class bumblebee_status.module.Module
"""
def load(module_name, config=core.config.Config([]), theme=None):
error = None
module_short, alias = (module_name.split(":") + [module_name])[0:2]
config.set("__alias__", alias)
try:
mod = importlib.import_module("modules.core.{}".format(module_short))
log.debug("importing {} from core".format(module_short))
return getattr(mod, "Module")(config, theme)
except ImportError as e:
try:
log.warning("failed to import {} from core: {}".format(module_short, e))
mod = importlib.import_module("modules.contrib.{}".format(module_short))
log.debug("importing {} from contrib".format(module_short))
return getattr(mod, "Module")(config, theme)
except ImportError as e:
try:
log.warning("failed to import {} from system: {}".format(module_short, e))
return import_user(module_short, config, theme)
except ImportError as e:
log.fatal("import failed: {}".format(e))
log.fatal("failed to import {}".format(module_short))
return Error(config=config, module=module_name, error="unable to load module")
class Module(core.input.Object):
"""Represents a module (single piece of functionality) of the bar
:param config: Configuration to apply to the module (defaults to an empty configuration)
:param theme: Theme for this module, defaults to None, which means whatever is configured in "config"
:param widgets: A list of bumblebee_status.widget.Widget objects that the module is comprised of
"""
def __init__(self, config=core.config.Config([]), theme=None, widgets=[]):
super().__init__()
self.background = False
self.__thread = None
self.__config = config
self.__widgets = widgets if isinstance(widgets, list) else [widgets]
self.module_name = self.__module__.split(".")[-1]
self.name = self.module_name
self.alias = self.__config.get("__alias__", None)
self.id = self.alias if self.alias else self.name
self.next_update = None
self.minimized = False
self.minimized = self.parameter("start-minimized", False)
self.theme = theme
for widget in self.__widgets:
widget.module = self
"""Override this to determine when to show this module
:return: True if the module should be hidden, False otherwise
:rtype: boolean
"""
def hidden(self):
return False
"""Override this to show the module even if it normally would be scrolled away
:return: True if the module should be hidden, False otherwise
:rtype: boolean
"""
def scroll(self):
return True
"""Retrieve CLI/configuration parameters for this module. For example, if
the module is called "test" and the user specifies "-p test.x=123" on the
commandline, using self.parameter("x") retrieves the value 123.
:param key: Name of the parameter to retrieve
:param default: Default value, if parameter is not set by user (defaults to None)
:return: Parameter value, or default
:rtype: string
"""
def parameter(self, key, default=None):
value = default
for prefix in [self.name, self.module_name, self.alias]:
value = self.__config.get("{}.{}".format(prefix, key), value)
if self.minimized:
value = self.__config.get("{}.minimized.{}".format(prefix, key), value)
return value
"""Set a parameter for this module
:param key: Name of the parameter to set
:param value: New value of the parameter
"""
def set(self, key, value):
self.__config.set("{}.{}".format(self.name, key), value)
"""Override this method to define tasks that should be done during each
update interval (for instance, querying an API, calling a CLI tool to get new
date, etc.
"""
def update(self):
pass
def update_wrapper(self):
if self.background == True:
if self.__thread and self.__thread.is_alive():
return # skip this update interval
self.__thread = threading.Thread(target=self.internal_update, args=(True,))
self.__thread.start()
else:
self.internal_update(False)
"""Wrapper method that ensures that all exceptions thrown by the
update() method are caught and displayed in a bumblebee_status.module.Error
module
"""
def internal_update(self, trigger_redraw=False):
try:
self.update()
if trigger_redraw:
core.event.trigger("update", [self.id], redraw_only=True)
except Exception as e:
self.set("interval", 1)
module = Error(config=self.__config, module="error", error=str(e))
self.__widgets = [module.widget()]
self.update = module.update
"""Retrieves the list of widgets for this module
:return: A list of widgets
:rtype: list of bumblebee_status.widget.Widgets
"""
def widgets(self):
return self.__widgets
"""Removes all widgets of this module"""
def clear_widgets(self):
del self.__widgets[:]
"""Adds a widget to the module
:param full_text: Text or callable (method) that defines the text of the widget (defaults to "")
:param name: Name of the widget, defaults to None, which means autogenerate
:return: The new widget
:rtype: bumblebee_status.widget.Widget
"""
def add_widget(self, full_text="", name=None, hidden=False):
widget_id = "{}::{}".format(self.name, len(self.widgets()))
widget = core.widget.Widget(full_text=full_text, name=name, widget_id=widget_id, hidden=hidden)
self.widgets().append(widget)
widget.module = self
return widget
"""Convenience method to retrieve a named widget
:param name: Name of widget to retrieve, defaults to None (in which case the first widget is returned)
:return: The widget with the corresponding name, None if not found
:rtype: bumblebee_status.widget.Widget
"""
def widget(self, name=None, widget_id=None):
if not name and not widget_id:
return self.widgets()[0]
for w in self.widgets():
if name and w.name == name:
return w
if w.id == widget_id:
return w
return None
"""Override this method to define states for the module
:param widget: Widget for which state should be returned
:return: a list of states for this widget
:rtype: list of strings
"""
def state(self, widget):
return []
"""Convenience method that sets warning and critical state for numbers
:param value: Current value to calculate state against
:param warn: Warning threshold
:parm crit: Critical threshold
:return: None if value is below both thresholds, "critical", "warning" as appropriate otherwise
:rtype: string
"""
def threshold_state(self, value, warn, crit):
if value > float(self.parameter("critical", crit)):
return "critical"
if value > float(self.parameter("warning", warn)):
return "warning"
return None
def register_callbacks(self):
actions = [
{"name": "left-click", "id": core.input.LEFT_MOUSE},
{"name": "right-click", "id": core.input.RIGHT_MOUSE},
{"name": "middle-click", "id": core.input.MIDDLE_MOUSE},
{"name": "wheel-up", "id": core.input.WHEEL_UP},
{"name": "wheel-down", "id": core.input.WHEEL_DOWN},
]
for action in actions:
if self.parameter(action["name"]):
core.input.register(
self,
action["id"],
self.parameter(action["name"]),
util.format.asbool(
self.parameter("{}-wait".format(action["name"]), False)
),
)
class Error(Module):
"""Represents an "error" module
:param module: The module name that produced the error
:param error: The error message to display
:param config: Configuration to apply to the module (defaults to an empty configuration)
:param theme: Theme for this module, defaults to None, which means whatever is configured in "config"
"""
def __init__(self, module, error, config=core.config.Config([]), theme=None):
super().__init__(config, theme, core.widget.Widget(self.full_text))
self.__module = module
self.__error = error
"""Returns the error message
:param widget: the error widget to display
"""
def full_text(self, widget):
return "{}: {}".format(self.__module, self.__error)
"""Overridden state, always returns critical (it *is* an error, after all"""
def state(self, widget):
return ["critical"]
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|
