#!/usr/bin/env python # -*- coding: utf-8 -*- """ widgets - GTK widgets to show the progress and status of apt Copyright (c) 2004,2005 Canonical Ltd. Authors: Michael Vogt Sebastian Heinlein 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. his 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 """ from gettext import gettext as _ import os import time import pygtk pygtk.require('2.0') import gtk import gobject import pango import vte import apt import apt_pkg # Seconds until a maintainer script will be regarded as hanging INSTALL_TIMEOUT = 4 * 60 class GOpProgress(gobject.GObject, apt.progress.OpProgress): __gsignals__ = {"status-changed":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, (gobject.TYPE_STRING, gobject.TYPE_INT)), "status-started":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ()), "status-finished":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ())} def __init__(self): apt.progress.OpProgress.__init__(self) gobject.GObject.__init__(self) def update(self, percent): self.emit("status-changed", self.op, percent) while gtk.events_pending(): gtk.main_iteration() def done(self): self.emit("status-finished") class GInstallProgress(gobject.GObject, apt.progress.InstallProgress): __gsignals__ = {"status-changed":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, (gobject.TYPE_STRING, gobject.TYPE_INT)), "status-started":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ()), "status-timeout":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ()), "status-error":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ()), "status-conffile":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ()), "status-finished":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ())} def __init__(self, term): apt.progress.InstallProgress.__init__(self) gobject.GObject.__init__(self) self.finished = False self.time_last_update = time.time() self.term = term reaper = vte.reaper_get() reaper.connect("child-exited", self.childExited) self.env = ["VTE_PTY_KEEP_FD=%s"% self.writefd, "DEBIAN_FRONTEND=gnome", "APT_LISTCHANGES_FRONTEND=gtk"] def childExited(self, term, pid, status): self.apt_status = os.WEXITSTATUS(status) self.finished = True def error(self, pkg, errormsg): self.emit("status-error") def conffile(self, current, new): self.emit("status-conffile") def startUpdate(self): self.emit("status-started") def finishUpdate(self): self.emit("status-finished") def statusChange(self, pkg, percent, status): self.time_last_update = time.time() self.emit("status-changed", status, percent) def updateInterface(self): apt.progress.InstallProgress.updateInterface(self) while gtk.events_pending(): gtk.main_iteration() if self.time_last_update + INSTALL_TIMEOUT < time.time(): self.emit("status-timeout") def fork(self): return self.term.forkpty(envv=self.env) def waitChild(self): while not self.finished: self.updateInterface() return self.apt_status class GDpkgInstallProgress(apt.progress.DpkgInstallProgress,GInstallProgress): def run(self, debfile): apt.progress.DpkgInstallProgress.run(self, debfile) def updateInterface(self): apt.progress.DpkgInstallProgress.updateInterface(self) while gtk.events_pending(): gtk.main_iteration() if self.time_last_update + INSTALL_TIMEOUT < time.time(): self.emit("status-timeout") class GFetchProgress(gobject.GObject, apt.progress.FetchProgress): __gsignals__ = {"status-changed":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, (gobject.TYPE_STRING, gobject.TYPE_INT)), "status-started":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ()), "status-finished":(gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ())} def __init__(self): apt.progress.FetchProgress.__init__(self) gobject.GObject.__init__(self) self._continue = True def start(self): self.emit("status-started") def stop(self): self.emit("status-finished") def cancel(self): self._continue = False def pulse(self): apt.progress.FetchProgress.pulse(self) currentItem = self.currentItems + 1 if currentItem > self.totalItems: currentItem = self.totalItems if self.currentCPS > 0: text = (_("Downloading file %(current)li of %(total)li with " "%(speed)s/s") % \ {"current" : currentItem, "total" : self.totalItems, "speed" : humanize_size(self.currentCPS)}) else: text = (_("Downloading file %(current)li of %(total)li") % \ {"current" : currentItem, "total" : self.totalItems }) self.emit("status-changed", text, self.percent) while gtk.events_pending(): gtk.main_iteration() return self._continue class GtkAptProgress(gtk.VBox): """ This widget provides a progress bar, a terminal and a status bar for showing the progress of package manipulation tasks. A simple example code snippet to install/remove a package: import pygtk pygtk.require('2.0') import gtk import apt.widgets win = gtk.Window() progress = apt.widgets.GtkAptProgress() win.set_title("GtkAptProgress Demo") win.add(progress) progress.show() win.show() cache = apt.cache.Cache(progress.get_open_progress()) cache["xterm"].markDelete() progress.show_terminal(expanded=True) cache.commit(progress.get_fetch_progress(), progress.get_install_progress()) gtk.main() """ def __init__(self): gtk.VBox.__init__(self) self.set_spacing(6) # Setup some child widgets self._expander = gtk.Expander(_("Details")) self._terminal = vte.Terminal() #self._terminal.set_font_from_string("monospace 10") self._expander.add(self._terminal) self._progressbar = gtk.ProgressBar() # Setup the always italic status label self._label = gtk.Label() attr_list = pango.AttrList() attr_list.insert(pango.AttrStyle(pango.STYLE_ITALIC, 0, -1)) self._label.set_attributes(attr_list) self._label.set_ellipsize(pango.ELLIPSIZE_END) self._label.set_alignment(0, 0) # add child widgets self.pack_start(self._progressbar, False) self.pack_start(self._label, False) self.pack_start(self._expander, False) # Setup the internal progress handlers self._progress_open = GOpProgress() self._progress_open.connect("status-changed", self._on_status_changed) self._progress_open.connect("status-started", self._on_status_started) self._progress_open.connect("status-finished", self._on_status_finished) self._progress_fetch = GFetchProgress() self._progress_fetch.connect("status-changed", self._on_status_changed) self._progress_fetch.connect("status-started", self._on_status_started) self._progress_fetch.connect("status-finished", self._on_status_finished) self._progress_install = GInstallProgress(self._terminal) self._progress_install.connect("status-changed", self._on_status_changed) self._progress_install.connect("status-started", self._on_status_started) self._progress_install.connect("status-finished", self._on_status_finished) self._progress_install.connect("status-timeout", self._on_status_timeout) self._progress_install.connect("status-error", self._on_status_timeout) self._progress_install.connect("status-conffile", self._on_status_timeout) self._progress_dpkg_install = GDpkgInstallProgress(self._terminal) self._progress_dpkg_install.connect("status-changed", self._on_status_changed) self._progress_dpkg_install.connect("status-started", self._on_status_started) self._progress_dpkg_install.connect("status-finished", self._on_status_finished) self._progress_dpkg_install.connect("status-timeout", self._on_status_timeout) self._progress_dpkg_install.connect("status-error", self._on_status_timeout) self._progress_dpkg_install.connect("status-conffile", self._on_status_timeout) def clear(self): """ Reset all status information """ self._label.set_label("") self._progress.set_fraction(0) self._expander.set_expanded(False) def get_open_progress(self): """ Return the cache opening progress handler. """ return self._progress_open def get_install_progress(self): """ Return the install progress handler """ return self._progress_install def get_dpkg_install_progress(self): """ Return the install progress handler for dpkg """ return self._dpkg_progress_install def get_fetch_progress(self): """ Return the fetch progress handler """ return self._progress_fetch def _on_status_started(self, progress): self._on_status_changed(progress, _("Starting..."), 0) def _on_status_finished(self, progress): self._on_status_changed(progress, _("Complete"), 100) def _on_status_changed(self, progress, status, percent): self._label.set_text(status) self._progressbar.pulse() self._progressbar.set_fraction(percent/100.0) def _on_status_timeout(self, progress): selt._expander.set_expanded(True) def cancel_download(self): """ Cancel a currently running download """ self._progress_fetch.cancel() def show_terminal(self, expanded=False): """ Show an expander with a terminal widget which provides a way to interact with dpkg """ self._expander.show() self._terminal.show() self._expander.set_expanded(True) def hide_terminal(self): """ Hide the expander with the terminal widget """ self._expander.hide() def show(self): gtk.HBox.show(self) self._label.show() self._progressbar.show() if __name__ == "__main__": import sys import debfile win = gtk.Window() apt_progress = GtkAptProgress() win.set_title("GtkAptProgress Demo") win.add(apt_progress) apt_progress.show() win.show() cache = apt.cache.Cache(apt_progress.get_open_progress()) pkg = cache["xterm"] if pkg.isInstalled: pkg.markDelete() else: pkg.markInstall() apt_progress.show_terminal(True) try: cache.commit(apt_progress.get_fetch_progress(), apt_progress.get_install_progress()) except: pass if len(sys.argv) > 1: deb = DebPackage(sys.argv[1], cache) deb.install(apt_progress.get_dpkg_install_progress()) gtk.main() # vim: ts=4 et sts=4