Source Code for Module glue.text_progress_bar

  1  # -*- coding: utf-8 -*- 
  2  # 
  3  # This program is free software; you can redistribute it and/or modify it 
  4  # under the terms of the GNU General Public License as published by the 
  5  # Free Software Foundation; either version 2 of the License, or (at your 
  6  # option) any later version. 
  7  # 
  8  # This program is distributed in the hope that it will be useful, but 
  9  # WITHOUT ANY WARRANTY; without even the implied warranty of 
 10  # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General 
 11  # Public License for more details. 
 12  # 
 13  # You should have received a copy of the GNU General Public License along 
 14  # with this program; if not, write to the Free Software Foundation, Inc., 
 15  # 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA. 
 16  """ 
 17  Text-mode progress bars 
 18  """ 
 19  from __future__ import division, print_function, unicode_literals 
 20  from six.moves import range 
 21   
 22  import locale 
 23  import math 
 24  import os 
 25  import struct 
 26  import sys 
 27  import collections 
 28   
 29  __all__ = ('ProgressBar', 'ProgressBarTheme') 
 30   
 31   
 32  # From http://stackoverflow.com/questions/566746 
 33 -def getTerminalSize(): 
 34      """ 
 35      returns (lines:int, cols:int) 
 36      """ 
 37   
 38      def ioctl_GWINSZ(fd): 
 39          # These two imports are only present on POSIX systems, so they must be 
 40          # guarded by a try block. 
 41          import fcntl 
 42          import termios 
 43          return struct.unpack("hh", fcntl.ioctl(fd, termios.TIOCGWINSZ, "1234")) 
 44      # try stdin, stdout, stderr 
 45      for fd in (0, 1, 2): 
 46          try: 
 47              return ioctl_GWINSZ(fd) 
 48          except: 
 49              pass 
 50      # try os.ctermid() 
 51      try: 
 52          fd = os.open(os.ctermid(), os.O_RDONLY) 
 53          try: 
 54              return ioctl_GWINSZ(fd) 
 55          finally: 
 56              os.close(fd) 
 57      except: 
 58          pass 
 59      # try environment variables 
 60      try: 
 61          return tuple(int(os.getenv(var)) for var in ("LINES", "COLUMNS")) 
 62      except: 
 63          pass 
 64      # i give up. return default. 
 65      return (25, 80) 
 66   
 67   
 68  _ProgressBarTheme = collections.namedtuple( 
 69      '_ProgressBarTheme', 'sequence twiddle_sequence left_border right_border') 
 70   
 71   
 72 -class ProgressBarTheme(_ProgressBarTheme): 
 73   
 74 -    def is_compatible_with_encoding(self, coding): 
 75          if not coding: 
 76              coding = locale.getpreferredencoding() 
 77          try: 
 78              for string in self: 
 79                  string.encode(coding) 
 80          except UnicodeEncodeError: 
 81              return False 
 82          else: 
 83              return True 
 84   
 85 -    def is_compatible_with_stream(self, stream): 
 86          return self.is_compatible_with_encoding(stream.encoding) 
 87   
 88   
 89  default_unicode_theme = ProgressBarTheme( 
 90      ' ▏▎▍▌▋▊▉█', '       ▏▎▍▌▋▊▉██▉▊▋▌▍▎▏ ', '▐', '▌') 
 91   
 92   
 93  default_ascii_theme = ProgressBarTheme( 
 94      ' .:!|', ' ..', ' |', u'| ') 
 95   
 96   
 97 -class ProgressBar: 
 98      """Display a text progress bar. 
 99   
100      A final line feed is printed when the ProgressBar is garbage collected. 
101      Explicitly deleting the object can force a line feed when desired.  As 
102      an alternative, using the ProgressBar as a context manager will ensure 
103      a final line feed is printed when the code block within which the 
104      ProgressBar is being used exits. 
105   
106      Example: 
107   
108      >>> with ProgressBar(max=3) as pb: 
109      ...    pb.update(1) 
110      ...    pb.update(2) 
111      ...    pb.update(3) 
112      ... 
113      """ 
114   
115 -    def __init__( 
116              self, text='Working', max=1, value=0, textwidth=24, fid=None, 
117              theme=None): 
118          if fid is None: 
119              self.fid = sys.stderr 
120          if hasattr(self.fid, 'fileno'): 
121              self.isatty = os.isatty(self.fid.fileno()) 
122          else: 
123              self.isatty = False 
124          if theme is None: 
125              if self.isatty and default_unicode_theme.is_compatible_with_stream( 
126                      self.fid) and 'xterm' in os.environ.get('TERM', ''): 
127                  theme = default_unicode_theme 
128              else: 
129                  theme = default_ascii_theme 
130          self.text = text 
131          self.max = max 
132          self.value = value 
133          self.textwidth = textwidth 
134          self.sequence = ('',) + tuple(theme.sequence) 
135          self.twiddle_sequence = tuple( 
136              theme.twiddle_sequence[-i:] + theme.twiddle_sequence[:-i] 
137              for i in range(len(theme.twiddle_sequence))) 
138          self.left_border = theme.left_border 
139          self.right_border = theme.right_border 
140          self.twiddle = 0 
141          self.linefed = False 
142   
143 -    def iterate(self, iterable, format="%s", print_every=1): 
144          """Use as a target of a for-loop to issue a progress update for every 
145          iteration. For example: 
146   
147          progress = ProgressBar() 
148          for text in progress.iterate(["foo", "bar", "bat"]): 
149              ... 
150          """ 
151   
152          # If iterable has a definite length, then set the maximum value of the 
153          # progress bar. Else, set the maximum value to -1 so that the progress 
154          # bar displays indeterminate progress (scrolling dots). 
155          try: 
156              length = len(iterable) 
157          except TypeError: 
158              self.max = -1 
159          else: 
160              self.max = length 
161   
162          # Iterate over the input, updating the progress bar for each element. 
163          for i, item in enumerate(iterable): 
164              yield item 
165              if i % print_every == 0: 
166                  self.update(i + 1, format % item) 
167   
168 -    def show(self): 
169          """Redraw the text progress bar.""" 
170          # Be silent if writing to a tty 
171          if not self.isatty: 
172              return 
173   
174          if len(self.text) > self.textwidth: 
175              label = self.text[:self.textwidth] 
176          else: 
177              label = self.text.rjust(self.textwidth) 
178   
179          terminalSize = getTerminalSize() 
180          if terminalSize is None: 
181              terminalSize = 80 
182          else: 
183              terminalSize = terminalSize[1] 
184   
185          barWidth = terminalSize - self.textwidth - len(self.left_border) \ 
186              - len(self.right_border) - 7 
187   
188          if self.value is None or self.value < 0: 
189              pattern = self.twiddle_sequence[ 
190                  self.twiddle % len(self.twiddle_sequence)] 
191              self.twiddle += 1 
192              barSymbols = (pattern * int(math.ceil(barWidth / len(self.twiddle_sequence))))[0:barWidth] 
193              progressFractionText = '      ' 
194          else: 
195              progressFraction = max(0.0, min(1.0, float(self.value) / self.max)) 
196   
197              fMinor, iMajor = math.modf(progressFraction * barWidth) 
198              iMajor = int(iMajor) 
199              iMinor = int(math.ceil(fMinor * (len(self.sequence) - 1))) 
200              iMajorMinor = int(math.ceil(progressFraction * barWidth)) 
201   
202              barSymbols = ( 
203                  (self.sequence[-1] * iMajor) + 
204                  self.sequence[iMinor] + 
205                  (self.sequence[1] * (barWidth - iMajorMinor))) 
206              progressFractionText = ('%.1f%%' % (100*progressFraction)).rjust(6) 
207   
208          print( 
209              '\r\x1B[1m', label, '\x1B[0m', self.left_border, '\x1B[36m', 
210              barSymbols, '\x1B[0m', self.right_border, progressFractionText, 
211              sep='', end='', file=self.fid) 
212          self.fid.flush() 
213          self.linefed = False 
214   
215 -    def update(self, value=None, text=None): 
216          """Redraw the progress bar, optionally changing the value and text 
217          and return the (possibly new) value.  For I/O performance, the 
218          progress bar might not be written to the terminal if the text does 
219          not change and the value changes by too little.  Use .show() to 
220          force a redraw.""" 
221          redraw = False 
222          if text is not None: 
223              redraw = text != self.text 
224              self.text = text 
225          if value is not None: 
226              redraw |= self.max == 0 or round(value/(0.0003*self.max)) != \ 
227                  round(self.value/(0.0003*self.max)) 
228              self.value = value 
229          if redraw: 
230              self.show() 
231          return self.value 
232   
233 -    def increment(self, delta=1, text=None): 
234          """Redraw the progress bar, incrementing the value by delta 
235          (default=1) and optionally changing the text.  Returns the 
236          ProgressBar's new value.  See also .update().""" 
237          return self.update(value=min(self.max, self.value + delta), text=text) 
238   
239 -    def linefeed(self): 
240          # Be silent if writing to a tty 
241          if not self.isatty: 
242              return 
243          if not self.linefed: 
244              print(file=self.fid) 
245              self.fid.flush() 
246              self.linefed = True 
247   
248 -    def __enter__(self): 
249          self.show() 
250          return self 
251   
252 -    def __exit__(self, exc_type, exc_value, tb): 
253          try: 
254              self.linefeed() 
255          except: 
256              pass 
257   
258 -    def __del__(self): 
259          self.linefeed() 
260   
261   
262 -def demo(): 
263      """Demonstrate progress bar.""" 
264      from time import sleep 
265      maxProgress = 1000 
266      with ProgressBar(max=maxProgress) as progressbar: 
267          for i in range(-100, maxProgress): 
268              sleep(0.01) 
269              progressbar.update(i + 1) 
270      progressbar2 = ProgressBar(max=maxProgress) 
271      for s in progressbar2.iterate(list(range(maxProgress))): 
272          sleep(0.01) 
273      for s in progressbar2.iterate( 
274              list(range(maxProgress)), format='iteration %d'): 
275          sleep(0.01) 
276   
277   
278  if __name__ == '__main__': 
279      demo() 
280