/usr/share/pyshared/debpartial_mirror/DisplayStatus.py is in debpartial-mirror 0.3.1.
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 | # debpartial_mirror - partial debian mirror package tool
# (c) 2004 Otavio Salvador <otavio@debian.org>, Enrico Zini <enrico@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
import fcntl
import os.path
import struct
import sys
import termios
class StatusItem:
id = 0
def __init__(self):
self._properties = None
def __getitem__(self, key):
if key not in ['id', 'finished', 'current', 'size', 'errored']:
raise KeyError("%s is not allowed on StatusItem." % key)
if self._properties is None:
StatusItem.id += 1
self._properties = {'id' : StatusItem.id,
'finished': False,
'current' : 0,
'size' : 0,
'errored' : ''}
return self._properties.get(key)
def __setitem__(self, key, value):
if key not in ['id', 'finished', 'current', 'size', 'errored']:
raise KeyError("%s is not allowed on StatusItem." % key)
if self._properties is None:
StatusItem.id += 1
self._properties = {'id' : StatusItem.id,
'finished': False,
'current' : 0,
'size' : 0,
'errored' : ''}
self._properties[key] = value
class BaseDisplayStatus:
_items = {}
def __getitem__ (self, key):
return self._items.get(key, None)
def start(self, url, size):
self._items[url] = StatusItem()
self._items[url]['size'] = size
self._update()
def update(self, url, current):
self._items[url]['current'] = current
self._update()
def errored(self, url, message):
if not self._items.has_key(url):
self._items[url] = StatusItem()
self._items[url]['errored'] = message
self._update()
def _update(self):
pass
def clean(self):
for i in self._items.keys():
self._items.pop(i)
class TextDisplayStatus(BaseDisplayStatus):
def __print_status(self):
self.__clear_line()
for url in self._items.keys():
if self._items[url]['finished'] or self._items[url]['errored']:
continue
try:
sys.stdout.write("[%d %s: %.2f%%]"
% (self._items[url]['id'],
os.path.basename(url).split('_')[0],
self._items[url]['current']*100/
(self._items[url]['size'])))
except ZeroDivisionError:
pass
sys.stdout.flush()
def __clear_line(self):
if sys.stdout.isatty():
sw = struct.unpack(
"hhhh",
fcntl.ioctl(1, termios.TIOCGWINSZ, " "*8)
)[1]
sys.stdout.write("\r" + " " * sw + "\r")
def start(self, url, size):
BaseDisplayStatus.start(self, url, size)
self.__clear_line()
sys.stdout.write("\rGetting %d: %s\n" % (self._items[url]['id'], url))
sys.stdout.flush()
def update(self, url, current):
BaseDisplayStatus.update(self, url, current)
self.__clear_line()
if current >= self._items[url]['size'] and \
not self._items[url]['finished']:
sys.stdout.write("\rDone: %s\n" % url)
self._items[url]['finished'] = True
self.__print_status()
def errored(self, url, message):
if not self._items.has_key(url) or not self._items[url]['errored']:
BaseDisplayStatus.errored(self, url, message)
self.__clear_line()
sys.stdout.write("\rFailed: %s %s\n" % (url, message))
sys.stdout.flush()
class LogDisplayStatus(BaseDisplayStatus):
def start(self, url, size):
BaseDisplayStatus.start(self, url, size)
print "Getting %d: %s" % (self._items[url]['id'], url)
def update(self, url, current):
BaseDisplayStatus.update(self, url, current)
if current >= self._items[url]['size'] and \
not self._items[url]['finished']:
self._items[url]['finished'] = True
def errored(self, url, message):
if not self._items.has_key(url) or not self._items[url]['errored']:
BaseDisplayStatus.errored(self, url, message)
print "Failed: %s %s\n" % (url, message)
class QuietDisplayStatus(BaseDisplayStatus):
def start(self, url, size):
pass
def update(self, url, current):
pass
def errored(self, url, message):
if not self._items.has_key(url) or not self._items[url]['errored']:
BaseDisplayStatus.errored(self, url, message)
print "Failed: %s %s\n" % (url, message)
class GtkDisplayStatus(BaseDisplayStatus):
#Same as TextDisplayStatus, but send data to a widget instead of print
pass
|