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
|
"""
Monitor.py
by Jason Conti
February 19, 2010
Monitors DBUS for org.freedesktop.Notifications.Notify messages, parses them,
and notifies listeners when they arrive.
"""
import dbus
import dbus.mainloop.glib
import glib
import gobject
import gtk
import time
class Message(gobject.GObject):
"""Parses a DBUS message in the Notify format specified at:
http://www.galago-project.org/specs/notification/0.9/index.html"""
def __init__(self, dbus_message, timestamp):
gobject.GObject.__init__(self)
self.timestamp = timestamp
args = dbus_message.get_args_list()
self.app_name = str(args[0])
self.replaces_id = args[1]
self.app_icon = str(args[2])
self.summary = str(args[3])
self.body = str(args[4])
self.actions = args[5]
self.hints = args[6]
self.expire_timeout = args[7]
def get_icon_pixbuf(self):
"""Creates and returns a new pixbuf for the icon included in this message,
or None if there is an exception."""
try:
return gtk.gdk.pixbuf_new_from_file_at_size(self.app_icon, 48, 48)
except:
return None
def formatted_timestamp(self):
"""Returned the timestamp in a different format."""
return time.strftime("%B %d, %Y at %I:%M:%S %p", self.timestamp)
class Monitor(gobject.GObject):
"""Monitors DBUS for org.freedesktop.Notifications.Notify messages, parses them,
and notifies listeners when they arrive."""
__gsignals__ = {
"message-received": (gobject.SIGNAL_RUN_LAST, gobject.TYPE_NONE, [Message])
}
def __init__(self):
gobject.GObject.__init__(self)
self._match_string = "type='method_call',interface='org.freedesktop.Notifications',member='Notify'"
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
self._bus = dbus.SessionBus()
self._bus.add_match_string(self._match_string)
self._bus.add_message_filter(self._message_filter)
def close(self):
self._bus.close()
def _message_filter(self, connection, dbus_message):
if dbus_message.get_member() == "Notify" and dbus_message.get_interface() == "org.freedesktop.Notifications":
glib.idle_add(self.emit, "message-received", Message(dbus_message, time.localtime()))
def test_receiver(w, m):
print "-"*50
print "timestamp", m.formatted_timestamp()
print "app_name", repr(m.app_name)
print "replaces_id", repr(m.replaces_id)
print "app_icon", repr(m.app_icon)
print "summary", repr(m.summary)
print "body", repr(m.body)
print "actions", repr(m.actions)
print "hints", repr(m.hints)
print "expire_timeout", repr(m.expire_timeout)
print "-"*50
def main():
x = Monitor()
x.connect("message-received", test_receiver)
gtk.main()
if gtk.pygtk_version < (2, 8, 0):
gobject.type_register(Message)
gobject.type_register(Monitor)
if __name__ == '__main__':
main()
|