6
Monitors DBUS for org.freedesktop.Notifications.Notify messages, parses them,
7
and notifies listeners when they arrive.
16
from dbus.mainloop.glib import DBusGMainLoop
17
from locale import nl_langinfo, T_FMT
21
from Icon import load_icon, load_icon_from_file
23
logger = logging.getLogger("Notification")
25
class ImageDataException(Exception):
28
class ImageData(object):
29
"""Parses the image_data hint from a DBUS message."""
30
def __init__(self, image_data):
31
if len(image_data) < 7:
32
raise ImageDataException("Invalid image_data: " + repr(image_data))
34
self.width = int(image_data[0])
35
self.height = int(image_data[1])
36
self.rowstride = int(image_data[2])
37
self.has_alpha = bool(image_data[3])
38
self.bits_per_sample = int(image_data[4])
39
self.channels = int(image_data[5])
40
self.data = self.dbus_array_to_str(image_data[6])
42
def dbus_array_to_str(self, array):
43
return "".join(map(chr, array))
45
def get_pixbuf(self, size):
46
"""Creates a pixbuf from the image data and scale it to the appropriate size."""
47
pixbuf = gtk.gdk.pixbuf_new_from_data(self.data, gtk.gdk.COLORSPACE_RGB,
48
self.has_alpha, self.bits_per_sample, self.width, self.height,
51
return pixbuf.scale_simple(size, size, gtk.gdk.INTERP_BILINEAR)
53
class MessageException(Exception):
56
class Message(gobject.GObject):
57
"""Parses a DBUS message in the Notify format specified at:
58
http://www.galago-project.org/specs/notification/0.9/index.html"""
63
X_CANONICAL_PRIVATE_SYNCHRONOUS = "x-canonical-private-synchronous"
64
def __init__(self, dbus_message = None, timestamp = None):
65
gobject.GObject.__init__(self)
67
self.timestamp = timestamp
69
args = dbus_message.get_args_list()
72
raise MessageException("Invalid message args_list: " + repr(args))
74
self.app_name = str(args[0])
75
self.replaces_id = args[1]
76
self.app_icon = str(args[2])
77
self.summary = str(args[3])
78
self.body = str(args[4])
79
self.actions = args[5]
80
self.hints = dict(args[6])
81
self.expire_timeout = args[7]
83
if "urgency" in self.hints:
84
urgency = self.hints["urgency"]
86
self.urgency = Message.LOW
88
self.urgency = Message.CRITICAL
90
self.urgency = Message.NORMAL
92
self.urgency = Message.NORMAL
94
if "image_data" in self.hints:
95
self.image_data = ImageData(self.hints["image_data"])
97
self.image_data = None
99
if self.image_data == None and "icon_data" in self.hints:
100
self.image_data = ImageData(self.hints["icon_data"])
104
def get_icon(self, size = 48):
105
"""Loads the icon into a pixbuf. Adapted from the load_icon code in
106
bubble.c of notify-osd."""
107
icon_name = self.app_icon
109
# Try to load the pixbuf from a file
110
if icon_name.startswith("file://") or icon_name.startswith("/"):
111
icon = load_icon_from_file(icon_name, size)
115
# Try to load the pixbuf from the current icon theme
116
elif icon_name != "":
117
icon = load_icon(icon_name, size)
121
# Try to load the image data from the message
122
elif self.image_data != None:
123
return self.image_data.get_pixbuf(32)
125
return self.get_default_icon(size)
127
def get_default_icon(self, size = 48):
128
"""Attempts to load the default message icon, returns None on failure."""
129
if self.urgency == Message.LOW:
130
return load_icon("notification-low", size)
131
elif self.urgency == Message.CRITICAL:
132
return load_icon("notification-critical", size)
134
return load_icon("notification-normal", size)
136
def is_volume_notification(self):
137
"""Returns true if this is a volume message. The volume notifications
138
in Ubuntu are a special case, and mostly a blank message. Clutters up
139
the display and provides no useful information, so it is better to
141
if Message.X_CANONICAL_PRIVATE_SYNCHRONOUS in self.hints:
142
return str(self.hints[Message.X_CANONICAL_PRIVATE_SYNCHRONOUS]) == "volume"
144
def log_message(self):
145
"""Write debug info about a message."""
148
"Message created at: " + Timestamp.locale_datetime(self.timestamp),
149
"app_name: " + repr(self.app_name),
150
"replaces_id: " + repr(self.replaces_id),
151
"app_icon: " + repr(self.app_icon),
152
"summary: " + repr(self.summary),
153
"body: " + repr(self.body),
154
"actions: " + repr(self.actions),
155
"expire_timeout: " + repr(self.expire_timeout),
159
# Log all the hints except image_data
160
for key in self.hints:
161
if key != "image_data":
162
result.append(" " + str(key) + ": " + repr(self.hints[key]))
164
# Log info about the image_data
165
if self.image_data != None:
166
result.append("image_data:")
167
result.append(" width: " + repr(self.image_data.width))
168
result.append(" height: " + repr(self.image_data.height))
169
result.append(" rowstride: " + repr(self.image_data.rowstride))
170
result.append(" has_alpha: " + repr(self.image_data.has_alpha))
171
result.append(" bits_per_sample: " + repr(self.image_data.bits_per_sample))
172
result.append(" channels: " + repr(self.image_data.channels))
174
result.append("-" * 50)
176
logger.debug("\n" + "\n".join(result))
178
class Notification(gobject.GObject):
179
"""Monitors DBUS for org.freedesktop.Notifications.Notify messages, parses them,
180
and notifies listeners when they arrive."""
182
"message-received": (gobject.SIGNAL_RUN_LAST, gobject.TYPE_NONE, [Message])
185
gobject.GObject.__init__(self)
187
self._blacklist = None
188
self._match_string = "type='method_call',interface='org.freedesktop.Notifications',member='Notify'"
190
DBusGMainLoop(set_as_default=True)
191
self._bus = dbus.SessionBus()
192
self._bus.add_match_string(self._match_string)
193
self._bus.add_message_filter(self._message_filter)
196
"""Closes the connection to the session bus."""
199
def set_blacklist(self, blacklist):
200
"""Defines the set of app_names of messages to be discarded."""
201
self._blacklist = blacklist
203
def _message_filter(self, connection, dbus_message):
204
"""Triggers when messages are received from the session bus."""
205
if dbus_message.get_member() == "Notify" and dbus_message.get_interface() == "org.freedesktop.Notifications":
207
message = Message(dbus_message, Timestamp.now())
209
logger.exception("Failed to parse dbus message: " + repr(dbus_message.get_args_list()))
211
# Discard unwanted messages
212
if message.is_volume_notification():
214
if self._blacklist and self._blacklist.get_bool(message.app_name, False):
216
glib.idle_add(self.emit, "message-received", message)
218
if gtk.pygtk_version < (2, 8, 0):
219
gobject.type_register(Message)
220
gobject.type_register(Notification)
222
if __name__ == '__main__':
added
removed
recent_notifications/Notification.py
