# Copyright (C) 2005 Aaron Bentley # Copyright (C) 2005 Canonical # # 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 """Simple text-mode progress indicator. To display an indicator, create a ProgressBar object. Call it, passing Progress objects indicating the current state. When done, call clear(). Progress is suppressed when output is not sent to a terminal, so as not to clutter log files. """ # TODO: should be a global option e.g. --silent that disables progress # indicators, preferably without needing to adjust all code that # potentially calls them. # TODO: If not on a tty perhaps just print '......' for the benefit of IDEs, etc # TODO: Optionally show elapsed time instead/as well as ETA; nicer # when the rate is unpredictable import sys import time import os def _width(): """Return estimated terminal width. TODO: Do something smart on Windows? TODO: Is there anything that gets a better update when the window is resized while the program is running? """ try: return int(os.environ['COLUMNS']) except (IndexError, KeyError, ValueError): return 80 def _supports_progress(f): if not hasattr(f, 'isatty'): return False if not f.isatty(): return False if os.environ.get('TERM') == 'dumb': # e.g. emacs compile window return False return True def ProgressBar(to_file=sys.stderr, **kwargs): """Abstract factory""" if _supports_progress(to_file): return TTYProgressBar(to_file=to_file, **kwargs) else: return DotsProgressBar(to_file=to_file, **kwargs) class _BaseProgressBar(object): def __init__(self, to_file=sys.stderr, show_pct=False, show_spinner=False, show_eta=True, show_bar=True, show_count=True): object.__init__(self) self.to_file = to_file self.last_msg = None self.last_cnt = None self.last_total = None self.show_pct = show_pct self.show_spinner = show_spinner self.show_eta = show_eta self.show_bar = show_bar self.show_count = show_count class DummyProgress(_BaseProgressBar): """Progress-bar standin that does nothing. This can be used as the default argument for methods that take an optional progress indicator.""" def tick(self): pass def update(self, msg=None, current=None, total=None): pass def clear(self): pass class DotsProgressBar(_BaseProgressBar): def __init__(self, **kwargs): _BaseProgressBar.__init__(self, **kwargs) self.last_msg = None self.need_nl = False def tick(self): self.update() def update(self, msg=None, current_cnt=None, total_cnt=None): if msg and msg != self.last_msg: if self.need_nl: self.to_file.write('\n') self.to_file.write(msg + ': ') self.last_msg = msg self.need_nl = True self.to_file.write('.') def clear(self): if self.need_nl: self.to_file.write('\n') class TTYProgressBar(_BaseProgressBar): """Progress bar display object. Several options are available to control the display. These can be passed as parameters to the constructor or assigned at any time: show_pct Show percentage complete. show_spinner Show rotating baton. This ticks over on every update even if the values don't change. show_eta Show predicted time-to-completion. show_bar Show bar graph. show_count Show numerical counts. The output file should be in line-buffered or unbuffered mode. """ SPIN_CHARS = r'/-\|' MIN_PAUSE = 0.1 # seconds def __init__(self, **kwargs): _BaseProgressBar.__init__(self, **kwargs) self.spin_pos = 0 self.width = _width() self.start_time = None self.last_update = None def throttle(self): """Return True if the bar was updated too recently""" now = time.time() if self.start_time is None: self.start_time = self.last_update = now return False else: interval = now - self.last_update if interval > 0 and interval < self.MIN_PAUSE: return True self.last_update = now return False def tick(self): self.update(self.last_msg, self.last_cnt, self.last_total) def update(self, msg, current_cnt=None, total_cnt=None): """Update and redraw progress bar.""" if current_cnt < 0: current_cnt = 0 if current_cnt > total_cnt: total_cnt = current_cnt # save these for the tick() function self.last_msg = msg self.last_cnt = current_cnt self.last_total = total_cnt if self.throttle(): return if self.show_eta and self.start_time and total_cnt: eta = get_eta(self.start_time, current_cnt, total_cnt) eta_str = " " + str_tdelta(eta) else: eta_str = "" if self.show_spinner: spin_str = self.SPIN_CHARS[self.spin_pos % 4] + ' ' else: spin_str = '' # always update this; it's also used for the bar self.spin_pos += 1 if self.show_pct and total_cnt and current_cnt: pct = 100.0 * current_cnt / total_cnt pct_str = ' (%5.1f%%)' % pct else: pct_str = '' if not self.show_count: count_str = '' elif current_cnt is None: count_str = '' elif total_cnt is None: count_str = ' %i' % (current_cnt) else: # make both fields the same size t = '%i' % (total_cnt) c = '%*i' % (len(t), current_cnt) count_str = ' ' + c + '/' + t if self.show_bar: # progress bar, if present, soaks up all remaining space cols = self.width - 1 - len(msg) - len(spin_str) - len(pct_str) \ - len(eta_str) - len(count_str) - 3 if total_cnt: # number of markers highlighted in bar markers = int(round(float(cols) * current_cnt / total_cnt)) bar_str = '[' + ('=' * markers).ljust(cols) + '] ' elif False: # don't know total, so can't show completion. # so just show an expanded spinning thingy m = self.spin_pos % cols ms = (' ' * m + '*').ljust(cols) bar_str = '[' + ms + '] ' else: bar_str = '' else: bar_str = '' m = spin_str + bar_str + msg + count_str + pct_str + eta_str assert len(m) < self.width self.to_file.write('\r' + m.ljust(self.width - 1)) #self.to_file.flush() def clear(self): self.to_file.write('\r%s\r' % (' ' * (self.width - 1))) #self.to_file.flush() def str_tdelta(delt): if delt is None: return "-:--:--" delt = int(round(delt)) return '%d:%02d:%02d' % (delt/3600, (delt/60) % 60, delt % 60) def get_eta(start_time, current, total, enough_samples=3): if start_time is None: return None if not total: return None if current < enough_samples: return None if current > total: return None # wtf? elapsed = time.time() - start_time if elapsed < 2.0: # not enough time to estimate return None total_duration = float(elapsed) * float(total) / float(current) assert total_duration >= elapsed return total_duration - elapsed def run_tests(): import doctest result = doctest.testmod() if result[1] > 0: if result[0] == 0: print "All tests passed" else: print "No tests to run" def demo(): sleep = time.sleep print 'dumb-terminal test:' pb = DotsProgressBar() for i in range(100): pb.update('Leoparden', i, 99) sleep(0.1) sleep(1.5) pb.clear() sleep(1.5) print 'smart-terminal test:' pb = ProgressBar(show_pct=True, show_bar=True, show_spinner=False) for i in range(100): pb.update('Elephanten', i, 99) sleep(0.1) sleep(2) pb.clear() sleep(1) print 'done!' if __name__ == "__main__": demo()