diff --git a/CHANGELOG.md b/CHANGELOG.md index df13bfd..5e53000 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/codetiming/_timer.py b/codetiming/_timer.py index 14645a8..ac96763 100644 --- a/codetiming/_timer.py +++ b/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 diff --git a/codetiming/_timers.py b/codetiming/_timers.py new file mode 100644 index 0000000..7ff2e6a --- /dev/null +++ b/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) diff --git a/tests/test_codetiming.py b/tests/test_codetiming.py index 4916213..fa3a559 100644 --- a/tests/test_codetiming.py +++ b/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