import os import importlib 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__) import sys """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)) mod = importlib.machinery.SourceFileLoader("modules.{}".format(module_short), os.path.expanduser("~/.config/bumblebee-status/modules/{}.py".format(module_short))).load_module() log.debug("importing {} from user".format(module_short)) return getattr(mod, "Module")(config, theme) except ImportError as e: log.fatal("failed to import {}: {}".format(module_short, e)) 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.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 """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) 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): widget_id = "{}::{}".format(self.name, len(self.widgets())) widget = core.widget.Widget(full_text=full_text, name=name, widget_id=widget_id) 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) """Overriden state, always returns critical (it *is* an error, after all""" def state(self, widget): return ["critical"] # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4