Viewing file: puplet (18.25 KB) -rwxr-xr-x Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) | #!/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()
|