« back to all changes in this revision
Viewing changes to src/dfeet/bus_watch.py
- browse files at revision 17
- compare with another revision
- download diff
- download tarball
- view history from revision 17
- Committer: Package Import Robot
- Author(s): Dmitry Shachnev
- Date: 2014-03-10 18:38:06 UTC
- mfrom: (1.3.8)
- Revision ID: package-import@ubuntu.com-20140310183806-gh60dro8ge09wcld
Tags: 0.3.8+really0.3.6-0ubuntu1
Revert to the previous version, 0.3.8 has some unwanted UI changes.
- files added:
- .pc/02-no-gettext.patch
- files removed:
- compile
- files modified:
- .pc/03-disable-dbus-tests.patch/src/tests/tests.py
added
removed
src/dfeet/bus_watch.py
5
5
6
6
from dfeet.uiloader import UILoader
7
7
from dfeet.introspection import AddressInfo
8
from dfeet.wnck_utils import IconTable
9
10
11
class BusNameBox(Gtk.VBox):
12
"""class to represent a BusName (eg 'org.freedesktop.NetworkManager')"""
13
def __init__(self, bus_name, unique_name):
14
super(BusNameBox, self).__init__(spacing=5, expand=True)
15
self.__bus_name = bus_name
16
self.__unique_name = unique_name
17
self.__process_id = 0
18
self.__command_line = ''
19
self.__activatable = False
20
self.__icon_table = IconTable.get_instance()
21
self.__icon_image = Gtk.Image.new_from_pixbuf(self.__icon_table.default_icon)
22
23
self.__hbox = Gtk.HBox(spacing=5, halign=Gtk.Align.START)
24
self.pack_start(self.__hbox, True, True, 0)
25
#icon
26
self.__hbox.pack_start(self.__icon_image, True, True, 0)
27
#other information
28
self.__vbox_right = Gtk.VBox(spacing=5, expand=True)
29
self.__hbox.pack_start(self.__vbox_right, True, True, 0)
30
31
#first element
32
self.__label_bus_name = Gtk.Label()
33
self.__label_bus_name.set_halign(Gtk.Align.START)
34
self.__vbox_right.pack_start(self.__label_bus_name, True, True, 0)
35
#second element
36
self.__label_info = Gtk.Label()
37
self.__label_info.set_halign(Gtk.Align.START)
38
self.__vbox_right.pack_start(self.__label_info, True, True, 0)
39
#separator for the boxes
40
self.pack_end(Gtk.Separator(orientation=Gtk.Orientation.HORIZONTAL), True, True, 0)
41
#update widget information
42
self.__update_widget()
43
self.show_all()
44
45
def __update_widget(self):
46
"""update the widget with the available information"""
47
if self.__process_id > 0:
48
self.__label_bus_name.set_markup("<b>{0}</b>".format(self.__bus_name))
49
else:
50
self.__label_bus_name.set_markup("<b><i>{0}</i></b>".format(self.__bus_name))
51
#update the label info
52
label_info_str = "<small>"
53
if self.__activatable:
54
label_info_str += "activatable: yes"
55
else:
56
label_info_str += "activatable: no"
57
if self.__process_id:
58
label_info_str += ", pid: {0}".format(self.__process_id)
59
#get the icon (if available)
60
if self.__process_id in self.__icon_table.app_map.keys():
61
self.__icon_image.set_from_pixbuf(self.__icon_table.app_map[self.__process_id])
62
else:
63
self.__icon_image.set_from_pixbuf(self.__icon_table.default_icon)
64
if self.__command_line:
65
label_info_str += ", cmd: {0}".format(self.__command_line)
66
label_info_str += "</small>"
67
self.__label_info.set_markup(label_info_str)
8
9
10
class DBusBusName(GObject.GObject):
11
"""class to represent a name on the bus"""
12
def __init__(self, bus_name_unique):
13
super(DBusBusName, self).__init__()
14
self.__bus_name_unique = bus_name_unique
15
self.__pid = 0
16
self.__cmdline = ''
68
17
69
18
def __repr__(self):
70
return "%s (pid: %s)" % (self.__bus_name, self.__process_id)
19
return "%s (pid: %s)" % (self.bus_name_unique, self.pid)
71
20
72
def __update_command_line(self):
73
"""get the command line of process-id is available"""
74
if self.__process_id > 0:
75
procpath = '/proc/' + str(self.__process_id) + '/cmdline'
21
def __update_cmdline(self):
22
if self.pid > 0:
23
procpath = '/proc/' + str(self.pid) + '/cmdline'
76
24
with open(procpath, 'r') as f:
77
self.__command_line = " ".join(f.readline().split('\0'))
25
self.__cmdline = " ".join(f.readline().split('\0'))
78
26
else:
79
self.__command_line = ''
80
81
@property
82
def bus_name(self):
83
return self.__bus_name
84
85
@property
86
def unique_name(self):
87
return self.__unique_name
88
89
@property
90
def activatable(self):
91
return self.__activatable
92
93
@activatable.setter
94
def activatable(self, act_new):
95
self.__activatable = act_new
96
97
#update the shown widget
98
self.__update_widget()
99
100
@property
101
def process_id(self):
102
return self.__process_id
103
104
@process_id.setter
105
def process_id(self, process_id_new):
106
self.__process_id = process_id_new
27
self.__cmdline = ''
28
29
@property
30
def bus_name_unique(self):
31
return self.__bus_name_unique
32
33
@property
34
def pid(self):
35
return self.__pid
36
37
@pid.setter
38
def pid(self, pid_new):
39
self.__pid = pid_new
107
40
try:
108
self.__update_command_line()
41
self.__update_cmdline()
109
42
except:
110
self.__command_line = ''
111
#update the shown widget
112
self.__update_widget()
43
self.__cmdline = ''
44
45
@property
46
def cmdline(self):
47
return self.__cmdline
113
48
114
49
115
50
class BusWatch(object):
116
51
"""watch for a given bus"""
117
def __init__(self, data_dir, bus_address):
118
self.__data_dir = data_dir
119
self.__bus_address = bus_address
52
def __init__(self, data_dir, address):
53
self.data_dir = data_dir
54
self.address = address
120
55
#setup UI
121
ui = UILoader(self.__data_dir, UILoader.UI_BUS)
122
self.__box_bus = ui.get_root_widget()
123
self.__scrolledwindow_listbox = ui.get_widget("scrolledwindow_listbox")
124
self.__bus_name_filter = ui.get_widget('entry_filter')
125
#create a listbox for all the busnames
126
self.__listbox = Gtk.ListBox(hexpand=True, vexpand=True, expand=True)
127
self.__listbox.set_sort_func(self.__listbox_sort_by_name, None)
128
self.__listbox.set_filter_func(self.__listbox_filter_by_name, None)
129
self.__scrolledwindow_listbox.add(self.__listbox)
130
self.__scrolledwindow_listbox.show_all()
131
#setup the bus connection
132
if self.__bus_address == Gio.BusType.SYSTEM or self.__bus_address == Gio.BusType.SESSION:
133
#TODO: do this async
134
self.connection = Gio.bus_get_sync(self.__bus_address, None)
135
elif Gio.dbus_is_supported_address(self.__bus_address):
136
#TODO: do this async
56
ui = UILoader(self.data_dir, UILoader.UI_BUSWATCH)
57
self.paned_buswatch = ui.get_root_widget()
58
self.liststore_model = ui.get_widget('liststore_buswatch')
59
self.treemodelfilter_buswatch = ui.get_widget('treemodelfilter_buswatch')
60
self.treemodelfilter_buswatch.set_visible_func(self.__treemodelfilter_buswatch_cb)
61
self.treemodelsort_buswatch = ui.get_widget("treemodelsort_buswatch")
62
self.treemodelsort_buswatch.set_sort_func(2, self.__sort_on_name)
63
self.treemodelsort_buswatch.set_sort_column_id(2, Gtk.SortType.ASCENDING)
64
self.treeview = ui.get_widget('treeview_buswatch')
65
self.entry_filter = ui.get_widget('entry_filter')
66
self.grid_bus_name_selected_info = ui.get_widget('grid_bus_name_info')
67
self.label_bus_name_selected_name = ui.get_widget('label_bus_name_selected_name')
68
self.label_bus_name_selected_pid = ui.get_widget('label_bus_name_selected_pid')
69
self.label_bus_name_selected_cmdline = ui.get_widget('label_bus_name_selected_cmdline')
70
self.addr_info = None # hold the currently selected AddressInfo object
71
72
self.treeview.connect('cursor-changed',
73
self.__tree_view_cursor_changed_cb)
74
self.entry_filter.connect("changed",
75
self.__entry_filter_changed_cb)
76
77
#setup the conection
78
if self.address == Gio.BusType.SYSTEM or self.address == Gio.BusType.SESSION:
79
self.connection = Gio.bus_get_sync(self.address, None)
80
elif Gio.dbus_is_supported_address(self.address):
137
81
self.connection = Gio.DBusConnection.new_for_address_sync(
138
self.__bus_address,
82
self.address,
139
83
Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT |
140
84
Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION,
141
85
None, None)
142
else:
143
raise ValueError("Invalid bus address '{0}'".format(self.__bus_address))
144
86
145
87
#setup signals
146
88
self.connection.signal_subscribe(None, "org.freedesktop.DBus", "NameOwnerChanged",
147
89
None, None, 0, self.__name_owner_changed_cb, None)
148
90
149
#refilter if someone wants to filter the busbox list
150
self.__bus_name_filter.connect("changed",
151
self.__bus_name_filter_changed_cb)
152
153
#change bus detail tree if a different bus is selected
154
self.__listbox.connect("row-selected",
155
self.__listbox_row_selected_cb)
156
157
#TODO: do this async
158
91
self.bus_proxy = Gio.DBusProxy.new_sync(self.connection,
159
92
Gio.DBusProxyFlags.NONE,
160
93
None,
162
95
'/org/freedesktop/DBus',
163
96
'org.freedesktop.DBus', None)
164
97
165
#get a list with activatable names
166
self.bus_proxy.ListActivatableNames('()',
167
result_handler=self.__list_act_names_handler,
168
error_handler=self.__list_act_names_error_handler)
169
170
98
#list all names
171
99
self.bus_proxy.ListNames('()',
172
100
result_handler=self.__list_names_handler,
173
101
error_handler=self.__list_names_error_handler)
174
102
175
@property
176
def box_bus(self):
177
"""the main widget for the bus"""
178
return self.__box_bus
179
180
def __bus_name_filter_changed_cb(self, bus_name_filter):
181
"""someone typed something in the searchbox - refilter"""
182
self.__listbox.invalidate_filter()
183
184
def __listbox_row_selected_cb(self, listbox, listbox_row):
185
"""someone selected a different row of the listbox"""
186
childs = self.box_bus.get_children()
187
#never remove first element - that's the listbox with the busnames
188
if len(childs) > 1:
189
self.box_bus.remove(childs[-1])
190
191
try:
192
del(self.__addr_info)
193
except:
194
pass
195
196
#get the selected busname
197
if listbox_row:
198
row_childs = listbox_row.get_children()
199
bus_name_box = row_childs[0]
200
#add the introspection info to the left side
201
self.__addr_info = AddressInfo(self.__data_dir,
202
self.__bus_address,
203
bus_name_box.bus_name,
204
bus_name_box.unique_name,
205
connection_is_bus=True)
206
self.box_bus.pack_end(self.__addr_info.introspect_box, True, True, 0)
207
self.box_bus.show_all()
103
def __treemodelfilter_buswatch_cb(self, model, iter, user_data):
104
#return model.get_value(iter, 1) in data
105
bus_name_obj = model.get(iter, 0)[0]
106
filter_text = self.entry_filter.get_text()
107
return filter_text.lower() in bus_name_obj.bus_name_unique.lower()
108
109
def __entry_filter_changed_cb(self, entry_filter):
110
self.treemodelfilter_buswatch.refilter()
111
112
def __tree_view_cursor_changed_cb(self, treeview):
113
"""do something when a row is selected"""
114
selection = self.treeview.get_selection()
115
if selection:
116
model, iter_ = selection.get_selected()
117
if not iter_:
118
return
119
120
bus_name_obj = model.get(iter_, 0)[0]
121
#remove current child
122
c2 = self.paned_buswatch.get_child2()
123
if c2:
124
c2.destroy()
125
try:
126
del(self.addr_info)
127
except:
128
pass
129
130
#add Introspection to paned
131
self.addr_info = AddressInfo(
132
self.data_dir, self.address, bus_name_obj.bus_name_unique, connection_is_bus=True)
133
self.paned_buswatch.add2(self.addr_info.introspect_box)
134
135
#update info about selected bus name
136
self.label_bus_name_selected_name.set_text(bus_name_obj.bus_name_unique)
137
self.label_bus_name_selected_pid.set_text("%s" % bus_name_obj.pid)
138
self.label_bus_name_selected_cmdline.set_text(bus_name_obj.cmdline)
139
self.grid_bus_name_selected_info.set_visible(True)
140
141
def __liststore_model_add(self, bus_name_obj):
142
"""add a DBusBusName object to the liststore model"""
143
#update bus info stuff
144
self.bus_proxy.GetConnectionUnixProcessID(
145
'(s)', bus_name_obj.bus_name_unique,
146
result_handler=self.__get_unix_process_id_cb,
147
error_handler=self.__get_unix_process_id_error_cb,
148
user_data=bus_name_obj)
149
#add bus to liststore
150
return self.liststore_model.append(
151
[bus_name_obj, 0, "%s" % (bus_name_obj.bus_name_unique), bus_name_obj.cmdline])
152
153
def __liststore_model_remove(self, bus_name_obj):
154
"""remove a DBusBusName object to the liststore model"""
155
for n, obj in enumerate(self.liststore_model):
156
if obj[2] == bus_name_obj.bus_name_unique:
157
del(self.liststore_model[n])
158
break
159
160
def __liststore_model_get(self, bus_name_obj):
161
"""get a object from the liststore"""
162
for n, obj in enumerate(self.liststore_model):
163
if obj[2] == bus_name_obj.bus_name_unique:
164
return obj
165
raise Exception("bus name object '%s' not found in liststore" % (bus_name_obj))
208
166
209
167
def __name_owner_changed_cb(self, connection, sender_name,
210
168
object_path, interface_name, signal_name,
214
172
old_owner = parameters[1]
215
173
new_owner = parameters[2]
216
174
175
bus_name_obj = DBusBusName(bus_name)
176
217
177
if bus_name[0] == ':':
218
178
if not old_owner:
219
bus_name_box = BusNameBox(bus_name, new_owner)
220
self.__listbox_add_bus_name(bus_name_box)
179
self.__liststore_model_add(bus_name_obj)
221
180
else:
222
self.__listbox_remove_bus_name(bus_name)
181
self.__liststore_model_remove(bus_name_obj)
223
182
else:
224
183
if new_owner:
225
bus_name_box = BusNameBox(bus_name, new_owner)
226
self.__listbox_add_bus_name(bus_name_box)
184
self.__liststore_model_add(bus_name_obj)
227
185
if old_owner:
228
self.__listbox_remove_bus_name(bus_name)
229
230
def __listbox_find_bus_name(self, bus_name):
231
"""find the given busname in the listbox or return None if not found"""
232
for listbox_child in self.__listbox.get_children():
233
if listbox_child.get_children()[0].bus_name == bus_name:
234
return listbox_child
235
#busname not found
236
return None
237
238
def __listbox_remove_bus_name(self, bus_name):
239
"""remove the given busname from the listbox"""
240
obj = self.__listbox_find_bus_name(bus_name)
241
if obj:
242
self.__listbox.remove(obj)
243
#if bus is activatable, add the bus name again
244
if bus_name in self.__activatable_names:
245
bnb = BusNameBox(bus_name, '')
246
self.__listbox_add_bus_name(bnb)
247
else:
248
print("can not remove busname '{0}'. busname not found".format(bus_name))
249
250
def __listbox_add_bus_name(self, bus_name_box):
251
"""add the given busnamebox to the listbox and update the info"""
252
#first check if busname is already listed
253
#ie an activatable (but inactive) busname
254
bn = self.__listbox_find_bus_name(bus_name_box.bus_name)
255
if bn:
256
#bus name is already in the list - use this
257
bus_name_box = bn.get_children()[0]
258
else:
259
#add busnamebox to the list
260
self.__listbox.add(bus_name_box)
261
262
#update bus info stuff
263
self.bus_proxy.GetConnectionUnixProcessID(
264
'(s)', bus_name_box.bus_name,
265
result_handler=self.__get_unix_process_id_cb,
266
error_handler=self.__get_unix_process_id_error_cb,
267
user_data=bus_name_box)
268
#check if bus name is dbus activatable
269
if bus_name_box.bus_name in self.__activatable_names:
270
bus_name_box.activatable = True
271
else:
272
bus_name_box.activatable = False
273
274
def __add_name(self, name, unique_name):
275
bus_name_box = BusNameBox(name, unique_name)
276
self.__listbox_add_bus_name(bus_name_box)
277
278
def __get_name_owner_cb(self, obj, unique_name, name):
279
self.__add_name(name, unique_name)
280
281
def __get_name_owner_error_cb(self, obj, error, name):
282
# no name owner, empty unique name
283
self.__add_name(name, '')
284
285
def __add_names(self, names):
186
self.__liststore_model_remove(bus_name_obj)
187
188
def __list_names_handler(self, obj, names, userdata):
286
189
for n in names:
287
#unique names are added right away
288
if n[0] == ':':
289
self.__add_name(n, n)
290
else:
291
self.bus_proxy.GetNameOwner('(s)', n,
292
result_handler=self.__get_name_owner_cb,
293
error_handler=self.__get_name_owner_error_cb,
294
user_data=n)
295
296
def __list_names_handler(self, obj, names, userdata):
297
self.__add_names(names)
190
bus_name_obj = DBusBusName(n)
191
self.__liststore_model_add(bus_name_obj)
298
192
299
193
def __list_names_error_handler(self, obj, error, userdata):
300
194
print("error getting bus names: %s" % str(error))
301
195
302
def __list_act_names_handler(self, obj, act_names, userdata):
303
#remember the activatable bus names
304
self.__activatable_names = act_names
305
#add all activatable bus names to the list
306
self.__add_names(act_names)
307
308
def __list_act_names_error_handler(self, obj, error, userdata):
309
self.__activatable_names = []
310
print("error getting activatable names: %s" % str(error))
311
312
def __get_unix_process_id_cb(self, obj, pid, bus_name_box):
313
bus_name_box.process_id = pid
314
315
def __get_unix_process_id_error_cb(self, obj, error, bus_name_box):
316
#print("error getting unix process id for %s: %s" % (
317
# bus_name_box.bus_name, str(error)))
318
bus_name_box.process_id = 0
319
320
def __listbox_filter_by_name(self, row, user_data):
321
bus_name_box_list = row.get_children()
322
return self.__bus_name_filter.get_text().lower() in bus_name_box_list[0].bus_name.lower()
323
324
def __listbox_sort_by_name(self, row1, row2, user_data):
325
"""sort function for listbox"""
326
child1 = row1.get_children()
327
child2 = row2.get_children()
328
un1 = child1[0].bus_name
329
un2 = child2[0].bus_name
196
def __get_unix_process_id_cb(self, obj, pid, bus_name_obj):
197
bus_name_obj.pid = pid
198
199
def __get_unix_process_id_error_cb(self, obj, error, bus_name_obj):
200
print(
201
"error getting unix process id for %s: %s" % (
202
bus_name_obj.bus_name_unique, str(error)))
203
bus_name_obj.pid = 0
204
205
def __sort_on_name(self, model, iter1, iter2, user_data):
206
un1 = model.get_value(iter1, 2)
207
un2 = model.get_value(iter2, 2)
330
208
331
209
# covert to integers if comparing two unique names
332
210
if un1[0] == ':' and un2[0] == ':':
Loggerhead is a web-based interface for Breezy
Version: 2.0.1