« back to all changes in this revision
Viewing changes to unity/Notification.py
- browse files at revision 175
- compare with another revision
- download diff
- download tarball
- view history from revision 175
- Committer: Jason Conti
- Date: 2011-03-26 21:20:00 UTC
- Revision ID: jason.conti@gmail.com-20110326212000-48rj1zased29tne4
Ported Notification.py to gobject introspection, everything except dbus. Going to attempt that next, but may revert if it doesn't work so well yet.
- files added:
- data
- indicator
- files removed:
- indicator-notify
added
removed
unity/Notification.py
1
"""
2
Notification.py
3
by Jason Conti
4
February 19, 2010
5
updated: March 26, 2011
6
7
Monitors DBUS for org.freedesktop.Notifications.Notify messages, parses them,
8
and notifies listeners when they arrive.
9
"""
10
11
import dbus
12
from gi.repository import GObject, Gtk
13
import logging
14
15
from dbus.mainloop.glib import DBusGMainLoop
16
17
import Icon
18
import Timestamp
19
20
logger = logging.getLogger("Notification")
21
22
class ImageDataException(Exception):
23
pass
24
25
class ImageData(object):
26
"""Parses the image_data hint from a DBUS message."""
27
def __init__(self, image_data):
28
if len(image_data) < 7:
29
raise ImageDataException("Invalid image_data: " + repr(image_data))
30
31
self.width = int(image_data[0])
32
self.height = int(image_data[1])
33
self.rowstride = int(image_data[2])
34
self.has_alpha = bool(image_data[3])
35
self.bits_per_sample = int(image_data[4])
36
self.channels = int(image_data[5])
37
self.data = self.dbus_array_to_str(image_data[6])
38
39
def dbus_array_to_str(self, array):
40
return "".join(map(chr, array))
41
42
def get_pixbuf(self, size):
43
"""Creates a pixbuf from the image data and scale it to the appropriate size."""
44
pixbuf = GdkPixbuf.Pixbuf.new_from_data(self.data, GdkPixbuf.Colorspace.RGB,
45
self.has_alpha, self.bits_per_sample, self.width, self.height,
46
self.rowstride)
47
48
return pixbuf.scale_simple(size, size, GdkPixbuf.InterpType.BILINEAR)
49
50
class MessageException(Exception):
51
pass
52
53
class Message(GObject.GObject):
54
"""Parses a DBUS message in the Notify format specified at:
55
http://www.galago-project.org/specs/notification/0.9/index.html"""
56
# Message urgency
57
LOW = 0
58
NORMAL = 1
59
CRITICAL = 2
60
X_CANONICAL_PRIVATE_SYNCHRONOUS = "x-canonical-private-synchronous"
61
def __init__(self, dbus_message = None, timestamp = None):
62
GObject.GObject.__init__(self)
63
64
self.timestamp = timestamp
65
66
args = dbus_message.get_args_list()
67
68
if len(args) != 8:
69
raise MessageException("Invalid message args_list: " + repr(args))
70
71
self.app_name = str(args[0])
72
self.replaces_id = args[1]
73
self.app_icon = str(args[2])
74
self.summary = str(args[3])
75
self.body = str(args[4])
76
self.actions = args[5]
77
self.hints = dict(args[6])
78
self.expire_timeout = args[7]
79
80
if "urgency" in self.hints:
81
urgency = self.hints["urgency"]
82
if urgency == 0:
83
self.urgency = Message.LOW
84
elif urgency == 2:
85
self.urgency = Message.CRITICAL
86
else:
87
self.urgency = Message.NORMAL
88
else:
89
self.urgency = Message.NORMAL
90
91
if "image_data" in self.hints:
92
self.image_data = ImageData(self.hints["image_data"])
93
else:
94
self.image_data = None
95
96
if self.image_data == None and "icon_data" in self.hints:
97
self.image_data = ImageData(self.hints["icon_data"])
98
99
self.log_message()
100
101
def get_icon(self, size = 48):
102
"""Loads the icon into a pixbuf. Adapted from the load_icon code in
103
bubble.c of notify-osd."""
104
icon_name = self.app_icon
105
106
# Try to load the pixbuf from a file
107
if icon_name.startswith("file://") or icon_name.startswith("/"):
108
icon = Icon.load_from_file(icon_name, size)
109
if icon != None:
110
return icon
111
112
# Try to load the pixbuf from the current icon theme
113
elif icon_name != "":
114
icon = Icon.load(icon_name, size)
115
if icon != None:
116
return icon
117
118
# Try to load the image data from the message
119
elif self.image_data != None:
120
return self.image_data.get_pixbuf(32)
121
122
return self.get_default_icon(size)
123
124
def get_default_icon(self, size = 48):
125
"""Attempts to load the default message icon, returns None on failure."""
126
if self.urgency == Message.LOW:
127
return Icon.load("notification-low", size)
128
elif self.urgency == Message.CRITICAL:
129
return Icon.load("notification-critical", size)
130
else:
131
return Icon.load("notification-normal", size)
132
133
def is_volume_notification(self):
134
"""Returns true if this is a volume message. The volume notifications
135
in Ubuntu are a special case, and mostly a blank message. Clutters up
136
the display and provides no useful information, so it is better to
137
discard them."""
138
if Message.X_CANONICAL_PRIVATE_SYNCHRONOUS in self.hints:
139
return str(self.hints[Message.X_CANONICAL_PRIVATE_SYNCHRONOUS]) == "volume"
140
141
def log_message(self):
142
"""Write debug info about a message."""
143
result = [
144
"-" * 50,
145
"Message created at: " + Timestamp.locale_datetime(self.timestamp),
146
"app_name: " + repr(self.app_name),
147
"replaces_id: " + repr(self.replaces_id),
148
"app_icon: " + repr(self.app_icon),
149
"summary: " + repr(self.summary),
150
"body: " + repr(self.body),
151
"actions: " + repr(self.actions),
152
"expire_timeout: " + repr(self.expire_timeout),
153
"hints:"
154
]
155
156
# Log all the hints except image_data
157
for key in self.hints:
158
if key != "image_data":
159
result.append(" " + str(key) + ": " + repr(self.hints[key]))
160
161
# Log info about the image_data
162
if self.image_data != None:
163
result.append("image_data:")
164
result.append(" width: " + repr(self.image_data.width))
165
result.append(" height: " + repr(self.image_data.height))
166
result.append(" rowstride: " + repr(self.image_data.rowstride))
167
result.append(" has_alpha: " + repr(self.image_data.has_alpha))
168
result.append(" bits_per_sample: " + repr(self.image_data.bits_per_sample))
169
result.append(" channels: " + repr(self.image_data.channels))
170
171
result.append("-" * 50)
172
173
logger.debug("\n" + "\n".join(result))
174
175
class Notification(GObject.GObject):
176
"""Monitors DBUS for org.freedesktop.Notifications.Notify messages, parses them,
177
and notifies listeners when they arrive."""
178
__gsignals__ = {
179
"message-received": (GObject.SignalFlags.RUN_LAST, None, [Message])
180
}
181
def __init__(self):
182
GObject.GObject.__init__(self)
183
184
self._blacklist = None
185
self._match_string = "type='method_call',interface='org.freedesktop.Notifications',member='Notify'"
186
187
DBusGMainLoop(set_as_default=True)
188
self._bus = dbus.SessionBus()
189
self._bus.add_match_string(self._match_string)
190
self._bus.add_message_filter(self._message_filter)
191
192
def close(self):
193
"""Closes the connection to the session bus."""
194
self._bus.close()
195
196
def set_blacklist(self, blacklist):
197
"""Defines the set of app_names of messages to be discarded."""
198
self._blacklist = blacklist
199
200
def _message_filter(self, connection, dbus_message):
201
"""Triggers when messages are received from the session bus."""
202
if dbus_message.get_member() == "Notify" and dbus_message.get_interface() == "org.freedesktop.Notifications":
203
try:
204
message = Message(dbus_message, Timestamp.now())
205
except:
206
logger.exception("Failed to parse dbus message: " + repr(dbus_message.get_args_list()))
207
else:
208
# Discard unwanted messages
209
if message.is_volume_notification():
210
return
211
if self._blacklist and self._blacklist.get_bool(message.app_name, False):
212
return
213
logger.debug("Sending message-received")
214
self.emit("message-received", message)
215
216
GObject.type_register(Message)
217
GObject.type_register(Notification)
218
Loggerhead is a web-based interface for Breezy
Version: 2.0.1