statnot (12005B)
1 #!/usr/bin/env python3 2 3 # 4 # statnot - Status and Notifications 5 # 6 # Lightweight notification-(to-become)-deamon intended to be used 7 # with lightweight WMs, like dwm. 8 # Receives Desktop Notifications (including libnotify / notify-send) 9 # See: http://www.galago-project.org/specs/notification/0.9/index.html 10 # 11 # Note: VERY early prototype, to get feedback. 12 # 13 # Copyright (c) 2009-2020 by the authors 14 # 15 # This program is free software; you can redistribute it and/or modify 16 # it under the terms of the GNU General Public License as published by 17 # the Free Software Foundation; either version 2 of the License, or 18 # (at your option) any later version. 19 # 20 # This program is distributed in the hope that it will be useful, 21 # but WITHOUT ANY WARRANTY; without even the implied warranty of 22 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 23 # GNU General Public License for more details. 24 # 25 # You should have received a copy of the GNU General Public License 26 # along with this program. If not, see <http://www.gnu.org/licenses/>. 27 # 28 29 import dbus 30 import dbus.service 31 import dbus.mainloop.glib 32 from gi.repository import GLib 33 import os 34 import subprocess 35 import sys 36 import _thread 37 import time 38 from html.entities import name2codepoint as n2cp 39 import re 40 41 # ===== CONFIGURATION DEFAULTS ===== 42 # 43 # See helpstring below for what each setting does 44 45 DEFAULT_NOTIFY_TIMEOUT = 3000 # milliseconds 46 MAX_NOTIFY_TIMEOUT = 5000 # milliseconds 47 NOTIFICATION_MAX_LENGTH = 100 # number of characters 48 STATUS_UPDATE_INTERVAL = 2.0 # seconds 49 STATUS_COMMAND = ["/bin/sh", "%s/.statusline.sh" % os.getenv("HOME")] 50 USE_STATUSTEXT=True 51 QUEUE_NOTIFICATIONS=True 52 53 # dwm 54 def update_text(text): 55 # Get first line 56 first_line = text.splitlines()[0] if text else '' 57 subprocess.call(["xsetroot", "-name", first_line]) 58 59 # ===== CONFIGURATION END ===== 60 61 def _getconfigvalue(configmodule, name, default): 62 if hasattr(configmodule, name): 63 return getattr(configmodule, name) 64 return default 65 66 def readconfig(filename): 67 import imp 68 try: 69 config = imp.load_source("config", filename) 70 except Exception as e: 71 print(f"Error: failed to read config file {filename}") 72 print(e) 73 sys.exit(2) 74 75 for setting in ("DEFAULT_NOTIFY_TIMEOUT", "MAX_NOTIFY_TIMEOUT", "NOTIFICATION_MAX_LENGTH", "STATUS_UPDATE_INTERVAL", 76 "STATUS_COMMAND", "USE_STATUSTEXT", "QUEUE_NOTIFICATIONS", "update_text"): 77 if hasattr(config, setting): 78 globals()[setting] = getattr(config, setting) 79 80 def strip_tags(value): 81 "Return the given HTML with all tags stripped." 82 return re.sub(r'<[^>]*?>', '', value) 83 84 # from http://snipplr.com/view/19472/decode-html-entities/ 85 # also on http://snippets.dzone.com/posts/show/4569 86 def substitute_entity(match): 87 ent = match.group(3) 88 if match.group(1) == "#": 89 if match.group(2) == '': 90 return unichr(int(ent)) 91 elif match.group(2) == 'x': 92 return unichr(int('0x'+ent, 16)) 93 else: 94 cp = n2cp.get(ent) 95 if cp: 96 return unichr(cp) 97 else: 98 return match.group() 99 100 def decode_htmlentities(string): 101 entity_re = re.compile(r'&(#?)(x?)(\w+);') 102 return entity_re.subn(substitute_entity, string)[0] 103 104 # List of not shown notifications. 105 # Array of arrays: [id, text, timeout in s] 106 # 0th element is being displayed right now, and may change 107 # Replacements of notification happens att add 108 # message_thread only checks first element for changes 109 notification_queue = [] 110 notification_queue_lock = _thread.allocate_lock() 111 112 def add_notification(notif): 113 with notification_queue_lock: 114 for index, n in enumerate(notification_queue): 115 if n[0] == notif[0]: # same id, replace instead of queue 116 n[1:] = notif[1:] 117 return 118 119 notification_queue.append(notif) 120 121 def next_notification(pop = False): 122 # No need to be thread safe here. Also most common scenario 123 if not notification_queue: 124 return None 125 126 with notification_queue_lock: 127 if QUEUE_NOTIFICATIONS: 128 # If there are several pending messages, discard the first 0-timeouts 129 while len(notification_queue) > 1 and notification_queue[0][2] == 0: 130 notification_queue.pop(0) 131 else: 132 while len(notification_queue) > 1: 133 notification_queue.pop(0) 134 135 if pop: 136 return notification_queue.pop(0) 137 else: 138 return notification_queue[0] 139 140 def get_statustext(notification = ''): 141 output = '' 142 try: 143 if not notification: 144 command = STATUS_COMMAND 145 else: 146 command = STATUS_COMMAND + [notification] 147 148 p = subprocess.Popen(command, stdout=subprocess.PIPE) 149 150 output = p.stdout.read() 151 except: 152 sys.stderr.write("%s: could not read status message (%s)\n" 153 % (sys.argv[0], ' '.join(STATUS_COMMAND))) 154 155 # Error - STATUS_COMMAND didn't exist or delivered empty result 156 # Fallback to notification only 157 if not output: 158 output = notification 159 160 return output 161 162 def message_thread(dummy): 163 last_status_update = 0 164 last_notification_update = 0 165 current_notification_text = '' 166 167 while 1: 168 notif = next_notification() 169 current_time = time.time() 170 update_status = False 171 172 if notif: 173 if notif[1] != current_notification_text: 174 update_status = True 175 176 elif current_time > last_notification_update + notif[2]: 177 # If requested timeout is zero, notification shows until 178 # a new notification arrives or a regular status mesasge 179 # cleans it 180 # This way is a bit risky, but works. Keep an eye on this 181 # when changing code 182 if notif[2] != 0: 183 update_status = True 184 185 # Pop expired notification 186 next_notification(True) 187 notif = next_notification() 188 189 if update_status == True: 190 last_notification_update = current_time 191 192 if current_time > last_status_update + STATUS_UPDATE_INTERVAL: 193 update_status = True 194 195 if update_status: 196 if notif: 197 current_notification_text = notif[1] 198 else: 199 current_notification_text = '' 200 201 if USE_STATUSTEXT: 202 update_text(get_statustext(current_notification_text)) 203 else: 204 if current_notification_text != '': 205 update_text(current_notification_text) 206 207 last_status_update = current_time 208 209 time.sleep(0.1) 210 211 class NotificationFetcher(dbus.service.Object): 212 _id = 0 213 214 @dbus.service.method("org.freedesktop.Notifications", 215 in_signature='susssasa{ss}i', 216 out_signature='u') 217 def Notify(self, app_name, notification_id, app_icon, 218 summary, body, actions, hints, expire_timeout): 219 if (expire_timeout < 0) or (expire_timeout > MAX_NOTIFY_TIMEOUT): 220 expire_timeout = DEFAULT_NOTIFY_TIMEOUT 221 222 if not notification_id: 223 self._id += 1 224 notification_id = self._id 225 226 text = (f"{summary} {body}").strip() 227 add_notification( [notification_id, 228 text[:NOTIFICATION_MAX_LENGTH], 229 int(expire_timeout) / 1000.0] ) 230 return notification_id 231 232 @dbus.service.method("org.freedesktop.Notifications", in_signature='', out_signature='as') 233 def GetCapabilities(self): 234 return ("body") 235 236 @dbus.service.signal('org.freedesktop.Notifications', signature='uu') 237 def NotificationClosed(self, id_in, reason_in): 238 pass 239 240 @dbus.service.method("org.freedesktop.Notifications", in_signature='u', out_signature='') 241 def CloseNotification(self, id): 242 pass 243 244 @dbus.service.method("org.freedesktop.Notifications", in_signature='', out_signature='ssss') 245 def GetServerInformation(self): 246 return ("statnot", "http://code.k2h.se", "0.0.2", "1") 247 248 if __name__ == '__main__': 249 for curarg in sys.argv[1:]: 250 if curarg in ('-v', '--version'): 251 print(f"{sys.argv[0]} CURVERSION") 252 sys.exit(1) 253 elif curarg in ('-h', '--help'): 254 print(f" Usage: {sys.argv[0]} [-h] [--help] [-v] [--version] [configuration file]\n" 255 " -h, --help: Print this help and exit\n" 256 " -v, --version: Print version and exit\n" 257 "\n" 258 " Configuration:\n" 259 " A file can be read to set the configuration.\n" 260 " This configuration file must be written in valid python,\n" 261 " which will be read if the filename is given on the command line.\n" 262 " You do only need to set the variables you want to change, and can\n" 263 " leave the rest out.\n" 264 "\n" 265 " Below is an example of a configuration which sets the defaults.\n" 266 "\n" 267 " # Default time a notification is show, unless specified in notification\n" 268 " DEFAULT_NOTIFY_TIMEOUT = 3000 # milliseconds\n" 269 " \n" 270 " # Maximum time a notification is allowed to show\n" 271 " MAX_NOTIFY_TIMEOUT = 5000 # milliseconds\n" 272 " \n" 273 " # Maximum number of characters in a notification.\n" 274 " NOTIFICATION_MAX_LENGTH = 100 # number of characters\n" 275 " \n" 276 " # Time between regular status updates\n" 277 " STATUS_UPDATE_INTERVAL = 2.0 # seconds\n" 278 " \n" 279 " # Command to fetch status text from. We read from stdout.\n" 280 " # Each argument must be an element in the array\n" 281 " # os must be imported to use os.getenv\n" 282 " import os\n" 283 " STATUS_COMMAND = ['/bin/sh', '%s/.statusline.sh' % os.getenv('HOME')]\n" 284 "\n" 285 " # Always show text from STATUS_COMMAND? If false, only show notifications\n" 286 " USE_STATUSTEXT=True\n" 287 " \n" 288 " # Put incoming notifications in a queue, so each one is shown.\n" 289 " # If false, the most recent notification is shown directly.\n" 290 " QUEUE_NOTIFICATIONS=True\n" 291 " \n" 292 " # update_text(text) is called when the status text should be updated\n" 293 " # If there is a pending notification to be formatted, it is appended as\n" 294 " # the final argument to the STATUS_COMMAND, e.g. as $1 in default shellscript\n" 295 "\n" 296 " # dwm statusbar update\n" 297 " import subprocess\n" 298 " def update_text(text):\n" 299 " subprocess.call(['xsetroot', '-name', text])\n") 300 sys.exit(1) 301 else: 302 readconfig(curarg) 303 304 dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) 305 session_bus = dbus.SessionBus() 306 name = dbus.service.BusName("org.freedesktop.Notifications", session_bus) 307 nf = NotificationFetcher(session_bus, '/org/freedesktop/Notifications') 308 309 # We must use contexts and iterations to run threads 310 # http://www.jejik.com/articles/2007/01/python-gstreamer_threading_and_the_main_loop/ 311 # Calling threads_init is not longer needed 312 # https://wiki.gnome.org/PyGObject/Threading 313 #GLib.threads_init() 314 context = GLib.MainLoop().get_context() 315 _thread.start_new_thread(message_thread, (None,)) 316 317 while 1: 318 context.iteration(True)