#
# dnfhelper.py
#
# Copyright (C) 2010-2015 Red Hat, Inc.
#
# 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, see <http://www.gnu.org/licenses/>.
#
# Red Hat Author(s): Martin Gracik <mgracik@redhat.com>
# Brian C. Lane <bcl@redhat.com>
#
import logging
logger = logging.getLogger("pylorax.dnfhelper")
import dnf
import dnf.transaction
import collections
import time
import pylorax.output as output
__all__ = ['LoraxDownloadCallback', 'LoraxRpmCallback']
def _paced(fn):
"""Execute `fn` no more often then every 2 seconds."""
def paced_fn(self, *args):
now = time.time()
if now - self.last_time < 2:
return
self.last_time = now
return fn(self, *args)
return paced_fn
[docs]class LoraxDownloadCallback(dnf.callback.DownloadProgress):
def __init__(self):
self.downloads = collections.defaultdict(int)
self.last_time = time.time()
self.total_files = 0
self.total_size = 0
self.pkgno = 0
self.total = 0
self.output = output.LoraxOutput()
@_paced
def _update(self):
msg = "Downloading %(pkgno)s / %(total_files)s RPMs, " \
"%(downloaded)s / %(total_size)s (%(percent)d%%) done.\n"
downloaded = sum(self.downloads.values())
vals = {
'downloaded' : downloaded,
'percent' : int(100 * downloaded/self.total_size),
'pkgno' : self.pkgno,
'total_files' : self.total_files,
'total_size' : self.total_size
}
self.output.write(msg % vals)
[docs] def end(self, payload, status, msg):
nevra = str(payload)
if status is dnf.callback.STATUS_OK:
self.downloads[nevra] = payload.download_size
self.pkgno += 1
self._update()
return
logger.critical("Failed to download '%s': %d - %s", nevra, status, msg)
[docs] def progress(self, payload, done):
nevra = str(payload)
self.downloads[nevra] = done
self._update()
# dnf 2.5.0 adds a new argument, accept it if it is passed
# pylint: disable=arguments-differ
[docs] def start(self, total_files, total_size, total_drpms=0):
self.total_files = total_files
self.total_size = total_size
[docs]class LoraxRpmCallback(dnf.callback.TransactionProgress):
def __init__(self):
super(LoraxRpmCallback, self).__init__()
self._last_ts = None
[docs] def progress(self, package, action, ti_done, ti_total, ts_done, ts_total):
if action == dnf.transaction.PKG_INSTALL: # pylint: disable=no-member
# do not report same package twice
if self._last_ts == ts_done:
return
self._last_ts = ts_done
msg = '(%d/%d) %s' % (ts_done, ts_total, package)
logger.info(msg)
elif action == dnf.transaction.TRANS_POST: # pylint: disable=no-member
msg = "Performing post-installation setup tasks"
logger.info(msg)
[docs] def error(self, message):
logger.warning(message)