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
|
"""
Lens.py
April 20, 2011
"""
import logging
from gi.repository import Dee, Gio, GLib, GObject, Unity
from Notification import Notification
import Timestamp
BUS_NAME = "net.launchpad.RecentNotifications"
OBJECT_PATH = "/net/launchpad/RecentNotifications"
DBUS_NAME_FLAG_DO_NOT_QUEUE = 0x4
DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER = 1
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("Lens")
class Group:
RECENT_NOTIFICATIONS = 0
class LensException(Exception):
pass
class Lens(object):
def __init__(self):
self._connection = Gio.bus_get_sync(Gio.BusType.SESSION, None)
if not self.request_name(BUS_NAME):
raise LensException("Failed to acquire bus name: {0}".format(BUS_NAME))
self._messages = []
self._entry = Unity.PlaceEntryInfo.new(OBJECT_PATH + "/mainentry")
self._sections_model = Dee.SharedModel.new(BUS_NAME + ".SectionsModel")
self._sections_model.set_schema("s", "s")
self._entry.props.sections_model = self._sections_model
self._groups_model = Dee.SharedModel.new(BUS_NAME + ".GroupsModel")
self._groups_model.set_schema("s", "s", "s")
self._entry.props.entry_renderer_info.props.groups_model = self._groups_model
self._results_model = Dee.SharedModel.new(BUS_NAME + ".ResultsModel")
self._results_model.set_schema("s", "s", "u", "s", "s", "s")
self._entry.props.entry_renderer_info.props.results_model = self._results_model
self._sections_model.connect("notify::synchronized", self.on_sections_synchronized)
self._groups_model.connect("notify::synchronized", self.on_groups_synchronized)
self._results_model.connect("notify::synchronized", self.on_results_synchronized)
self._entry.connect("notify::active-search", self.on_search_changed)
self._entry.connect("notify::active-section", self.on_section_changed)
self._notify = Notification()
self._notify.connect("message-received", self.on_message_received)
self._controller = Unity.PlaceController.new(OBJECT_PATH)
self._controller.add_entry(self._entry)
self._controller.export()
def append_message_to_results(self, message):
if message.app_icon == "":
icon = "file:///home/jconti/Projects/bzr/recent-notifications/icons/notification-normal.svg"
else:
icon = message.app_icon
self._results_model.prepend("", icon, Group.RECENT_NOTIFICATIONS,
"text/html", message.summary, "{0}\n{1} from {2}".format(message.body,
Timestamp.locale_datetime(message.timestamp), message.app_name))
def on_message_received(self, monitor, message):
self._messages.append(message)
if self._results_model.get_property("synchronized"):
self.append_message_to_results(message)
self._results_model.flush_revision_queue()
def on_groups_synchronized(self, groups_model, *args):
groups_model.clear()
groups_model.append("UnityHorizontalTileRenderer",
"Recent Notifications",
"file:///home/jconti/Projects/bzr/recent-notifications/icons/notification-unread.svg")
def on_results_synchronized(self, results_model, *args):
results_model.clear()
for message in self._messages:
self.append_message_to_results(message)
results_model.flush_revision_queue()
def on_sections_synchronized(self, sections_model, *args):
sections_model.clear()
sections_model.append("All Applications",
Gio.ThemedIcon.new("document").to_string())
def on_search_changed(self, entry, *args):
entry.freeze_notify()
active_search = entry.get_property("active_search")
search_string = active_search.get_search_string()
logger.debug("Searching for: {0}".format(search_string))
active_search.finished()
GObject.idle_add(self.thaw, entry)
def on_section_changed(self, entry, section):
active_section = entry.get_property("active_section")
logger.debug("Changed section: {0}".format(active_section))
def request_name(self, name):
proxy = Gio.DBusProxy.new_sync(self._connection, 0, None,
"org.freedesktop.DBus", "/org/freedesktop/DBus",
"org.freedesktop.DBus", None)
result = proxy.call_sync("RequestName",
GLib.Variant("(su)", (BUS_NAME, DBUS_NAME_FLAG_DO_NOT_QUEUE)),
Gio.DBusCallFlags.NONE, -1, None)
value = result.unpack()
if len(value) < 1 or value[0] != DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER:
logger.debug("RequestName returned value: {0}".format(value))
return False
else:
return True
def thaw(self, entry):
entry.thaw_notify()
return False
def main():
lens = Lens()
loop = GObject.MainLoop()
loop.run()
if __name__ == "__main__":
main()
|