This is a simple project to receive D-Bus desktop notifications and display them with a systray application.
Purposefully displaying notifications only when the systray application is right-clicked, if OSD notifications is your preference there are a multitude of options available: notification-daemon, notify-osd, dunst, etc.
| #!/usr/bin/env python3 | |
| # Systray notifier | |
| import wx, wx.adv | |
| import dbus, dbus.mainloop.glib, dbus.service | |
| import gi.repository.GLib | |
| import copy, datetime, logging, multiprocessing, threading | |
| class App(wx.App): | |
| class TaskBarIcon(wx.adv.TaskBarIcon): | |
| def __init__(self, app): | |
| assert super().IsAvailable() | |
| super().__init__() | |
| self.app = app | |
| self.update_icon(False) | |
| def update_icon(self, present): | |
| # apt-get install gnome-icon-theme | |
| path = '/usr/share/icons/gnome/16x16/status/' + ('user-available.png' if present else 'user-invisible.png') | |
| self.SetIcon(wx.Icon(path)) | |
| def CreatePopupMenu(self): | |
| notes = self.app.get_notes() | |
| if len(notes) == 0: | |
| return | |
| def delete(event): | |
| self.app.del_note(event.GetId()) | |
| menu = wx.Menu() | |
| for i in range(len(notes)): | |
| if i > 0: | |
| menu.AppendSeparator() | |
| menu.Append(wx.MenuItem(menu, i, notes[i])) | |
| menu.Bind(wx.EVT_MENU, delete) | |
| return menu | |
| def OnInit(self): | |
| frame = wx.Frame(None) | |
| self.SetTopWindow(frame) | |
| self.__systray = self.TaskBarIcon(self) | |
| self.__notes = [] | |
| self.__lock = threading.Lock() | |
| return True | |
| def add_note(self, note): | |
| self.__lock.acquire() | |
| self.__notes += [note] | |
| was_empty = len(self.__notes) == 1 | |
| self.__lock.release() | |
| # https://wiki.wxpython.org/CallAfter | |
| if was_empty: | |
| wx.CallAfter(self.__systray.update_icon, True) | |
| def get_notes(self): | |
| self.__lock.acquire() | |
| rv = copy.copy(self.__notes) | |
| self.__lock.release() | |
| return rv | |
| def del_note(self, i): | |
| self.__lock.acquire() | |
| self.__notes.pop(i) | |
| empty = len(self.__notes) == 0 | |
| self.__lock.release() | |
| # https://wiki.wxpython.org/CallAfter | |
| if empty: | |
| wx.CallAfter(self.__systray.update_icon, False) | |
| def dbus_daemon(queue): | |
| # Refs: https://www.xpra.org/trac/ticket/22?cversion=0&cnum_hist=13 | |
| # https://developer.gnome.org/notification-spec/#protocol | |
| class NoteService(dbus.service.Object): | |
| def __init__(self): | |
| bus_name = dbus.service.BusName('org.freedesktop.Notifications', bus=dbus.SessionBus()) | |
| super().__init__(bus_name, '/org/freedesktop/Notifications') | |
| @dbus.service.method('org.freedesktop.Notifications') | |
| def GetCapabilities(self): | |
| return [] | |
| @dbus.service.method('org.freedesktop.Notifications', in_signature='sisssa{is}a{is}i', out_signature='u') | |
| def Notify(self, app_name, replaces_id, app_icon, summary, body, actions, hints, expire_timeout): | |
| note = str(summary) | |
| if str(body): | |
| note += '\n' + str(body) | |
| logging.info('received: %s', note) | |
| note = datetime.datetime.now().strftime('%a %m/%d %H:%M:%S - ') + note | |
| queue.put(note) | |
| return 0 | |
| @dbus.service.method('org.freedesktop.Notifications', in_signature='i') | |
| def CloseNotification(self, id): | |
| pass | |
| @dbus.service.method('org.freedesktop.Notifications', out_signature='ssss') | |
| def GetServerInformation(self): | |
| return ['systray-notify', 'vendor', '1.0.0', '1.12'] | |
| logging.basicConfig(format='%(asctime)s:%(levelname)s:%(funcName)s:%(message)s', level=logging.INFO) | |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) | |
| NoteService() | |
| gi.repository.GLib.MainLoop().run() | |
| if __name__ == '__main__': | |
| # The two frameworks (wxpython & dbus) have their own main loops, hence use seperate processes | |
| queue = multiprocessing.Queue() | |
| multiprocessing.Process(target=dbus_daemon, args=(queue,)).start() | |
| app = App() | |
| def update(): | |
| while True: | |
| app.add_note(queue.get()) | |
| threading.Thread(target=update).start() | |
| app.MainLoop() |
