Timer can be scheduled for one-shot or repeating alarms. Can be given

npryce · npryce · commit 258e6e40ebcc · 2013-01-21T23:31:29.000Z
offset and interval and then started/stopped.
diff --git a/quick2wire/test_timerfd.py b/quick2wire/test_timerfd.py
@@ -1,5 +1,5 @@
 
-from time import time
+from time import time, sleep
 from contextlib import closing
 from quick2wire.timerfd import Timer, timespec, itimerspec
 
@@ -9,18 +9,68 @@ def test_timespec_can_be_created_from_seconds():
     assert t.tv_sec == 4
     assert t.tv_nsec == 125000000
 
+
 def test_itimerspec_can_be_created_from_seconds():
     t = itimerspec.from_seconds(offset=4.125, interval=1.25)
     assert t.it_value.tv_sec == 4
     assert t.it_value.tv_nsec == 125000000
     assert t.it_interval.tv_sec == 1
     assert t.it_interval.tv_nsec == 250000000
 
+
 def test_timer_waits_for_time_to_pass():
-    with closing(Timer()) as timer:
+    with closing(Timer(offset=0.125)) as timer:
         start = time()
-        timer.schedule(0.125)
-        n = timer.wait()
+        
+        timer.start()
+        timer.wait()
+        
+        duration = time() - start
+        
+        assert duration >= 0.125
+
+
+def test_timer_can_repeat_with_interval():
+    with closing(Timer(interval=0.125)) as timer:
+        start = time()
+        
+        timer.start()
+        timer.wait()
+        timer.wait()
+        
         duration = time() - start
         
-        assert duration / n >= 0.125
+        assert duration >= 0.25
+
+
+def test_timer_can_repeat_with_interval_after_offset():
+    with closing(Timer(offset=0.25, interval=0.125)) as timer:
+        start = time()
+        
+        timer.start()
+        timer.wait()
+        timer.wait()
+        timer.wait()
+        
+        duration = time() - start
+        
+        assert duration >= 0.5
+
+
+def test_timer_cannot_be_started_if_offset_and_interval_are_both_zero():
+    with closing(Timer()) as timer:
+        try:
+            timer.start()
+            assert False, "should have thrown ValueError"
+        except ValueError:
+            # expected
+            pass
+
+
+def test_timer_reports_how_many_times_it_triggered_since_last_wait():
+    with closing(Timer(interval=0.0125)) as timer:
+        timer.start()
+        sleep(0.5)
+        n = timer.wait()
+        
+        assert n >= 4
diff --git a/quick2wire/timerfd.py b/quick2wire/timerfd.py
@@ -101,25 +101,45 @@ def from_seconds(cls, offset, interval):
 
 
 class Timer:
-    def __init__(self, clock=CLOCK_REALTIME, blocking=True):
+    def __init__(self, clock=CLOCK_REALTIME, offset=0, interval=0, blocking=True):
         self._fd = timerfd_create(clock, (not blocking)*TFD_NONBLOCK)
-        if self._fd < 0:
-            e = get_errno()
-            raise OSError(e, errno.strerror(e))
+        self._offset = offset
+        self._interval = interval
+        self._started = False
     
     def close(self):
         os.close(self._fd)
     
-    def schedule(self, offset=0, interval=0):
-        spec = itimerspec(it_value=timespec.from_seconds(offset), 
-                          it_interval=timespec.from_seconds(interval))
+    @property
+    def offset(self):
+        return self._offset
+    
+    @offset.setter
+    def offset(self, new_offset):
+        self._offset = new_offset
+        if self._started:
+            self._schedule()
+    
+    @property
+    def interval(self):
+        return self._interval
+    
+    @interval.setter
+    def interval(self, new_interval):
+        self._interval = new_interval
+        if self._started:
+            self._schedule()
+    
+    def start(self):
+        if self._offset == 0 and self._interval == 0:
+            raise ValueError("timer will not fire because offset and interval are both zero")
         
-        print(spec)
+        self._schedule(self._offset or self._interval, self._interval)
+        self._started = True
         
-        status = timerfd_settime(self._fd, 0, byref(spec), None)
-        if status < 0:
-            e = get_errno()
-            raise OSError(e, errno.strerror(e))
+    def stop(self):
+        self._schedule(0, 0)
+        self._started = False
     
     def wait(self):
         try:
@@ -130,4 +150,8 @@ def wait(self):
                 return 0
             else:
                 raise e
-        
+    
+    def _schedule(self, offset, interval):
+        spec = itimerspec.from_seconds(offset, interval)
+        timerfd_settime(self._fd, 0, byref(spec), None)
+    
