rpmdevtools/progressbar.py
Michal Domonkos 987b185581 Drop progressbar2 dependency and use our own
This is to reduce the dependency set of rpmdevtools.

Resolves: #1927784
2021-02-11 15:50:56 +01:00

134 lines
3.8 KiB
Python

#!/usr/bin/python3
# -*- coding: utf-8 -*-
#
# A simple text-based progress bar, compatible with the basic API of:
# https://github.com/WoLpH/python-progressbar
#
# Copyright (C) 2021 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
# USA.
import shutil
import sys
import time
class ProgressBar:
FORMAT = '{value:>10} / {max_value:<10} [{bars}]'
BARS = '= '
SPINLEN = 5
def __init__(self, stream=sys.stderr, max_width=80, fps=10):
self._stream = stream
self._max_width = max_width
self._min_delay = 1 / fps
@staticmethod
def _format_value(value):
raise NotImplementedError()
def start(self, max_value):
self._value = 0
self._max_value = max_value or 0
self._status = dict()
self._spinner = 0
self._timestamp = 0
self.update(0)
def update(self, value):
self._value = value
if value > self._max_value:
self._max_value = 0
ts = time.time()
if (ts - self._timestamp) < self._min_delay:
return
self._timestamp = ts
status = {'value': self._format_value(value),
'max_value': self._format_value(self._max_value) \
if self._max_value else '???',
'bars': ''}
termw = min(shutil.get_terminal_size()[0], self._max_width)
nbars = max(termw - len(self.FORMAT.format(**status)), 0)
nfill = nskip = 0
if self._max_value:
nfill = round(nbars * value / self._max_value)
elif nbars > self.SPINLEN:
nfill = self.SPINLEN
nskip = self._spinner % (nbars - self.SPINLEN)
self._spinner = nskip + 1
status['bars'] = self.BARS[1] * nskip + \
self.BARS[0] * nfill + \
self.BARS[1] * (nbars - nfill - nskip)
if status == self._status:
return
self._status = status
self._stream.write('\r')
self._stream.write(self.FORMAT.format(**self._status))
self._stream.flush()
def finish(self):
self._max_value = self._value
self._timestamp = 0 # Force an update
self.update(self._value)
self._stream.write('\n')
self._stream.flush()
class DataTransferBar(ProgressBar):
@staticmethod
def _format_value(value):
symbols = ' KMGTPEZY'
depth = 0
max_depth = len(symbols) - 1
unit = 1024.0
# 1023.95 should be formatted as 1.0 (not 1024.0)
# More info: https://stackoverflow.com/a/63839503
thres = unit - 0.05
while value >= thres and depth < max_depth:
depth += 1
value /= unit
symbol = ' %siB' % symbols[depth] if depth > 0 else ''
return '%.1f%s' % (value, symbol)
if __name__ == '__main__':
# Show a dummy bar for debugging purposes
bar = DataTransferBar()
size = 50*1024*1024
chunk = 1024*1234
recvd = 0
bar.start(size)
while recvd < (size - chunk):
recvd += chunk
bar.update(recvd)
time.sleep(0.1)
bar.update(size)
bar.finish()