#!/usr/bin/python -tt # # Copyright 2006-2007 Red Hat, Inc. # # Jeremy Katz # # 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; version 2 only # # 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 Library 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, sys # # Python gettext: # import gettext import subprocess import time import pygtk pygtk.require("2.0") import gtk import gtk.glade import gobject import gconf import pynotify import dbus import dbus.glib from pirut import * from rhpl.translate import _, N_, textdomain GLADE_FILE = "puplet.glade" ALWAYS_SHOW_KEY = "/apps/puplet/always_show" LAST_UPDATE_CHECK_KEY = "/apps/puplet/last_update_check" NUM_CHECK_FAILS_KEY = "/apps/puplet/num_check_failures" I18N_DOMAIN = "pirut" # # Python gettext: # t = gettext.translation(I18N_DOMAIN, "/usr/share/locale", fallback = True) # _ = t.lgettext class Puplet: def __init__(self): if os.path.exists("data/" + GLADE_FILE): xmlfn = "data/" + GLADE_FILE else: xmlfn = "/usr/share/pirut/ui/" + GLADE_FILE self.xml = gtk.glade.XML(xmlfn, "pupletBox", domain=I18N_DOMAIN) self.menuxml = gtk.glade.XML(xmlfn, "pupletPopupMenu", domain = I18N_DOMAIN) self.pupletMenu = self.menuxml.get_widget("pupletPopupMenu") self.trayicon = gtk.status_icon_new_from_file("/usr/share/pirut/pixmaps/puplet-updated.png") self._oldpix = None self._oldtip = None self._setUpdatedPixbuf() self._connectSignals() self.client = gconf.client_get_default() if not self.client.dir_exists("/apps/puplet"): self.client.add_dir("/apps/puplet", gconf.CLIENT_PRELOAD_NONE) self.client.notify_add(ALWAYS_SHOW_KEY, self._alwaysShowChange) self.connectionfailures = 0 self.security = False self.needsreboot = False self.lastnum = 0 self.bus = dbus.SystemBus() self.updatesObject = None if not pynotify.init("puplet"): print >> sys.stderr, "Error: unable to initialize pynotify" self.notify = None def _setUrgentAvailablePixbuf(self): self._oldpix = self.trayicon.get_pixbuf() if os.path.exists("data/puplet-available.png"): pix = gtk.gdk.pixbuf_new_from_file("data/puplet-available.png") else: pix = gtk.gdk.pixbuf_new_from_file("/usr/share/pirut/pixmaps/puplet-available.png") self.trayicon.set_from_pixbuf(pix) def _setAvailablePixbuf(self): self._oldpix = self.trayicon.get_pixbuf() if os.path.exists("data/puplet-available.png"): pix = gtk.gdk.pixbuf_new_from_file("data/puplet-available.png") else: pix = gtk.gdk.pixbuf_new_from_file("/usr/share/pirut/pixmaps/puplet-available.png") self.trayicon.set_from_pixbuf(pix) def _setDownloadingPixbuf(self): self._oldpix = self.trayicon.get_pixbuf() if os.path.exists("data/puplet-downloading.png"): pix = gtk.gdk.pixbuf_new_from_file("data/puplet-downloading.png") else: pix = gtk.gdk.pixbuf_new_from_file("/usr/share/pirut/pixmaps/puplet-downloading.png") self.trayicon.set_from_pixbuf(pix) def _setErrorPixbuf(self): self._oldpix = self.trayicon.get_pixbuf() if os.path.exists("data/puplet-error.png"): pix = gtk.gdk.pixbuf_new_from_file("data/puplet-error.png") else: pix = gtk.gdk.pixbuf_new_from_file("/usr/share/pirut/pixmaps/puplet-error.png") self.trayicon.set_from_pixbuf(pix) def _setUpdatedPixbuf(self): self._oldpix = self.trayicon.get_pixbuf() if os.path.exists("data/puplet-updated.png"): pix = gtk.gdk.pixbuf_new_from_file("data/puplet-updated.png") else: pix = gtk.gdk.pixbuf_new_from_file("/usr/share/pirut/pixmaps/puplet-updated.png") self.trayicon.set_from_pixbuf(pix) def _resetPixbuf(self): if self._oldpix is not None: self.trayicon.set_from_pixbuf(self._oldpix) self.trayicon.set_tooltip(self._oldtip) def _alwaysShowChange (self, client, cxid, key, data): show = False if key.value.type != gconf.VALUE_BOOL: print >> sys.stderr, "Unable to parse value of key %s!" %(ALWAYS_SHOW_KEY,) else: if key.value.get_bool(): show = True else: try: u = self.updatesObject.GetUpdateInfo(dbus_interface="edu.duke.linux.yum") except dbus.DBusException: u = 0 if len(u) > 0: show = True if show: self.trayicon.set_visible(True) else: self.trayicon.set_visible(False) def _connectSignals(self): menusigs = { "on_view_updates_activate": self._viewUpdates, "on_quit_activate": self._quit, "on_apply_updates_activate": self._applyUpdates, "on_refresh_activate": self._refreshInfo } self.menuxml.signal_autoconnect(menusigs) self.trayicon.connect("popup_menu", self._pupletPopup) self.trayicon.connect("activate", self._clicked) def _clicked(self, status): def menu_pos(menu): return gtk.status_icon_position_menu(menu, self.trayicon) self.pupletMenu.popup(None, None, menu_pos, 0, gtk.get_current_event_time()) self.pupletMenu.show() def _pupletPopup(self, status, button, time): def menu_pos(menu): return gtk.status_icon_position_menu(menu, self.trayicon) self.pupletMenu.popup(None, None, menu_pos, button, time) self.pupletMenu.show() def _refreshInfo(self, *args): self._setDownloadingPixbuf() self.trayicon.set_tooltip(_("Retrieving update information")) try: # if the user clicked Refresh, then show the notification again # and always give a failure if there is one if len(args) and isinstance(args[0], gtk.ImageMenuItem): self.lastnum = -1 if self.updatesObject is None: self._getOnDbus() self.updatesObject.CheckNow(dbus_interface="edu.duke.linux.yum") except dbus.DBusException, e: if self.updatesObject is not None and \ e._dbus_error_name == "org.freedesktop.DBus.Error.ServiceUnknown": self.updatesObject = None self.connectionfailures += 1 self._resetPixbuf() gobject.idle_add(self._refreshInfo) return False self._setErrorPixbuf() self.trayicon.set_tooltip(_("Unable to connect to yum-updatesd. Please ensure that the yum-updatesd package is installed and that the service is running.")) # we couldn't connect -- try again print >> sys.stderr, "Unable to connect to yum-updatesd. Please ensure that the yum-updatesd " print >> sys.stderr, "package is installed and that the service is running." self.connectionfailures += 1 if self.connectionfailures == 10: print >> sys.stderr, "Max failures exceeded, exiting now" sys.exit(0) gobject.timeout_add(600 * 1000, self._refreshInfo) return False def _countUpdates(self, uplst): # we want to count the same way as pup. # FIXME: we should actually _SHARE_ code here :) upds = {} security = False needsreboot = False for (new, old) in uplst: srpm = new["sourcerpm"] if upds.has_key(srpm): upds[srpm].append( (new, old) ) else: upds[srpm] = [ (new, old) ] if new.has_key('type') and new['type'] == 'security': security = True if new.has_key('suggests_reboot') and new['suggests_reboot'] or \ new['name'] in rebootpkgs: needsreboot = True self.security = security self.needsreboot = needsreboot return len(upds.keys()) def _showNotify(self, notify, urgency = pynotify.URGENCY_NORMAL, timeout = 0): if self.notify is not None: self.notify.close() notify.attach_to_status_icon(self.trayicon) notify.set_urgency(urgency) notify.set_timeout(timeout) self.notify = notify if not self.notify.show(): print >> sys.stderr, "Failed to send notification!" def _resetCheckClock(self): # we successfully contacted the servers, so reset the clock # about warning as long as the user hasn't disabled this if self.client.get_int(LAST_UPDATE_CHECK_KEY) != -1 and self.client.get_int(NUM_CHECK_FAILS_KEY) != -1: self.client.set_int(LAST_UPDATE_CHECK_KEY, int(time.time())) self.client.set_int(NUM_CHECK_FAILS_KEY, 0) def _disableCheckFailure(self, *args): self.client.set_int(LAST_UPDATE_CHECK_KEY, -1) self.client.set_int(NUM_CHECK_FAILS_KEY, -1) def _runUpdates(self, autoapply = False): def checkPupRunning(p): if p.poll() is not None: self._refreshInfo() return False return True args = [ "/usr/bin/getproxy", "/usr/bin/pup" ] if autoapply: args.append("--apply") p = subprocess.Popen(args, close_fds = True) self._setDownloadingPixbuf() gobject.timeout_add(5 * 1000, checkPupRunning, p) def _viewUpdates(self, *args): self._runUpdates(False) def _applyUpdates(self, *args): self._runUpdates(True) def _restartSystem(self, *args): subprocess.call(["/usr/bin/reboot"]) self._quit(); def _getOnDbus(self): self.updatesObject = self.bus.get_object("edu.duke.linux.yum", "/Updatesd") self.connectionfailures = 0 def updates_avail_handler(str): self._resetCheckClock() try: num = self._countUpdates(self.updatesObject.GetUpdateInfo(dbus_interface="edu.duke.linux.yum")) except dbus.DBusException, e: # failed to count the updates... print >> sys.stderr, "Error getting update info: %s" %(e,) self.updatesObject = None self.connectionfailures += 1 self._resetPixbuf() gobject.timeout_add(1000, self._refreshInfo) return def updates_avail_notify(): title = _("Updates Available") # # Python gettext: # text = t.ngettext("There is %d package update available.", # "There are %d package updates available.", # num) % num # Gah, rhpl fails: if num != 1: text = _("There are %d package updates available.") else: text = _("There is %d package updates available.") text = text % num urgency = pynotify.URGENCY_NORMAL if self.security: title = _("Security Updates Available") urgency = pynotify.URGENCY_CRITICAL if num == self.lastnum: return self.lastnum = num notify = pynotify.Notification(title, text) notify.add_action("update", _("View Updates..."), self._viewUpdates) self._showNotify(notify, urgency) # s = t.ngettext("%d update available", # "%d updates available", num) % num if num == 1: s = _("%d update available") else: s = _("%d updates available") s = s % num self.trayicon.set_tooltip(s) if self.security: self._setUrgentAvailablePixbuf() else: self._setAvailablePixbuf() self.trayicon.set_visible(True) # FIXME: this is an ugly hack... #203652 gobject.timeout_add(3000, updates_avail_notify) def updates_installed_handler(updinfo): self._resetCheckClock() def hidetrayicon(*args): self.trayicon.set_visible(False) return False num = self._countUpdates(updinfo) self._setUpdatedPixbuf() l = t.ngettext("%d update successfully installed.", "%d updates successfully installed.", num) % num self.trayicon.set_tooltip(l) if not self.needsreboot: notify = pynotify.Notification(_("Update Successful"), l) else: notify = pynotify.Notification(_("Update Successful"), l + "\n" + _("Due to the updates installed, it is recommended " \ "that you reboot your system. You can either " \ "reboot now or choose to do so at a later time.")) notify.add_action("reboot", _("_Reboot now"), self._restartSystem) self._showNotify(notify) if not self.client.get_bool(ALWAYS_SHOW_KEY): gobject.timeout_add(15 * 1000, hidetrayicon) def updates_failed_handler(str): self._resetCheckClock() self.trayicon.set_tooltip(_("Automatic update installation failed!")) self._setErrorPixbuf() self.trayicon.set_visible(True) def no_updates_avail_handler(str): self._resetCheckClock() self.trayicon.set_tooltip(_("No updates available")) self._setUpdatedPixbuf() if not self.client.get_bool(ALWAYS_SHOW_KEY): self.trayicon.set_visible(False) def check_failed_handler(str): print >> sys.stderr, "error getting update info: ", str last = self.client.get_int(LAST_UPDATE_CHECK_KEY) fails = self.client.get_int(NUM_CHECK_FAILS_KEY) now = int(time.time()) failtitle = _("System is not receiving updates") failtxt = _("Your system is currently not receiving software updates. Please check your network connection and/or set up software updates.") showFail = False if fails == -1: # allow the user to set to -1 to disable pass elif fails < 3: # once a day for 3 days if now - last > 86400: showFail = True elif fails < 6: # then, once a week for 3 weeks if now - last > 86400 * 7: showFail = True elif fails < 10: # then once a month for a few months if now - last > 86400 * 30: showFail = True # if the error is that the network isn't available and we # haven't been failing, then don't show the popup. this avoids # getting told on first login that there's not a network # available and then having NM bring it up if str == "No network available" and fails == 0: showFail = False self.client.set_int(NUM_CHECK_FAILS_KEY, fails+1) # we want to set the pixbuf if they explicitly requested # a refresh even if they asked to never be notified again if showFail or self.lastnum == -1: self._setErrorPixbuf() self.trayicon.set_visible(True) self.trayicon.set_tooltip(_("Error retrieving update information")) notify = pynotify.Notification(failtitle, failtxt) notify.add_action("disablecheck", _("Don't notify me again"), self._disableCheckFailure) if showFail: # FIXME: this is an ugly hack... #203652 gobject.timeout_add(3000, self._showNotify, notify) self.client.set_int(LAST_UPDATE_CHECK_KEY, now) self.client.set_int(NUM_CHECK_FAILS_KEY, fails+1) self.lastnum = 0 else: self._resetPixbuf() self.bus.add_signal_receiver(updates_avail_handler, "UpdatesAvailableSignal", dbus_interface="edu.duke.linux.yum") self.bus.add_signal_receiver(no_updates_avail_handler, "NoUpdatesAvailableSignal", dbus_interface="edu.duke.linux.yum") self.bus.add_signal_receiver(updates_failed_handler, "UpdatesFailedSignal", dbus_interface="edu.duke.linux.yum") self.bus.add_signal_receiver(updates_installed_handler, "UpdatesAppliedSignal", dbus_interface="edu.duke.linux.yum") self.bus.add_signal_receiver(check_failed_handler, "CheckFailedSignal", dbus_interface="edu.duke.linux.yum") def run(self): self.trayicon.set_visible(False) if self.client.get_bool(ALWAYS_SHOW_KEY): self.trayicon.set_visible(True) self._refreshInfo() gtk.main() def _quit(self, *args): gtk.main_quit() sys.exit(0) def main(): textdomain(I18N_DOMAIN) gtk.glade.bindtextdomain(I18N_DOMAIN, "/usr/share/locale") p = Puplet() p.run() if __name__ == "__main__": main()