1  import time, sched, threading 
 2   
 3 -class latent: 
 4      """ 
 5      Basic tests of periodic scheduling 
 6      """ 
 7 -    def __init__(self, interval): 
 8          self.interval = interval 
 9          self.sch = sched.scheduler(time.time, time.sleep) 
10   
11 -    def run(self): 
12          self.count = 1 
13          self.sch.enter(self.interval, 1, self._loop, ()) 
14          self.startime = time.time() 
15          self.sch.run() 
16          print self.startime 
17   
18 -    def _loop(self): 
19          print time.time() 
20          self.count += 1 
21          elapsed = time.time() - self.startime 
22          next_elapsed = self.count * self.interval 
23          delay = next_elapsed - elapsed 
24          if delay > 0: 
25              self.sch.enter(delay, 1, self._loop, ()) 
26   
27 -class time_thread(threading.Thread): 
28      """ 
29      Basic test of a cyclic timer thread. 
30      """ 
31 -    def __init__(self, interval, stopEvent): 
32          self.interval = interval 
33          self.stop = stopEvent 
34          threading.Thread.__init__(self) 
35   
36 -    def run(self): 
37          count = 0 
38          start_time = time.time() 
39          print start_time 
40          while not self.stop.isSet(): 
41              # Useful stuff to go in here 
42              print time.time() 
43              # End of useful stuff 
44              count += 1 
45              elapsed = time.time() - start_time 
46              next_elapsed = self.interval * count 
47              delay = next_elapsed - elapsed 
48              if delay > 0: 
49                  time.sleep(delay) 
50