Timers class

CHANGELOG.md

@@ -7,7 +7,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased]
 
-## [1.1.0] - 2020-01-14
+### Changed
+
+- `Timer.timers` changed from regular to `dict` to a custom dictionary supporting basic statistics for named timers.
+
+
+## [1.1.0] - 2020-01-15
 
 ### Added
 

codetiming/_timer.py

@@ -9,7 +9,10 @@
 import time
 from contextlib import ContextDecorator
 from dataclasses import dataclass, field
-from typing import Any, Callable, ClassVar, Dict, Optional
+from typing import Any, Callable, ClassVar, Optional
+
+# Codetiming imports
+from codetiming._timers import Timers
 
 
 class TimerError(Exception):
@@ -20,18 +23,13 @@ class TimerError(Exception):
 class Timer(ContextDecorator):
     """Time your code using a class, context manager, or decorator"""
 
-    timers: ClassVar[Dict[str, float]] = dict()
+    timers: ClassVar[Timers] = Timers()
     _start_time: Optional[float] = field(default=None, init=False, repr=False)
     name: Optional[str] = None
     text: str = "Elapsed time: {:0.4f} seconds"
     logger: Optional[Callable[[str], None]] = print
     last: float = field(default=math.nan, init=False, repr=False)
 
-    def __post_init__(self) -> None:
-        """Initialization: add timer to dict of timers"""
-        if self.name:
-            self.timers.setdefault(self.name, 0)
-
     def start(self) -> None:
         """Start a new timer"""
         if self._start_time is not None:
@@ -52,7 +50,7 @@ def stop(self) -> float:
         if self.logger:
             self.logger(self.text.format(self.last))
         if self.name:
-            self.timers[self.name] += self.last
+            self.timers.add(self.name, self.last)
 
         return self.last
 

codetiming/_timers.py

@@ -0,0 +1,75 @@
+"""Dictionary-like structure with information about timers"""
+
+# Standard library imports
+import collections
+import math
+import statistics
+from typing import TYPE_CHECKING, Any, Callable, Dict, List
+
+# Annotate generic UserDict
+if TYPE_CHECKING:
+    UserDict = collections.UserDict[str, float]  # pragma: no cover
+else:
+    UserDict = collections.UserDict
+
+
+class Timers(UserDict):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Add a private dictionary keeping track of all timings"""
+        super().__init__(*args, **kwargs)
+        self._timings: Dict[str, List[float]] = collections.defaultdict(list)
+
+    def add(self, name: str, value: float) -> None:
+        """Add a timing value to the given timer"""
+        self._timings[name].append(value)
+        self.data.setdefault(name, 0)
+        self.data[name] += value
+
+    def clear(self) -> None:
+        """Clear timers"""
+        self.data.clear()
+        self._timings.clear()
+
+    def __setitem__(self, name: str, value: float) -> None:
+        """Disallow setting of timer values"""
+        raise TypeError(
+            f"{self.__class__.__name__!r} does not support item assignment. "
+            "Use '.add()' to update values."
+        )
+
+    def apply(self, func: Callable[[List[float]], float], name: str) -> float:
+        """Apply a function to the results of one named timer"""
+        if name in self._timings:
+            return func(self._timings[name])
+        raise KeyError(name)
+
+    def count(self, name: str) -> float:
+        """Number of timings"""
+        return self.apply(len, name=name)
+
+    def total(self, name: str) -> float:
+        """Total time for timers"""
+        return self.apply(sum, name=name)
+
+    def min(self, name: str) -> float:
+        """Minimal value of timings"""
+        return self.apply(lambda values: min(values or [0]), name=name)
+
+    def max(self, name: str) -> float:
+        """Maximal value of timings"""
+        return self.apply(lambda values: max(values or [0]), name=name)
+
+    def mean(self, name: str) -> float:
+        """Mean value of timings"""
+        return self.apply(lambda values: statistics.mean(values or [0]), name=name)
+
+    def median(self, name: str) -> float:
+        """Median value of timings"""
+        return self.apply(lambda values: statistics.median(values or [0]), name=name)
+
+    def stdev(self, name: str) -> float:
+        """Standard deviation of timings"""
+        if name in self._timings:
+            value = self._timings[name]
+            return statistics.stdev(value) if len(value) >= 2 else math.nan
+        raise KeyError(name)

tests/test_codetiming.py

@@ -21,14 +21,19 @@
 RE_TIME_MESSAGE = re.compile(TIME_PREFIX + r" 0\.\d{4} seconds")
 
 
+def waste_time(num=1000):
+    """Just waste a little bit of time"""
+    sum(n ** 2 for n in range(num))
+
+
 @Timer(text=TIME_MESSAGE)
-def timewaster(num):
+def decorated_timewaste(num=1000):
     """Just waste a little bit of time"""
     sum(n ** 2 for n in range(num))
 
 
 @Timer(name="accumulator", text=TIME_MESSAGE)
-def accumulated_timewaste(num):
+def accumulated_timewaste(num=1000):
     """Just waste a little bit of time"""
     sum(n ** 2 for n in range(num))
 
@@ -48,7 +53,7 @@ def __call__(self, message):
 #
 def test_timer_as_decorator(capsys):
     """Test that decorated function prints timing information"""
-    timewaster(1000)
+    decorated_timewaste()
     stdout, stderr = capsys.readouterr()
     assert RE_TIME_MESSAGE.match(stdout)
     assert stdout.count("\n") == 1
@@ -58,7 +63,7 @@ def test_timer_as_decorator(capsys):
 def test_timer_as_context_manager(capsys):
     """Test that timed context prints timing information"""
     with Timer(text=TIME_MESSAGE):
-        sum(n ** 2 for n in range(1000))
+        waste_time()
     stdout, stderr = capsys.readouterr()
     assert RE_TIME_MESSAGE.match(stdout)
     assert stdout.count("\n") == 1
@@ -69,7 +74,7 @@ def test_explicit_timer(capsys):
     """Test that timed section prints timing information"""
     t = Timer(text=TIME_MESSAGE)
     t.start()
-    sum(n ** 2 for n in range(1000))
+    waste_time()
     t.stop()
     stdout, stderr = capsys.readouterr()
     assert RE_TIME_MESSAGE.match(stdout)
@@ -96,14 +101,14 @@ def test_custom_logger():
     """Test that we can use a custom logger"""
     logger = CustomLogger()
     with Timer(text=TIME_MESSAGE, logger=logger):
-        sum(n ** 2 for n in range(1000))
+        waste_time()
     assert RE_TIME_MESSAGE.match(logger.messages)
 
 
 def test_timer_without_text(capsys):
     """Test that timer with logger=None does not print anything"""
     with Timer(logger=None):
-        sum(n ** 2 for n in range(1000))
+        waste_time()
 
     stdout, stderr = capsys.readouterr()
     assert stdout == ""
@@ -112,8 +117,8 @@ def test_timer_without_text(capsys):
 
 def test_accumulated_decorator(capsys):
     """Test that decorated timer can accumulate"""
-    accumulated_timewaste(1000)
-    accumulated_timewaste(1000)
+    accumulated_timewaste()
+    accumulated_timewaste()
 
     stdout, stderr = capsys.readouterr()
     lines = stdout.strip().split("\n")
@@ -127,9 +132,9 @@ def test_accumulated_context_manager(capsys):
     """Test that context manager timer can accumulate"""
     t = Timer(name="accumulator", text=TIME_MESSAGE)
     with t:
-        sum(n ** 2 for n in range(1000))
+        waste_time()
     with t:
-        sum(n ** 2 for n in range(1000))
+        waste_time()
 
     stdout, stderr = capsys.readouterr()
     lines = stdout.strip().split("\n")
@@ -144,10 +149,10 @@ def test_accumulated_explicit_timer(capsys):
     t = Timer(name="accumulated_explicit_timer", text=TIME_MESSAGE)
     total = 0
     t.start()
-    sum(n ** 2 for n in range(1000))
+    waste_time()
     total += t.stop()
     t.start()
-    sum(n ** 2 for n in range(1000))
+    waste_time()
     total += t.stop()
 
     stdout, stderr = capsys.readouterr()
@@ -179,3 +184,57 @@ def test_timer_sets_last():
         time.sleep(0.02)
 
     assert t.last >= 0.02
+
+
+def test_timers_cleared():
+    """Test that timers can be cleared"""
+    with Timer(name="timer_to_be_cleared"):
+        waste_time()
+
+    assert "timer_to_be_cleared" in Timer.timers
+    Timer.timers.clear()
+    assert not Timer.timers
+
+
+def test_running_cleared_timers():
+    """Test that timers can still be run after they're cleared"""
+    t = Timer(name="timer_to_be_cleared")
+    Timer.timers.clear()
+
+    accumulated_timewaste()
+    with t:
+        waste_time()
+
+    assert "accumulator" in Timer.timers
+    assert "timer_to_be_cleared" in Timer.timers
+
+
+def test_timers_stats():
+    """Test that we can get basic statistics from timers"""
+    name = "timer_with_stats"
+    t = Timer(name=name)
+    for num in range(5, 10):
+        with t:
+            waste_time(num=100 * num)
+
+    stats = Timer.timers
+    assert stats.total(name) == stats[name]
+    assert stats.count(name) == 5
+    assert stats.min(name) <= stats.median(name) <= stats.max(name)
+    assert stats.mean(name) >= stats.min(name)
+    assert stats.stdev(name) >= 0
+
+
+def test_stats_missing_timers():
+    """Test that getting statistics from non-existent timers raises exception"""
+    with pytest.raises(KeyError):
+        Timer.timers.count("non_existent_timer")
+
+    with pytest.raises(KeyError):
+        Timer.timers.stdev("non_existent_timer")
+
+
+def test_setting_timers_exception():
+    """Test that setting .timers items raises exception"""
+    with pytest.raises(TypeError):
+        Timer.timers["set_timer"] = 1.23