/sys_notify.py
Created Jul 29, 2018
Qt DBus sys notify module for Linux systems
| #!/usr/bin/env python | |
| # Copyright (c) 2018 Kurt Jacobson | |
| # <kurtcjacobson@gmail.com> | |
| # | |
| # This program is free software: you can redistribute it and/or modify | |
| # it under the terms of the GNU General Public License as published by | |
| # the Free Software Foundation, either version 2 of the License, or | |
| # (at your option) any later version. | |
| # | |
| # This program is distributed in the hope that it will be useful, | |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
| # GNU General Public License for more details. | |
| import dbus | |
| from collections import OrderedDict | |
| DBusQtMainLoop = None | |
| try: | |
| from dbus.mainloop.pyqt5 import DBusQtMainLoop | |
| except: | |
| print "Could not import DBusQtMainLoop, is package 'python-dbus.mainloop.pyqt5' installed?" | |
| APP_NAME = '' | |
| DBUS_IFACE = None | |
| NOTIFICATIONS = {} | |
| class Urgency: | |
| """freedesktop.org notification urgency levels""" | |
| LOW, NORMAL, CRITICAL = range(3) | |
| class UninitializedError(RuntimeError): | |
| """Error raised if you try to show an error before initializing""" | |
| pass | |
| def init(app_name): | |
| """Initializes the DBus connection""" | |
| global APP_NAME, DBUS_IFACE | |
| APP_NAME = app_name | |
| name = "org.freedesktop.Notifications" | |
| path = "/org/freedesktop/Notifications" | |
| interface = "org.freedesktop.Notifications" | |
| mainloop = None | |
| if DBusQtMainLoop is not None: | |
| mainloop = DBusQtMainLoop(set_as_default=True) | |
| bus = dbus.SessionBus(mainloop) | |
| proxy = bus.get_object(name, path) | |
| DBUS_IFACE = dbus.Interface(proxy, interface) | |
| if mainloop is not None: | |
| # We have a mainloop, so connect callbacks | |
| DBUS_IFACE.connect_to_signal('ActionInvoked', _onActionInvoked) | |
| DBUS_IFACE.connect_to_signal('NotificationClosed', _onNotificationClosed) | |
| def _onActionInvoked(nid, action): | |
| """Called when a notification action is clicked""" | |
| nid, action = int(nid), str(action) | |
| try: | |
| notification = NOTIFICATIONS[nid] | |
| except KeyError: | |
| # must have been created by some other program | |
| return | |
| notification._onActionInvoked(action) | |
| def _onNotificationClosed(nid, reason): | |
| """Called when the notification is closed""" | |
| nid, reason = int(nid), int(reason) | |
| try: | |
| notification = NOTIFICATIONS[nid] | |
| except KeyError: | |
| # must have been created by some other program | |
| return | |
| notification._onNotificationClosed(notification) | |
| del NOTIFICATIONS[nid] | |
| class Notification(object): | |
| """Notification object""" | |
| id = 0 | |
| timeout = -1 | |
| _onNotificationClosed = lambda *args: None | |
| def __init__(self, title, body='', icon='', timeout=-1): | |
| """Initializes a new notification object. | |
| Args: | |
| title (str): The title of the notification | |
| body (str, optional): The body text of the notification | |
| icon (str, optional): The icon to display with the notification | |
| timeout (TYPE, optional): The time in ms before the notification hides, -1 for default, 0 for never | |
| """ | |
| self.title = title # title of the notification | |
| self.body = body # the body text of the notification | |
| self.icon = icon # the path to the icon to use | |
| self.timeout = timeout # time in ms before the notification disappears | |
| self.hints = {} # dict of various display hints | |
| self.actions = OrderedDict() # actions names and their callbacks | |
| self.data = {} # arbitrary user data | |
| def show(self): | |
| if DBUS_IFACE is None: | |
| raise UninitializedError("You must call 'notify.init()' before 'notify.show()'") | |
| """Asks the notification server to show the notification""" | |
| nid = DBUS_IFACE.Notify(APP_NAME, | |
| self.id, | |
| self.icon, | |
| self.title, | |
| self.body, | |
| self._makeActionsList(), | |
| self.hints, | |
| self.timeout, | |
| ) | |
| self.id = int(nid) | |
| NOTIFICATIONS[self.id] = self | |
| return True | |
| def close(self): | |
| """Ask the notification server to close the notification""" | |
| if self.id != 0: | |
| DBUS_IFACE.CloseNotification(self.id) | |
| def onClosed(self, callback): | |
| """Set the callback called when the notification is closed""" | |
| self._onNotificationClosed = callback | |
| def setUrgency(self, value): | |
| """Set the freedesktop.org notification urgency level""" | |
| if value not in range(3): | |
| raise ValueError("Unknown urgency level '%s' specified" % level) | |
| self.hints['urgency'] = dbus.Byte(value) | |
| def setSoundFile(self, sound_file): | |
| """Sets a sound file to play when the notification shows""" | |
| self.hints['sound-file'] = sound_file | |
| def setSoundName(self, sound_name): | |
| """Set a freedesktop.org sound name to play when notification shows""" | |
| self.hints['sound-name'] = sound_name | |
| def setIconPath(self, icon_path): | |
| """Set the URI of the icon to display in the notification""" | |
| self.hints['image-path'] = 'file://' + icon_path | |
| def setQIcon(self, q_icon): | |
| # FixMe this would be convenient, but may not be possible | |
| raise NotImplemented | |
| def setLocation(self, x_pos, y_pos): | |
| """Sets the location to display the notification""" | |
| self.hints['x'] = int(x_pos) | |
| self.hints['y'] = int(y_pos) | |
| def setCategory(self, category): | |
| """Sets the the freedesktop.org notification category""" | |
| self.hints['category'] = category | |
| def setTimeout(self, timeout): | |
| """Set the display duration in milliseconds, -1 for default""" | |
| if not isinstance(timeout, int): | |
| raise TypeError("Timeout value '%s' was not int" % timeout) | |
| self.timeout = timeout | |
| def setHint(self, key, value): | |
| """Set one of the other hints""" | |
| self.hints[key] = value | |
| def addAction(self, action, label, callback, user_data=None): | |
| """Add an action to the notification. | |
| Args: | |
| action (str): A sort key identifying the action | |
| label (str): The text to display on the action button | |
| callback (bound method): The method to call when the action is activated | |
| user_data (any, optional): Any user data to be passed to the action callback | |
| """ | |
| self.actions[action] = (label, callback, user_data) | |
| def _makeActionsList(self): | |
| """Make the actions array to send over DBus""" | |
| arr = [] | |
| for action, (label, callback, user_data) in self.actions.items(): | |
| arr.append(action) | |
| arr.append(label) | |
| return arr | |
| def _onActionInvoked(self, action): | |
| """Called when the user activates a notification action""" | |
| try: | |
| label, callback, user_data = self.actions[action] | |
| except KeyError: | |
| return | |
| if user_data is None: | |
| callback(self, action) | |
| else: | |
| callback(self, action, user_data) | |
| # ----------------------- E X A M P L E ----------------------- | |
| def onHelp(n, action): | |
| assert(action == "help"), "Action was not help!" | |
| print "You clicked Help action" | |
| n.close() | |
| def onIgnore(n, action, data): | |
| assert(action == "ignore"), "Action was not ignore!" | |
| print "You clicked Ignore action" | |
| print "Passed user data was: ", data | |
| n.close() | |
| def onClose(n): | |
| print "Notification closed" | |
| app.quit() | |
| if __name__ == "__main__": | |
| import sys | |
| from PyQt5.QtCore import QCoreApplication | |
| app = QCoreApplication(sys.argv) | |
| # Initialize the DBus connection to the notification server | |
| init("demo") | |
| # Initialize a new notification object | |
| n = Notification("Demo Notification", | |
| "This notification is very important as it " + | |
| "notifies you that notifications are working.", | |
| timeout=3000 | |
| ) | |
| n.setUrgency(Urgency.NORMAL) | |
| n.setCategory("device") | |
| n.setIconPath("/usr/share/icons/Tango/scalable/status/dialog-error.svg") | |
| # no user data | |
| n.addAction("help", "Help", onHelp) | |
| # passing arbitrary user data to the callback | |
| n.addAction("ignore", "Ignore", onIgnore, 12345) | |
| n.onClosed(onClose) | |
| n.show() | |
| app.exec_() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment