-
Notifications
You must be signed in to change notification settings - Fork 5
/
daemon.py
executable file
·781 lines (677 loc) · 30.4 KB
/
daemon.py
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
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
#!/usr/bin/env python
import collections as cs, itertools as it, operator as op, functools as ft
import os, sys, traceback, math, time
from dbus.mainloop.glib import DBusGMainLoop
import dbus, dbus.service
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Gdk', '3.0')
try: from gi.repository import GLib, Gdk
except RuntimeError as err: # less verbose errors in case X isn't running
print(f'Gdk init error, exiting: {err.message}', file=sys.stderr)
sys.exit(1)
if __name__ == '__main__':
# Try to import submodules from the same path, not the site-packages
from os.path import join, realpath, dirname
module_root = realpath(dirname(dirname(__file__)))
if module_root not in sys.path: sys.path.insert(0, module_root)
from notification_thing.display import NotificationDisplay, strip_markup
from notification_thing.pubsub import PubSub
from notification_thing.file_logger import FileLogger
from notification_thing import core
else:
from .display import NotificationDisplay
from .pubsub import PubSub
from .file_logger import FileLogger
from . import core
optz, poll_interval, close_reasons, urgency_levels =\
core.optz, core.poll_interval, core.close_reasons, core.urgency_levels
def flatten_dict(data, path=tuple()):
dst = list()
for k,v in data.items():
k = path + (k,)
if isinstance(v, cs.abc.Mapping):
for v in flatten_dict(v, k): dst.append(v)
else: dst.append((k, v))
return dst
def ts_diff_format( seconds, add_ago=False,
_units_days=dict(y=365.25, mo=30.5, w=7, d=0),
_units_s=dict(h=3600, m=60, s=0) ):
days = seconds // (24*3600)
seconds -= days * (24*3600)
res, d = list(), days
for unit, unit_days in sorted(
_units_days.items(), key=op.itemgetter(1), reverse=True):
if d > unit_days or res:
res.append('{0:.0f}{1}'.format(
math.floor(d / unit_days) if unit_days else d, unit ))
if len(res) >= 2 or not unit_days: break
d = days % unit_days
if len(res) < 2:
s = seconds
for unit, unit_s in sorted(
_units_s.items(), key=op.itemgetter(1), reverse=True):
if s > unit_s or res:
res.append('{0:.0f}{1}'.format(s / unit_s if unit_s else s, unit))
if len(res) >= 2 or not unit_s: break
s = seconds % unit_s
if not res: return 'just now'
else:
if add_ago: res.append('ago')
return ' '.join(res)
class NotificationDaemon(dbus.service.Object):
dbus_props = dbus.PROPERTIES_IFACE
dbus_iface = 'org.freedesktop.Notifications'
dbus_path = '/org/freedesktop/Notifications'
plugged, timeout_cleanup = False, True
_activity_timer = None
def __init__(self, bus, pubsub=None, logger=None):
# super().__init__(bus, self.dbus_path)
dbus.service.Object.__init__( self, bus,
self.dbus_path, dbus.service.BusName(self.dbus_iface, bus) )
tick_strangle_max = op.truediv(optz.tbf_max_delay, optz.tbf_tick)
self._note_limit = core.FC_TokenBucket(
tick=optz.tbf_tick, burst=optz.tbf_size,
tick_strangle=lambda x: min(x*optz.tbf_inc, tick_strangle_max),
tick_free=lambda x: max(op.truediv(x, optz.tbf_dec), 1) )
self._note_buffer = core.RRQ(optz.queue_len)
self._note_history = core.RRQ(optz.history_len)
self._note_windows = dict()
self._note_id_pool = it.chain.from_iterable(map(ft.partial(range, 1), it.repeat(2**30)))
self._renderer = NotificationDisplay(
optz.layout_margin, optz.layout_anchor, optz.layout_direction,
icon_scale=optz.icon_scale, markup_default=not optz.markup_disable,
markup_warn=optz.markup_warn_on_err, markup_strip=optz.markup_strip_on_err )
self._activity_event()
self.pubsub = pubsub
if pubsub:
GLib.io_add_watch( pubsub.fileno(),
GLib.PRIORITY_DEFAULT, GLib.IO_IN | GLib.IO_PRI, self._notify_pubsub )
self.logger = logger
if optz.test_message:
# Also test crazy web-of-90s markup here :P
summary = 'Notification daemon started <small><tt>¯\(°_o)/¯</tt></small>'
body = (
'Desktop notification daemon started successfully on host: <u>{host}</u>'
'\nVersion: <span stretch="extraexpanded">{v}</span>'
'\nCode path: <small>{code}</small>'
'\nSound enabled: <span color="{sound_color}">{sound}</span>'
'\nPubSub enabled: <span color="{pubsub_color}">{pubsub}</span>' )\
.format(
host=os.uname()[1], v=core.__version__,
sound_color='green' if optz.filter_sound else 'red',
sound=str(bool(optz.filter_sound)).lower(),
pubsub_color='green' if pubsub else 'red',
pubsub=str(bool(pubsub)).lower(),
code=os.path.abspath(os.path.dirname(core.__file__)) )
if not self._renderer.markup_default:
summary, body = map(strip_markup, [summary, body])
self.display(summary, body)
if optz.test_sound and optz.filter_sound: optz.filter_sound['play'](optz.test_sound)
def exit(self, reason=None):
log.debug(f'Exiting cleanly%s', ', reason: {reason or ""}')
sys.exit()
def _activity_event(self, callback=False):
if callback:
if not self._note_windows:
self.exit(reason=f'activity timeout ({optz.activity_timeout}s)')
else:
log.debug( 'Ignoring inacivity timeout event'
' due to existing windows (retry in %ss).', optz.activity_timeout )
self._activity_timer = None
if self._activity_timer: GLib.source_remove(self._activity_timer)
if optz.activity_timeout and optz.activity_timeout > 0:
self._activity_timer = GLib.timeout_add_seconds(
optz.activity_timeout, self._activity_event, True )
@dbus.service.method(dbus_iface, '', 'ssss')
def GetServerInformation(self):
self._activity_event()
return 'notification-thing', 'mk.fraggod@gmail.com', 'git', '1.2'
@dbus.service.method(dbus_iface, '', 'as')
def GetCapabilities(self):
# action-icons, actions, body, body-hyperlinks, body-images,
# body-markup, icon-multi, icon-static, persistence, sound
self._activity_event()
caps = ['body', 'persistence', 'icon-static']
if not self._renderer.markup_default: caps.append('body-markup')
return sorted(caps)
@dbus.service.signal(dbus_iface, 'uu')
def NotificationClosed(self, nid, reason=None):
log.debug(
'NotificationClosed signal (id: %s, reason: %s)',
nid, close_reasons.by_id(reason) )
@dbus.service.signal(dbus_iface, 'us')
def ActionInvoked(self, nid, action_key):
log.debug('Um... some action invoked? Params: %s', [nid, action_key])
@dbus.service.method(dbus_props, 'ss', 'v')
def Get(self, iface, k):
return self.GetAll(iface)[k]
@dbus.service.method(dbus_props, 's', 'a{sv}')
def GetAll(self, iface):
self._activity_event()
return dict( urgent=optz.urgency_check,
plug=self.plugged, cleanup=self.timeout_cleanup )
@dbus.service.method(dbus_props, 'ssv', '')
def Set(self, iface, k, v):
self._activity_event()
if isinstance(v, str):
if v == 'toggle': k, v = f'{k}_toggle', True
elif v.lower() in {'n', 'no', 'false', 'disable', 'off', '-'}: v = False
k, v = str(k), bool(v)
if k.endswith('_toggle'):
k = k[:-7]
v = not self.Get(iface, k)
log.debug('Property change: %s = %s', k, v)
if k == 'urgent':
if optz.urgency_check == v: return
optz.urgency_check = v
if optz.status_notify:
self.display(
'Urgent messages passthrough {}'.format(
'enabled' if optz.urgency_check else 'disabled' ) )
elif k == 'plug':
if self.plugged == v: return
if v:
self.plugged = True
log.debug('Notification queue plugged')
if optz.status_notify:
self.display( 'Notification proxy: queue is plugged',
'Only urgent messages will be passed through'
if optz.urgency_check else 'All messages will be stalled' )
else:
self.plugged = False
log.debug('Notification queue unplugged')
if optz.status_notify:
self.display('Notification proxy: queue is unplugged')
if self._note_buffer:
log.debug('Flushing plugged queue')
self.flush()
elif k == 'cleanup':
if self.timeout_cleanup == v: return
self.timeout_cleanup = v
log.debug('Cleanup timeout: %s', self.timeout_cleanup)
if optz.status_notify:
self.display( 'Notification proxy: cleanup timeout is {}'\
.format('enabled' if self.timeout_cleanup else 'disabled') )
elif optz.status_notify:
self.display( 'notification-thing:'
' unrecognized parameter', f'Key: {k!r}, value: {v!r}' )
return
self.PropertiesChanged(iface, {k: v}, [])
@dbus.service.signal(dbus_props, 'sa{sv}as')
def PropertiesChanged(self, iface, props_changed, props_invalidated):
log.debug( 'PropertiesChanged signal: %s',
[iface, props_changed, props_invalidated] )
@dbus.service.method(dbus_iface, '', '')
def Flush(self):
log.debug('Manual flush of the notification buffer')
self._activity_event()
return self.flush(force=True)
@dbus.service.method(dbus_iface, '', '')
def List(self):
log.debug('NotificationList call')
self._activity_event()
return list(self._note_windows.keys())
@dbus.service.method(dbus_iface, '', 'u')
def Redisplay(self):
log.debug('Redisplay call')
self._activity_event()
if not self._note_history: return 0
note = self._note_history.pop()
return self.display(note, redisplay=True)
@dbus.service.method(dbus_iface, 'du', '')
def Cleanup(self, timeout, max_count):
log.debug( 'NotificationCleanup call'
' (timeout=%.1fs, max_count=%s)', timeout, max_count )
self._activity_event()
if max_count <= 0: max_count = None
ts_min = time.monotonic() - timeout
for nid, note in sorted(self._note_windows.items(), key=lambda t: t[1].created):
if note.created > ts_min: break
self.close(nid, reason=close_reasons.closed)
if max_count is not None:
max_count -= 1
if max_count <= 0: break
def _notify_pubsub(self, _fd, _ev):
try:
while True:
msg = self.pubsub.recv()
if msg is None: break
self._activity_event()
note = msg.note
prefix, ts_diff = msg.hostname, time.time() - msg.ts
if ts_diff > 15 * 60: # older than 15min
prefix = f'{prefix}[{ts_diff_format(ts_diff)}]'
note.summary = f'{prefix} // {note.summary}'
note.hints['x-nt-from-remote'] = msg.hostname
self.filter_display(note)
except: log.exception('Unhandled error with remote notification')
finally: return True # for glib to keep watcher
@dbus.service.method(dbus_iface, 'susssasa{sv}i', 'u')
def Notify(self, app_name, nid, icon, summary, body, actions, hints, timeout):
self._activity_event()
try:
note = core.Notification.from_dbus(
app_name, nid, icon, summary, body, actions, hints, timeout )
if self.pubsub:
try: self._note_plaintext(note) # make sure plain version is cached
except Exception as err:
log.info('Failed to attach plain version to net message: %s', err)
self.pubsub.send(note)
if nid: self.close(nid, reason=close_reasons.closed)
return self.filter_display(note)
except Exception:
log.exception('Unhandled error')
return 0
@dbus.service.method(dbus_iface, 'u', '')
def CloseNotification(self, nid):
log.debug('CloseNotification call (id: %s)', nid)
self._activity_event()
self.close(nid, reason=close_reasons.closed)
_filter_ts_chk = 0
_filter_callback = None, 0
def _notification_check(self, summary, body, note=None):
(cb, mtime), ts = self._filter_callback, time.monotonic()
if self._filter_ts_chk < ts - poll_interval:
self._filter_ts_chk = ts
try: ts = int(os.stat(optz.filter_file).st_mtime)
except (OSError, IOError): return True
if ts > mtime:
mtime = ts
try: cb = core.get_filter(optz.filter_file, optz.filter_sound)
except:
ex, self._filter_callback = traceback.format_exc(), (None, 0)
log.debug( 'Failed to load'
' notification filters (from %s):\n%s', optz.filter_file, ex )
if optz.status_notify:
self.display('notification-thing: failed to load notification filters', ex)
return True
else:
log.debug('(Re)Loaded notification filters')
self._filter_callback = cb, mtime
if cb is None: return True # no filtering defined
elif not callable(cb): return bool(cb)
try: return cb(summary, body, note)
except:
ex = traceback.format_exc()
log.debug('Failed to run notification filters:\n%s', ex)
if optz.status_notify:
self.display('notification-thing: notification filters failed', ex)
return True
def _note_plaintext(self, note):
note_plain = note.get('plain')
if note_plain: summary, body = note_plain
else:
summary, body = self._renderer.get_note_text(note)
note.plain = summary, body
return summary, body
def _fullscreen_check(self, jitter=5):
screen = Gdk.Screen.get_default()
win = screen.get_active_window() # deprecated, but still useful
if not win: return False
win_state = win.get_state()
w, h = win.get_width(), win.get_height()
# get_geometry fails with "BadDrawable" from X if the window is closing,
# and x/y parameters there are not absolute and useful anyway.
# x, y, w, h = win.get_geometry()
# screen.get_width() / screen.get_height() are deprecated
return (win_state & win_state.FULLSCREEN)\
or (w >= screen.get_width() - jitter and h >= screen.get_height() - jitter)
def filter_display(self, note):
'Main method which all notifications are passed to for processing/display.'
note_summary, note_body = self._note_plaintext(note)
filter_pass = self._notification_check(note_summary, note_body, note)
try: urgency = int(note.hints['urgency'])
except (KeyError, ValueError): urgency = None
if self.logger and (filter_pass or optz.log_filtered):
try:
self.logger.write( note_summary, note_body,
urgency=urgency, ts=time.time() - (time.monotonic() - note.created) )
except:
ex = traceback.format_exc()
log.debug('Notification logger failed:\n%s', ex)
if optz.status_notify: self.display('notification-thing: notification logger failed', ex)
if not filter_pass:
log.debug('Dropped notification due to negative filtering result: %r', note_summary)
return 0
if optz.urgency_check and urgency == core.urgency_levels.critical:
self._note_limit.consume()
log.debug('Urgent message immediate passthru, tokens left: %s', self._note_limit.tokens)
return self.display(note)
plug = self.plugged or (optz.fs_check and self._fullscreen_check())
if plug or not self._note_limit.consume(): # Delay notification
to = self._note_limit.get_eta() if not plug else poll_interval
self._note_buffer.append(note)
log.debug( 'Queueing notification. Reason: %s. Flush attempt in %ss',
'plug or fullscreen window detected' if plug else 'notification rate limit', to )
self.flush(timeout=to)
return 0
if self._note_buffer:
self._note_buffer.append(note)
log.debug('Token-flush of notification buffer')
self.flush()
return 0
else:
log.debug('Token-pass, %s token(s) left', self._note_limit.tokens)
return self.display(note)
_flush_timer = _flush_id = None
def flush(self, force=False, timeout=None):
if self._flush_timer:
GLib.source_remove(self._flush_timer)
self._flush_timer = None
if timeout:
log.debug('Scheduled notification buffer flush in %ss', timeout)
self._flush_timer = GLib.timeout_add(int(timeout * 1000), self.flush)
return
if not self._note_buffer:
log.debug('Flush event with empty notification buffer')
return
log.debug(
'Flushing notification buffer (%s msgs, %s dropped)',
len(self._note_buffer), self._note_buffer.dropped )
self._note_limit.consume(force=True)
if not force:
if optz.fs_check and (self.plugged or self._fullscreen_check()):
log.debug( '%s detected, delaying buffer flush by %ss',
('Fullscreen window' if not self.plugged else 'Plug'), poll_interval )
self.flush(timeout=poll_interval)
return
if self._note_buffer:
# Decided not to use replace_id here - several feeds are okay
self._flush_id = self.display( self._note_buffer[0]\
if len(self._note_buffer) == 1\
else core.Notification.system_message(
'Feed' if not self._note_buffer.dropped
else f'Feed ({self._note_buffer.dropped} dropped)',
'\n\n'.join(it.starmap( '--- {}\n {}'.format,
map(op.itemgetter('summary', 'body'), self._note_buffer) )),
app_name='notification-feed', icon=optz.feed_icon ) )
self._note_buffer.flush()
log.debug('Notification buffer flushed')
def display(self, note_or_summary, body='', redisplay=False):
if isinstance(note_or_summary, core.Notification):
if body: raise TypeError('Notification object or summary/body should be passed, not both.')
note = note_or_summary
else:
note = core.Notification.system_message(note_or_summary, body)
if not redisplay:
clone = note.clone()
clone.display_time = time.monotonic()
self._note_history.append(clone)
else:
ts = getattr(note, 'display_time', None)
if ts: note.body += ( '\n\n[from '
f'{ts_diff_format(time.monotonic() - ts, add_ago=True)}]' )
if note.replaces_id in self._note_windows:
self.close(note.replaces_id, close_reasons.closed)
note.id = self._note_windows[note.replaces_id]
else: note.id = next(self._note_id_pool)
nid = note.id
self._renderer.display( note,
cb_hover=ft.partial(self.close, delay=True),
cb_leave=ft.partial(self.close, delay=False),
cb_dismiss=ft.partial(self.close, reason=close_reasons.dismissed) )
self._note_windows[nid] = note
if self.timeout_cleanup and note.timeout > 0:
note.timer_created, note.timer_left = time.monotonic(), note.timeout / 1000.0
note.timer_id = GLib.timeout_add(
note.timeout, self.close, nid, close_reasons.expired )
log.debug(
'Created notification (id: %s, timeout: %s (ms))',
nid, self.timeout_cleanup and note.timeout )
return nid
def close(self, nid=None, reason=close_reasons.undefined, delay=None):
if nid:
note = self._note_windows.get(nid, None)
if note:
if getattr(note, 'timer_id', None):
GLib.source_remove(note.timer_id)
if delay is None: del self._note_windows[nid]
elif hasattr(note, 'timer_id'): # these get sent very often
if delay:
if note.timer_id:
note.timer_id, note.timer_left = None,\
note.timer_left - (time.monotonic() - note.timer_created)
else:
note.timer_created = time.monotonic()
note.timer_id = GLib.timeout_add(
int(max(note.timer_left, 1) * 1000),
self.close, nid, close_reasons.expired )
return
if delay is None: # try it, even if there's no note object
log.debug(
'Closing notification(s) (id: %s, reason: %s)',
nid, close_reasons.by_id(reason) )
try: self._renderer.close(nid)
except self._renderer.NoWindowError: pass # no such window
else: self.NotificationClosed(nid, reason)
else: # close all of them
for nid in list(self._note_windows.keys()): self.close(nid, reason)
def main(argv=None):
global optz, log
import argparse
parser = argparse.ArgumentParser(description='Desktop notification server.')
parser.add_argument('--conf', metavar='path',
help='Read option values from specified YAML configuration file.'
' Keys in subsectons (like "tbf.size") will be joined with'
' parent section name with dash (e.g. --tbf-size).'
' Any values specified on command line will override corresponding ones from file.'
' See also notification_thing.example.yaml file.')
parser.add_argument('--conf-missing-ok', action='store_true',
help='Safely ignore missing or inaccessible configuration files.')
parser.add_argument('--debug', action='store_true', help='Enable debug logging to stderr.')
group = parser.add_argument_group('Daemon operation settings and notification defaults')
group.add_argument('-c', '--activity-timeout',
type=float, default=float(optz['activity_timeout']), metavar='seconds',
help='No-activity (dbus calls) timeout before closing the daemon instance'
' (less or equal zero - infinite, default: %(default)ss)')
group.add_argument('-t', '--popup-timeout',
type=float, default=float(optz['popup_timeout']*1000), metavar='seconds',
help='Default timeout for notification popups removal (default: %(default)sms)')
group.add_argument('-q', '--queue-len',
type=int, default=optz['queue_len'], metavar='n',
help='How many messages should be queued on tbf overflow (default: %(default)s)')
group.add_argument('-s', '--history-len',
type=int, default=optz['history_len'], metavar='n',
help='How many last *displayed* messages to'
' remember to display again on demand (default: %(default)s)')
group = parser.add_argument_group('Test/startup messages and sounds')
group.add_argument('--test-message', action='store_true',
help='Issue test notification right after start.')
group.add_argument('--test-sound', metavar='path',
help='Play specified sound on startup,'
' if sound support is available and enabled for filtering rules.')
group = parser.add_argument_group('Basic notification processing options')
group.add_argument('-f', '--no-fs-check',
action='store_false', dest='fs_check', default=True,
help='Dont queue messages if active window is fullscreen')
group.add_argument('-u', '--no-urgency-check',
action='store_false', dest='urgency_check', default=True,
help='Queue messages even if urgency is critical')
group.add_argument('--no-status-notify',
action='store_false', dest='status_notify', default=True,
help='Do not send notification on changes in daemon settings.')
group = parser.add_argument_group('Scheme-based notification filtering')
group.add_argument('--filter-file', default='~/.notification_filter', metavar='path',
help='Read simple scheme rules for filtering notifications from file (default: %(default)s).')
group.add_argument('--filter-test', nargs=2, metavar=('summary', 'body'),
help='Do not start daemon, just test given summary'
' and body against filter-file and print the result back to terminal.')
group.add_argument('--no-filter-sound',
action='store_false', dest='filter_sound', default=True,
help='Make sound calls in --filters-file scheme interpreter a no-op.'
' Only makes sense if sound calls in filters are actually used'
' and libcanberra is available, otherwise there wont be any sounds anyway.')
group = parser.add_argument_group('Display layout (e.g. position, direction, etc) options')
group.add_argument('--layout-anchor',
choices=core.layout_anchor, default='top_left',
help='Screen corner notifications gravitate to (default: %(default)s).')
group.add_argument('--layout-direction',
choices=core.layout_direction, default='vertical',
help='Direction for notification stack growth'
' from --layout-anchor corner (default: %(default)s).')
group.add_argument('--layout-margin', default=3, type=int, metavar='px',
help='Margin between notifications, screen edges, and some misc stuff (default: %(default)spx).')
group = parser.add_argument_group('Icon scaling options')
group.add_argument('--icon-width', '--img-w', type=int, metavar='px',
help='Scale all icons (preserving aspect ratio) to specified width.'
' In --icon-height is also specified, icons will be "fitted" into a WxH box, never larger.')
group.add_argument('--icon-height', '--img-h', type=int, metavar='px',
help='Scale all icons (preserving aspect ratio) to specified height.'
' In --icon-width is also specified, icons will be "fitted" into a WxH box, never larger.')
group.add_argument('--icon-size-max', '--img-max', metavar='WxH',
help='Scale icons larger than specified size'
' in any dimension down to it, preserving aspect ratio.'
' Value must be in "[w]x[h]" format, with w/h'
' being integers for size in pixels, either one or both of them can be specified.'
' Differs from --icon-width/--icon-height in'
' that it only scales-down larger ones, not all icons.')
group.add_argument('--icon-size-min', '--img-min', metavar='WxH',
help='Same as --icon-size-max, but scales-up smaller images to a specified min size.')
group = parser.add_argument_group('Text pango markup options')
group.add_argument('--markup-disable', action='store_true',
help='Enable pango markup (tags, somewhat similar to html)'
' processing in all message summary/body parts by default.'
' These will either be rendered by pango'
' or stripped/shown-as-is (see --markup-strip-on-err option), if invalid.'
' "x-nt-markup" bool hint can be used to'
' enable/disable markup on per-message basis, regardless of this option value.'
' See "Markup" section of the README for more details.')
group.add_argument('--markup-strip-on-err', action='store_true',
help='Strip markup tags if pango fails to parse them'
' (when parsing markup is enabled) instead of rendering message with them as text.')
group.add_argument('--markup-warn-on-err', action='store_true',
help='Issue loggin warning if passed markup tags cannot be parsed (when it is enabled).')
group = parser.add_argument_group('Token-bucket message rate-limiting options')
group.add_argument('--tbf-size',
type=int, default=optz['tbf_size'], metavar='n',
help='Token-bucket message-flow filter (tbf) bucket size (default: %(default)s)')
group.add_argument('--tbf-tick',
type=float, default=optz['tbf_tick'], metavar='seconds',
help='tbf update interval (new token),'
' so token_inflow = token / tbf_tick (default: %(default)ss)')
group.add_argument('--tbf-max-delay',
type=float, default=optz['tbf_max_delay'], metavar='seconds',
help='Maxmum amount of seconds, between message queue flush (default: %(default)ss)')
group.add_argument('--tbf-inc',
type=float, default=optz['tbf_inc'], metavar='value',
help='tbf_tick multiplier on consequent tbf overflow (default: %(default)s)')
group.add_argument('--tbf-dec',
type=float, default=optz['tbf_dec'], metavar='value',
help='tbf_tick divider on successful grab from non-empty bucket,'
' wont lower multiplier below 1 (default: %(default)s)')
group.add_argument('--feed-icon',
type=float, default=optz['feed_icon'], metavar='icon',
help='Icon name/path to use for aggregated ("feed") notification.'
' Can be full path to icon, local "file://..." url,'
' or name from xdg theme (e.g. "face-smile").')
group = parser.add_argument_group('Network pub/sub options')
group.add_argument('--net-pub-bind',
action='append', metavar='ip:port',
help='Publish messages over network on a specified socket endpoint.'
' Can be either ip:port (assumed to be tcp socket,'
' e.g. 1.2.3.4:5678) or full zmq url (e.g. tcp://1.2.3.4:5678).'
' Can be specified multiple times.')
group.add_argument('--net-pub-connect',
action='append', metavar='ip:port',
help='Send published messages to specified subscriber socket (see --net-sub-bind).'
' Same format as for --net-pub-bind. Can be specified multiple times.')
group.add_argument('--net-sub-bind',
action='append', metavar='ip:port',
help='Create subscriber socket that'
' publishers can connect and send messages to (see --net-pub-connect).'
' Same format as for --net-pub-bind. Can be specified multiple times.')
group.add_argument('--net-sub-connect',
action='append', metavar='ip:port',
help='Receive published messages from a specified pub socket (see --net-pub-bind).'
' Same format as for --net-pub-bind. Can be specified multiple times.')
group.add_argument('--net-settings', metavar='yaml',
help='Optional yaml/json encoded settings'
' for PubSub class init (e.g. hostname, buffer, reconnect_max, etc).')
group = parser.add_argument_group('Message logging options')
group.add_argument('--log-file', metavar='file',
help='Path to a log file to record all messages to.'
' Initial ~ or ~user components are auto-expanded to home dirs.')
group.add_argument('--log-filtered', action='store_true',
help='Enable logging for messages that did not pass through filtering checks.')
group.add_argument('--log-rotate-files', type=int, metavar='file-count',
help='Enable automatic log file roatation with specified number of old files kept.')
group.add_argument('--log-rotate-size',
type=int, metavar='bytes', default='1048576',
help='Rotate files after they get larger'
' than specified number of bytes. Default: 1 MiB.')
args = argv or sys.argv[1:]
optz = parser.parse_args(args)
if optz.conf_missing_ok and not os.access(optz.conf, os.R_OK): optz.conf = None
if optz.conf:
import yaml
for k, v in flatten_dict(yaml.safe_load(open(optz.conf)) or dict()):
if v is None: continue
k = '_'.join(k).replace('-', '_')
if not hasattr(optz, k):
parser.error(f'Unrecognized option in config file ({optz.conf}): {k}')
setattr(optz, k, v)
optz = parser.parse_args(args, optz) # re-parse to override cli-specified values
for k in 'layout_anchor', 'layout_direction':
setattr(optz, k, getattr(core, k)[getattr(optz, k)])
import logging
logging.basicConfig(level=logging.DEBUG if optz.debug else logging.WARNING)
log = logging.getLogger('daemon')
optz.filter_file = os.path.expanduser(optz.filter_file)
core.Notification.default_timeout = optz.popup_timeout
if optz.filter_sound:
optz.filter_sound = core.get_sound_env(
force_sync=optz.filter_test, trap_errors=not (optz.filter_test or optz.debug) )
if optz.filter_test:
func = core.get_filter(optz.filter_file, optz.filter_sound)
filtering_result = func(*optz.filter_test)
msg_repr = 'Message - summary: {!r}, body: {!r}'.format(*optz.filter_test)
print('{}\nFiltering result: {} ({})'.format( msg_repr,
filtering_result, 'will pass' if filtering_result else "won't pass" ))
return
optz.icon_scale = dict()
if optz.icon_width or optz.icon_height:
optz.icon_scale['fixed'] = optz.icon_width, optz.icon_height
else:
for k in 'min', 'max':
v = getattr(optz, f'icon_size_{k}')
if not v: continue
try:
if 'x' not in v: raise ValueError
if v.endswith('x'): v += '0'
if v.startswith('x'): v = '0' + v
w, h = map(int, v.split('x'))
except:
parser.error( f'Invalid --icon-size-{k}'
f' value (must be in "[w]x[h]" format): {v!r}' )
optz.icon_scale[k] = w, h
DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()
pubsub = None
if optz.net_pub_bind or optz.net_pub_connect\
or optz.net_sub_bind or optz.net_sub_connect:
if optz.net_settings and not isinstance(optz.net_settings, cs.abc.Mapping):
import yaml
optz.net_settings = yaml.safe_load(optz.net_settings)
pubsub = PubSub(**(optz.net_settings or dict()))
for addrs, call in [
(optz.net_pub_bind, pubsub.bind_pub),
(optz.net_sub_bind, pubsub.bind_sub),
(optz.net_pub_connect, pubsub.connect),
(optz.net_sub_connect, pubsub.subscribe) ]:
if isinstance(addrs, str): addrs = [addrs]
for addr in set(addrs or set()):
log.debug('zmq-link: %s %s', call.__func__.__name__, addr)
call(addr)
logger = None
if optz.log_file:
logger = FileLogger( os.path.expanduser(optz.log_file),
files=optz.log_rotate_files, size_limit=optz.log_rotate_size )
try: daemon = NotificationDaemon(bus, pubsub, logger)
except core.StartupFailure as err:
log.error('Failed to initialize the daemon: %s', err)
return 1
loop = GLib.MainLoop()
log.debug('Starting gobject loop')
loop.run()
if __name__ == '__main__': sys.exit(main())