Best way to report progress at fixed intervals

[MRAB]

Slaunger wrote:
> Hi comp.lang.python
> 
> I am a novice Python 2.5 programmer, who write some cmd line scripts
> for processing large amounts of data.
> 
> I would like to have possibility to regularly print out the progress
> made during the processing, say every 1 seconds, and i am wondering
> what a proper generic way to do this is.
> 
> I have created this test example to show the general problem. Running
> the script gives me the output:
> 
> Work through all 20 steps reporting progress every 1.0 secs...
> Work step 0
> Work step 1
> Work step 2
> Work step 3
> Work step 4
> Processed 4 of 20
> Work step 5
> Work step 6
> Work step 7
> Work step 8
> Processed 8 of 20
> Work step 9
> Work step 10
> Work step 11
> Work step 12
> Work step 13
> Processed 13 of 20
> Work step 14
> Work step 15
> Work step 16
> Work step 17
> Processed 17 of 20
> Work step 18
> Work step 19
> Finished working through 20 steps
> 
> The script that does this is as follows:
> 
> testregularprogress.py:
> 
> """
> Test module for testing generic ways of displaying progress
> information
> at regular intervals.
> """
> import random
> import threading
> import time
> 
> def work(i):
>     """
>     Dummy process function, which takes a random time in the interval
>     0.0-0.5 secs to execute
>     """
>     print "Work step %d" % i
>     time.sleep(0.5 * random.random())
> 
> 
> def workAll(verbose=True, max_iter=20, progress_interval=1.0):
> 
>     class _Progress(object):
> 
>         def __init__(self):
>             self.no = 0
>             self.max = max_iter
>             self.start_timer = verbose
> 
>         def __str__(self):
>             self.start_timer = True # I do not like this appraoch
>             return "Processed %d of %d" % (self.no, self.max)
> 
>     p = _Progress()
> 
>     def report_progress():
>         print p
> 
>     if verbose:
>         print "Work through all %d steps reporting progress every
> %3.1f secs..." % \
>             (max_iter, progress_interval)
> 
>     for i in xrange(max_iter):
>         if p.start_timer :
>             p.start_timer = False # Let the progress instance set the
> flag
>             timer = threading.Timer(progress_interval,
> report_progress)
>             timer.start()
>         work(i)
>         p.no = i + 1
> 
>     # Kill the last timer, which is still active at this time
>     timer.cancel()
> 
>     if verbose:
>         print "Finished working through %d steps" % max_iter
> 
> if __name__ == "__main__":
>     workAll()
> 
> Quite frankly, I do not like what I have made! It is a mess,
> responsibilities are mixed, and it seems overly complicated. But I
> can't figure out how to do this right.
> 
> I would therefore like some feedback on this proposed generic "report
> progress at regular intervals" approach presented here. What could I
> do better?
> 
I've come up with this:

"""
Test module for testing generic ways of displaying progress
information
at regular intervals.
"""
import random
import threading
import time

def work(i):
     """
     Dummy process function, which takes a random time in the interval
     0.0-0.5 secs to execute
     """
     print "Work step %d" % i
     time.sleep(0.5 * random.random())

def workAll(verbose=True, max_iter=20, progress_interval=1.0):
     class _Progress(threading.Thread):
         def __init__(self, progress_interval=1.0):
             threading.Thread.__init__(self)
             self.setDaemon(True)
             self.progress_interval = progress_interval
             self.progress = None
             self.active = True

         def stop(self):
             self.active = False

         def run(self):
             while self.active:
                 if self.progress is not None:
                     print self.progress
                 time.sleep(self.progress_interval)

     if verbose:
         print "Work through all %d steps reporting progress every %3.1f 
secs..." % \
             (max_iter, progress_interval)

     p = _Progress(progress_interval)
     p.start()

     for i in xrange(max_iter):
         p.progress = "Processed %d of %d" % (i + 1, max_iter)
         work(i)

     p.stop()

     if verbose:
         print "Finished working through %d steps" % max_iter

if __name__ == "__main__":
     workAll()