Below is the file 'interapplet.py' from this revision. You can also download the file.
#!/usr/bin/python # InterApplet # Copyright (C) 2005 Grahame Bowland # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA import os import sys import re import pygtk pygtk.require('2.0') import gtk import gnomeapplet import gobject import gamin import dbus import popen2 import fcntl import pipes state_directory = "/var/lib/interapplet/state" ifstate_file = '/etc/network/ifstate' interfaces_file = '/etc/network/interfaces' update_state_program = '/usr/lib/interapplet/set-state.sh' normal_logo = '/usr/share/interapplet/normal.png' active_logo = '/usr/share/interapplet/active.gif' show_log_xml = '''\ <popup name="button3"> <menuitem name="log_item" verb="log_verb_show" label="_Show log" /> <menuitem name="interapplet_about" verb="interapplet_about" _label="_About" pixtype="stock" pixname="gnome-stock-about"/> </popup>''' def set_nonblocking(fd): fl = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NDELAY) class InterfaceInfo: def __init__(self, file): self.dialups = set() self.interfaces = set() self.mappings = set() self.mappable = set() mapping = re.compile(r'^\s?mapping\s+([\w_:]+)') map = re.compile(r'^\s?map\s+([\w_]+)\s+([\w_]+)') auto = re.compile(r'^\s?auto[\s^]') iface = re.compile(r'^\s?iface\s+([\w_]+)\s+([\w_]+)\s+([\w_]+)') fd = open(file) for line in fd: m = mapping.match(line) if m: self.mappings.add(m.groups()[0]) m = iface.match(line) if m: name, family, method = m.groups() if method == "loopback": continue elif method == "ppp" or method == "wvdial": self.dialups.add(name) else: self.interfaces.add(name) class InterfaceState: def __init__(self, file): self.state = {} mapping = re.compile(r'^([\w_]+)=([\w_]+)$') for line in open(file): m = mapping.match(line) if m: self.state[m.groups()[0]] = m.groups()[1] class DeviceInfo: def __init__(self, change_cb): self.change_cb = change_cb self.udi_to_device = {} self.bus = dbus.Bus(dbus.Bus.TYPE_SYSTEM) self.hal_service = self.bus.get_service("org.freedesktop.Hal") self.hal_manager = self.hal_service.get_object("/org/freedesktop/Hal/Manager", "org.freedesktop.Hal.Manager") network_devices = self.hal_manager.FindDeviceByCapability('net') for device_udi in network_devices: self.add_by_udi(device_udi) self.bus.add_signal_receiver(self.device_change_cb, "DeviceAdded", "org.freedesktop.Hal.Manager", "org.freedesktop.Hal", "/org/freedesktop/Hal/Manager") self.bus.add_signal_receiver(self.device_change_cb, "DeviceRemoved", "org.freedesktop.Hal.Manager", "org.freedesktop.Hal", "/org/freedesktop/Hal/Manager") def add_by_udi(self, udi): if not self.udi_to_device.has_key(udi): device = self.hal_service.get_object(udi, "org.freedesktop.Hal.Device") category = device.GetProperty('info.category') if not category.startswith('net.'): return product = device.GetProperty('info.product') interface = device.GetProperty('net.interface') self.udi_to_device[udi] = ("%s: %s" % (interface, product), interface) def remove_by_udi(self, udi): if self.udi_to_device.has_key(udi): self.udi_to_device.pop(udi) def get_devices(self): return self.udi_to_device.values() def device_change_cb(self, dbus_if, dbus_member, dbus_svc, dbus_obj_path, dbus_message): if dbus_member == "DeviceAdded": [device_udi] = dbus_message.get_args_list() print "DeviceAdded", device_udi self.add_by_udi(device_udi) self.change_cb() elif dbus_member == "DeviceRemoved": [device_udi] = dbus_message.get_args_list() print "DeviceRemoved", device_udi self.remove_by_udi(device_udi) self.change_cb() class InterfaceApplet(gnomeapplet.Applet): def __init__(self, applet, iid): self.run_queue = [] self.currently_running = None self.applet = applet self.applet.connect("delete-event", self.cleanup) self.box = gtk.EventBox() self.icon = gtk.Image() self.set_normal_logo() self.box.connect("button-press-event", self.icon_clicked_cb) self.box.add(self.icon) self.applet.add(self.box) self.applet.show_all() self.applet.setup_menu(show_log_xml, [('log_verb_show', self.log_verb_cb), ('interapplet_about', self.about_cb)]) self.tooltips = gtk.Tooltips() self.set_applet_tip(None) self.log_window = gtk.Window() self.log_window.resize(600,150) self.log_window.set_title("InterApplet Log") self.scrolled_window = gtk.ScrolledWindow() self.scrolled_window.set_policy(gtk.POLICY_AUTOMATIC, gtk.POLICY_AUTOMATIC) self.text_view = gtk.TextView() self.text_view.set_editable(False) self.text_view.set_cursor_visible(False) self.scrolled_window.add(self.text_view) self.scrolled_window.show_all() self.log_window_x = None self.log_window_y = None self.log_window.add(self.scrolled_window) self.log_window.connect("delete-event", self.log_window.hide_on_delete) self.log_window_hide() # load interface state and update it using gamin self.if_info = InterfaceInfo(interfaces_file) self.if_state = InterfaceState(ifstate_file) self.device_info = DeviceInfo(self.devices_changed_cb) self.menu = self.build_menu() self.mon = gamin.WatchMonitor() self.mon.watch_file(ifstate_file, self.interfaces_changed_cb) self.mon.watch_file(interfaces_file, self.interfaces_changed_cb) # watch gamin for events gobject.io_add_watch(self.mon.get_fd(), gobject.IO_IN, self.mon_cb) def cleanup(self): # some awful un-python thing we have to do to stop memory leaks del self.applet gtk.main_quit() def set_normal_logo(self): self.icon.set_from_file(normal_logo) def set_active_logo(self): self.icon.set_from_file(active_logo) def update_state(self, mapping, interface): command = "%s %s %s" % (update_state_program, pipes.quote(mapping), pipes.quote(interface)) self.run_command("/usr/bin/gksudo %s" % (pipes.quote(command))) self.log_append_text("\n*** Mapping '%s' now mapped to '%s'\n" % (mapping, interface)) def log_window_hide(self): self.log_window_x, self.log_window_y = self.log_window.get_position() self.log_window.hide() def log_window_show(self): # if self.log_window_x != None and self.log_window_y != None: # self.log_window.move(self.log_window_x, self.log_window_y) self.log_window.show() def log_verb_cb(self, applet, verb): if verb == "log_verb_show": self.log_window_show() elif verb == "log_verb_hide": self.log_window_hide() def log_append_text(self, text): buffer = self.text_view.get_buffer() iter = buffer.get_end_iter() buffer.insert(iter, text) iter = buffer.get_end_iter() self.text_view.scroll_to_iter(iter, 0) def child_data_cb(self, source, condition, user_data=None): data = source.read() if len(data) == 0: # EOF source.close() return False else: self.log_append_text(data) return True def wait_cb(self): # if there is no process running, then nothing to do if self.currently_running == None: return False command, pid = self.currently_running exited_pid, exited_status = os.waitpid(pid, os.WNOHANG) if exited_pid == pid: self.log_append_text("\n*** Command '%s' (%d) has exit with status: %d\n" % (command, pid, exited_status)) self.set_normal_logo() self.currently_running = None self.set_applet_tip(None) self.run_command(None) return False else: # it hasn't quit yet; around we go again return True def set_applet_tip(self, tip): if tip == None: self.tooltips.disable() else: self.tooltips.set_tip(self.applet, tip) self.tooltips.enable() def run_command(self, command): # we only wish to have one command running at any time. # so, firstly this function checks: is self.currently_running None? # if not, append to the run_queue and return # if so, start the command we've been passed # set a periodic timer to try and waitpid() for that command # that timer could do animation! # when waitpid() finds the process, we check the queue and run the # next thing. if command != None: self.run_queue.append(command) if len(self.run_queue) == 0: return if self.currently_running != None: # there is already a command running; # we have appended to the queue. nothing more to do. return to_run, self.run_queue = self.run_queue[0], self.run_queue[1:] # open a pipe to a process. set things up so that we append any output / input from # the child to our vte buffer or whatever. p = popen2.Popen3(to_run, capturestderr=True) set_nonblocking(p.fromchild) set_nonblocking(p.childerr) p.tochild.close() self.currently_running = (to_run, p.pid) gobject.io_add_watch(p.fromchild, gobject.IO_IN | gobject.IO_ERR | gobject.IO_HUP, self.child_data_cb) gobject.io_add_watch(p.childerr, gobject.IO_IN | gobject.IO_ERR | gobject.IO_HUP, self.child_data_cb) # start a callback to watch for this command exiting. self.log_append_text("\n*** Command '%s' (%d) is now running.\n" % (to_run, p.pid)) self.set_applet_tip("Command running: %s" % (to_run)) self.set_active_logo() gobject.timeout_add(500, self.wait_cb) def disable_interface(self, int): command = '/sbin/ifdown %s' % (pipes.quote(int)) self.run_command("/usr/bin/gksudo %s" % (pipes.quote(command))) def enable_interface(self, int, mapping=None): command = '/sbin/ifup ' + pipes.quote(int) if mapping != None: command += "=" + pipes.quote(mapping) self.run_command("/usr/bin/gksudo %s" % (pipes.quote(command))) def mon_cb(self, source, condition, user_data=None): self.mon.handle_events() return True def devices_changed_cb(self): self.menu = self.build_menu() def interfaces_changed_cb(self, path, event): # we only care about the file getting created, or the file # being modified if event != gamin.GAMChanged and event != gamin.GAMCreated: return if path == ifstate_file: self.if_state = InterfaceState(ifstate_file) elif path == interfaces_file: self.if_info = InterfaceInfo(interfaces_file) self.menu = self.build_menu() def radio_menu_action_cb(self, widget, event, param): # ok, a button was pressed on our radio button # we want the UI to be driven by the gamin events, not by # the user.. until we _actually_ enter this new state # leave the UI as it was # # we return True from this callback to prevent the radio # button from updating. self.menu_action_cb(widget, param) # we have to do this manually; we've blocked # the normal way for it to happen self.menu.popdown() return True def menu_action_cb(self, item, param): apply_to, action = param[:2] params = param[2:] if action == "disable": self.disable_interface(apply_to) elif action == "enable": self.enable_interface(apply_to) elif action == "interface": new_interface = params[0] current_interface = self.if_state.state.get(action) if new_interface == current_interface: return self.disable_interface(apply_to) self.update_state(apply_to, params[0]) self.enable_interface(apply_to, params[0]) def build_mapping_menu(self, mapping): menu = gtk.Menu() interfaces = list(self.if_info.interfaces) interfaces.sort() group = None current_interface = self.if_state.state.get(mapping) # add a special "disable" option if we are enabled if current_interface != None: item = gtk.MenuItem("Disable") item.connect("activate", self.menu_action_cb, (mapping, "disable")) item.show() menu.append(item) else: item = gtk.MenuItem("Not enabled") item.show() item.set_sensitive(False) menu.append(item) # then a separator item = gtk.SeparatorMenuItem() item.show() menu.append(item) # add options to change our selected network if len(interfaces) == 0: item = gtk.ImageMenuItem("No interfaces to map to. See README file.") item.set_sensitive(False) item.show() menu.append(item) for interface in interfaces: item = gtk.RadioMenuItem(group, interface, use_underline=False) if interface == current_interface: item.set_active(True) if group == None: group = item item.connect("button-release-event", self.radio_menu_action_cb, (mapping, "interface", interface)) item.show() menu.append(item) return menu def build_dialup_menu(self, dialup): menu = gtk.Menu() if self.if_state.state.has_key(dialup): item = gtk.MenuItem("Disconnect") item.show() item.connect("activate", self.menu_action_cb, (dialup, "disable")) menu.append(item) else: item = gtk.MenuItem("Connect") item.connect("activate", self.menu_action_cb, (dialup, "enable")) item.show() menu.append(item) return menu def build_menu(self): menu = gtk.Menu() mappings = list(self.device_info.get_devices()) mappings.sort(lambda x, y: cmp(x[0], y[0])) nin = gtk.Image() if len(mappings) == 0: item = gtk.ImageMenuItem("No mapable interfaces found.") item.set_sensitive(False) item.show() menu.append(item) for description, mapping in mappings: self.build_mapping_menu(mapping) item = gtk.ImageMenuItem(description) item.set_submenu(self.build_mapping_menu(mapping)) item.show() menu.append(item) dialups = list(self.if_info.dialups) if len(dialups) > 0: item = gtk.SeparatorMenuItem() item.show() menu.append(item) for dialup in dialups: self.build_dialup_menu(dialup) item = gtk.MenuItem(dialup) item.set_submenu(self.build_dialup_menu(dialup)) item.show() menu.append(item) return menu def position_popup_cb(self, widget): x, y = self.applet.window.get_origin() applet_width = self.applet.allocation.width applet_height = self.applet.allocation.height orientation = self.applet.get_orient() if orientation == gnomeapplet.ORIENT_UP: y -= applet_height elif orientation == gnomeapplet.ORIENT_DOWN: y += applet_height elif orientation == gnomeapplet.ORIENT_LEFT: x -= applet_width elif orientation == gnomeapplet.ORIENT_RIGHT: x += applet_width return (x, y, True) def icon_clicked_cb(self, widget, event, user=None): if event.type == gtk.gdk.BUTTON_PRESS and event.button == 1: self.menu.popup(None,None,self.position_popup_cb,event.button,event.time) return False def about_cb(self, widget, user=None): about = gtk.AboutDialog() about.set_name("InterApplet") about.set_version("0.02") about.set_comments("""\ If you like InterApplet, mail me a postcard! PO BOX 3105, Broadway, Nedlands WA 6009, Australia""") about.set_copyright("Copyright 2005 Grahame Bowland") about.set_website("http://grahame.angrygoats.net/interapplet.shtml") about.set_authors(["Grahame Bowland"]) about.set_license("""\ InterApplet Copyright (C) 2005 Grahame Bowland This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA """) about.run() about.destroy() def interface_applet_factory(applet, iid): InterfaceApplet(applet, iid) return gtk.TRUE if __name__ == '__main__': if len(sys.argv) > 1 and sys.argv[1] == 'run-in-window': main_window = gtk.Window(gtk.WINDOW_TOPLEVEL) main_window.set_title('Network Location Applet') main_window.connect('destroy', gtk.main_quit) app = gnomeapplet.Applet() interface_applet_factory(app, None) app.reparent(main_window) main_window.show_all() gtk.main() else: gnomeapplet.bonobo_factory("OAFIID:GNOME_InterApplet_Factory", gnomeapplet.Applet.__gtype__, "InterApplet", "0", interface_applet_factory) sys.exit()