Source Code for Module gen.utils.longop

  1  import time 
  2   
  3  from callback import Callback 
  4   
  5 -class LongOpStatus(Callback): 
  6      """LongOpStatus provides a way of communicating the status of a long 
  7      running operations. The intended use is that when a long running operation 
  8      is about to start it should create an instance of this class and emit 
  9      it so that any listeners can pick it up and use it to record the status  
 10      of the operation. 
 11   
 12   
 13      Signals 
 14      ======= 
 15       
 16        op-heartbeat - emitted every 'interval' calls to heartbeat.  
 17        op-end       - emitted once when the operation completes. 
 18   
 19       
 20      Example usage: 
 21   
 22      class MyClass(Callback): 
 23   
 24          __signals__ = { 
 25             'op-start'   : object 
 26          } 
 27       
 28          def long(self): 
 29          status = LongOpStatus("doing long job", 100, 10) 
 30   
 31              for i in xrange(0,99): 
 32              time.sleep(0.1) 
 33              status.heartbeat() 
 34   
 35              status.end() 
 36       
 37   
 38      class MyListener(object): 
 39   
 40           def __init__(self): 
 41           self._op = MyClass() 
 42           self._op.connect('op-start', self.start) 
 43           self._current_op = None 
 44   
 45           def start(self,long_op): 
 46           self._current_op.connect('op-heartbeat', self.heartbeat) 
 47           self._current_op.connect('op-end', self.stop) 
 48   
 49           def hearbeat(self): 
 50           # update status display 
 51   
 52           def stop(self): 
 53           # close the status display 
 54               self._current_op = None 
 55      """ 
 56   
 57      __signals__ = { 
 58          'op-heartbeat'   : None, 
 59          'op-end'         : None 
 60          } 
 61   
 62 -    def __init__(self, msg="",  
 63                   total_steps=None,  
 64                   interval=1, 
 65                   can_cancel=False): 
 66          """ 
 67          @param msg: A Message to indicated the purpose of the operation. 
 68          @type msg: string 
 69           
 70          @param total_steps: The total number of steps that the operation  
 71          will perform. 
 72          @type total_steps: 
 73           
 74          @param interval: The number of iterations between emissions. 
 75          @type interval: 
 76           
 77          @param can_cancel: Set to True if the operation can be cancelled. 
 78          If this is set the operation that creates the status object should 
 79          check the 'should_cancel' method regularly so that it can cancel  
 80          the operation. 
 81          @type can_cancel: 
 82          """ 
 83          Callback.__init__(self) 
 84          self._msg = msg 
 85          self._total_steps = total_steps 
 86          # don't allow intervals less that 1 
 87          self._interval = max(interval,1)  
 88          self._can_cancel = can_cancel 
 89           
 90          self._cancel = False 
 91          self._count = 0 
 92          self._countdown = interval 
 93          self._secs_left = 0 
 94          self._start = time.time() 
 95          self._running = True 
 96   
 97 -    def __del__(self): 
 98          if self._running: 
 99              self.emit('op-end') 
100               
101 -    def heartbeat(self): 
102          """This should be called for each step in the operation. It will 
103          emit a 'op-heartbeat' every 'interval' steps. It recalcuates the 
104          'estimated_secs_to_complete' from the time taken for previous  
105          steps. 
106          """ 
107          self._countdown -= 1 
108          if self._countdown <= 0: 
109              elapsed = time.time() - self._start 
110              self._secs_left = \ 
111                  ( elapsed / self._interval ) \ 
112                  * (self._total_steps - self._count) 
113              self._count += self._interval 
114              self._countdown = self._interval 
115              self._start = time.time() 
116              self.emit('op-heartbeat') 
117   
118 -    def estimated_secs_to_complete(self): 
119          """Return the number of seconds estimated left before operation  
120          completes. This will change as 'hearbeat' is called. 
121       
122          @return: estimated seconds to complete. 
123          @rtype: int 
124          """ 
125          return self._secs_left 
126   
127 -    def cancel(self): 
128          """Inform the operation that it should complete. 
129          """ 
130          self._cancel = True 
131          self.end() 
132   
133 -    def end(self): 
134          """End the operation. Causes the 'op-end' signal to be emitted.     
135          """ 
136          self.emit('op-end') 
137          self._running = False 
138   
139 -    def should_cancel(self): 
140          """Return true of the user has asked for the operation to be cancelled. 
141       
142          @return: True of the operation should be cancelled. 
143          @rtype: bool 
144          """ 
145          return self._cancel 
146   
147 -    def can_cancel(self): 
148          """@return: True if the operation can be cancelled. 
149             @rtype: bool 
150             """ 
151          return self._can_cancel 
152   
153 -    def get_msg(self): 
154          """@return: The current status description messages. 
155             @rtype: string 
156             """ 
157          return self._msg 
158   
159 -    def set_msg(self, msg):         
160          """Set the current description message. 
161           
162          @param msg: The description message. 
163          @type msg: string 
164          """ 
165          self._msg = msg 
166           
167 -    def get_total_steps(self): 
168          """Get to total number of steps. NOTE: this is not the  
169          number of times that the 'op-heartbeat' message will be 
170          emited. 'op-heartbeat' is emited get_total_steps/interval 
171          times. 
172           
173          @return: total number of steps. 
174          @rtype: int 
175          """ 
176          return self._total_steps 
177   
178 -    def get_interval(self): 
179          """Get the interval between 'op-hearbeat' signals. 
180           
181          @return: the interval between 'op-hearbeat' signals. 
182          @rtype: int 
183          """ 
184          return self._interval  
185   
186   
187  if __name__ == '__main__': 
188   
189      s = LongOpStatus("msg", 100, 10) 
190   
191 -    def heartbeat(): 
192          print "heartbeat ", s.estimated_secs_to_complete() 
193   
194 -    def end(): 
195          print "end" 
196   
197      s.connect('op-heartbeat', heartbeat) 
198      s.connect('op-end', end) 
199   
200      for i in xrange(0, 99): 
201          time.sleep(0.1) 
202          s.heartbeat() 
203   
204      s.end() 
205