/usr/share/pyshared/apt/progress/text.py is in python-apt 0.8.3ubuntu7.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 | # Copyright (c) 2009 Julian Andres Klode <jak@debian.org>
#
# 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
"""Progress reporting for text interfaces."""
import os
import sys
import apt_pkg
from apt.progress import base
__all__ = ['AcquireProgress', 'CdromProgress', 'OpProgress']
def _(msg):
"""Translate the message, also try apt if translation is missing."""
res = apt_pkg.gettext(msg)
if res == msg:
res = apt_pkg.gettext(msg, "apt")
return res
class TextProgress(object):
"""Internal Base class for text progress classes."""
def __init__(self, outfile=None):
self._file = outfile or sys.stdout
self._width = 0
def _write(self, msg, newline=True, maximize=False):
"""Write the message on the terminal, fill remaining space."""
self._file.write("\r")
self._file.write(msg)
# Fill remaining stuff with whitespace
if self._width > len(msg):
self._file.write((self._width - len(msg)) * ' ')
elif maximize: # Needed for OpProgress.
self._width = max(self._width, len(msg))
if newline:
self._file.write("\n")
else:
#self._file.write("\r")
self._file.flush()
class OpProgress(base.OpProgress, TextProgress):
"""Operation progress reporting.
This closely resembles OpTextProgress in libapt-pkg.
"""
def __init__(self, outfile=None):
TextProgress.__init__(self, outfile)
base.OpProgress.__init__(self)
self.old_op = ""
def update(self, percent=None):
"""Called periodically to update the user interface."""
base.OpProgress.update(self, percent)
if self.major_change and self.old_op:
self._write(self.old_op)
self._write("%s... %i%%\r" % (self.op, self.percent), False, True)
self.old_op = self.op
def done(self):
"""Called once an operation has been completed."""
base.OpProgress.done(self)
if self.old_op:
self._write(_("%c%s... Done") % ('\r', self.old_op), True, True)
self.old_op = ""
class AcquireProgress(base.AcquireProgress, TextProgress):
"""AcquireProgress for the text interface."""
def __init__(self, outfile=None):
TextProgress.__init__(self, outfile)
base.AcquireProgress.__init__(self)
self._signal = None
self._width = 80
self._id = 1
def start(self):
"""Start an Acquire progress.
In this case, the function sets up a signal handler for SIGWINCH, i.e.
window resize signals. And it also sets id to 1.
"""
base.AcquireProgress.start(self)
import signal
self._signal = signal.signal(signal.SIGWINCH, self._winch)
# Get the window size.
self._winch()
self._id = 1L
def _winch(self, *dummy):
"""Signal handler for window resize signals."""
if hasattr(self._file, "fileno") and os.isatty(self._file.fileno()):
import fcntl
import termios
import struct
buf = fcntl.ioctl(self._file, termios.TIOCGWINSZ, 8 * ' ')
dummy, col, dummy, dummy = struct.unpack('hhhh', buf)
self._width = col - 1 # 1 for the cursor
def ims_hit(self, item):
"""Called when an item is update (e.g. not modified on the server)."""
base.AcquireProgress.ims_hit(self, item)
line = _('Hit ') + item.description
if item.owner.filesize:
line += ' [%sB]' % apt_pkg.size_to_str(item.owner.filesize)
self._write(line)
def fail(self, item):
"""Called when an item is failed."""
base.AcquireProgress.fail(self, item)
if item.owner.status == item.owner.STAT_DONE:
self._write(_("Ign ") + item.description)
else:
self._write(_("Err ") + item.description)
self._write(" %s" % item.owner.error_text)
def fetch(self, item):
"""Called when some of the item's data is fetched."""
base.AcquireProgress.fetch(self, item)
# It's complete already (e.g. Hit)
if item.owner.complete:
return
item.owner.id = self._id
self._id += 1
line = _("Get:") + "%s %s" % (item.owner.id, item.description)
if item.owner.filesize:
line += (" [%sB]" % apt_pkg.size_to_str(item.owner.filesize))
self._write(line)
def pulse(self, owner):
"""Periodically invoked while the Acquire process is underway.
Return False if the user asked to cancel the whole Acquire process."""
base.AcquireProgress.pulse(self, owner)
percent = (((self.current_bytes + self.current_items) * 100.0) /
float(self.total_bytes + self.total_items))
shown = False
tval = '%i%%' % percent
end = ""
if self.current_cps:
eta = long(float(self.total_bytes - self.current_bytes) /
self.current_cps)
end = " %sB/s %s" % (apt_pkg.size_to_str(self.current_cps),
apt_pkg.time_to_str(eta))
for worker in owner.workers:
val = ''
if not worker.current_item:
if worker.status:
val = ' [%s]' % worker.status
if len(tval) + len(val) + len(end) >= self._width:
break
tval += val
shown = True
continue
shown = True
if worker.current_item.owner.id:
val += " [%i %s" % (worker.current_item.owner.id,
worker.current_item.shortdesc)
else:
val += ' [%s' % worker.current_item.description
if worker.current_item.owner.mode:
val += ' %s' % worker.current_item.owner.mode
val += ' %sB' % apt_pkg.size_to_str(worker.current_size)
# Add the total size and percent
if worker.total_size and not worker.current_item.owner.complete:
val += "/%sB %i%%" % (apt_pkg.size_to_str(worker.total_size),
worker.current_size*100.0/worker.total_size)
val += ']'
if len(tval) + len(val) + len(end) >= self._width:
# Display as many items as screen width
break
else:
tval += val
if not shown:
tval += _(" [Working]")
if self.current_cps:
tval += (self._width - len(end) - len(tval)) * ' ' + end
self._write(tval, False)
return True
def media_change(self, medium, drive):
"""Prompt the user to change the inserted removable media."""
base.AcquireProgress.media_change(self, medium, drive)
self._write(_("Media change: please insert the disc labeled\n"
" '%s'\n"
"in the drive '%s' and press enter\n") % (medium, drive))
return raw_input() not in ('c', 'C')
def stop(self):
"""Invoked when the Acquire process stops running."""
base.AcquireProgress.stop(self)
# Trick for getting a translation from apt
self._write((_("Fetched %sB in %s (%sB/s)\n") % (
apt_pkg.size_to_str(self.fetched_bytes),
apt_pkg.time_to_str(self.elapsed_time),
apt_pkg.size_to_str(self.current_cps))).rstrip("\n"))
# Delete the signal again.
import signal
signal.signal(signal.SIGWINCH, self._signal)
class CdromProgress(base.CdromProgress, TextProgress):
"""Text CD-ROM progress."""
def ask_cdrom_name(self):
"""Ask the user to provide a name for the disc."""
base.CdromProgress.ask_cdrom_name(self)
self._write(_("Please provide a name for this Disc, such as "
"'Debian 2.1r1 Disk 1'"), False)
try:
return raw_input(":")
except KeyboardInterrupt:
return
def update(self, text, current):
"""Set the current progress."""
base.CdromProgress.update(self, text, current)
if text:
self._write(text, False)
def change_cdrom(self):
"""Ask the user to change the CD-ROM."""
base.CdromProgress.change_cdrom(self)
self._write(_("Please insert a Disc in the drive and press enter"),
False)
try:
return (raw_input() == '')
except KeyboardInterrupt:
return False
|