feat: add Timer, a small context-manager/manual elapsed-time timer

stream2py/__init__.py

@@ -176,5 +176,6 @@ def info(self):
 from stream2py.stream_buffer import *
 from stream2py.source_reader import *
 from stream2py.stream_source import *
+from stream2py.util import Timer
 
 # from stream2py.simply import mk_stream_buffer

stream2py/tests/test_util.py

@@ -1,6 +1,8 @@
 """Testing the util.py module"""
 
-from stream2py.util import contextualize_with_instance
+import pytest
+
+from stream2py.util import contextualize_with_instance, Timer
 
 
 class StreamHasNotBeenStarted(RuntimeError):
@@ -126,3 +128,78 @@ def test_reader(reader):
         it_worked = False
 
     assert it_worked  # Hurray!
+
+
+# ---------------------------------------------------------------------------------------
+# Timer
+
+
+def test_timer_manual_start_stop():
+    """A manually started timer reports a positive, monotonically growing elapsed time."""
+    timer = Timer()
+    timer.start()
+    first = timer.elapsed()
+    assert first >= 0
+    second = timer.elapsed()
+    assert second >= first  # monotonic clock: never goes backwards
+    timer.stop()
+
+
+def test_timer_as_context_manager():
+    """Entering the context starts the timer and yields the timer itself."""
+    with Timer() as timer:
+        assert isinstance(timer, Timer)
+        assert timer.elapsed() >= 0
+    # leaving the context stops it
+    assert timer.start_time is None
+
+
+def test_timer_egress_is_applied():
+    """The egress function transforms the elapsed seconds."""
+    with Timer(lambda seconds: 'transformed') as timer:
+        assert timer.elapsed() == 'transformed'
+
+
+def test_timer_is_reusable_across_contexts():
+    """The same instance can be re-entered after being stopped."""
+    timer = Timer()
+    with timer:
+        pass
+    assert timer.start_time is None
+    with timer:
+        assert timer.elapsed() >= 0
+    assert timer.start_time is None
+
+
+def test_timer_elapsed_before_start_raises_informatively():
+    """Asking a stopped timer for elapsed time raises ValueError, not TypeError.
+
+    This is the regression guard for the original implementation, which detected the
+    not-started case by catching TypeError from ``time() - None``. That conflated "timer
+    not started" with "egress itself raised TypeError" -- see the test below.
+    """
+    with pytest.raises(ValueError, match='not running'):
+        Timer().elapsed()
+
+
+def test_timer_does_not_swallow_egress_errors():
+    """An exception raised by egress propagates, rather than being silently dropped.
+
+    The original implementation wrapped the whole computation in ``try/except TypeError``
+    and, when ``start_time`` was not None, fell off the end of the function -- returning
+    None and hiding the real error.
+    """
+
+    def broken_egress(seconds):
+        raise TypeError('egress is broken')
+
+    with Timer(broken_egress) as timer:
+        with pytest.raises(TypeError, match='egress is broken'):
+            timer.elapsed()
+
+
+def test_timer_stop_accepts_context_manager_exit_args():
+    """__exit__ is stop(), so it must tolerate the (exc_type, exc, tb) triple."""
+    timer = Timer().start()
+    timer.stop(ValueError, ValueError('x'), None)
+    assert timer.start_time is None

stream2py/util.py

@@ -5,6 +5,84 @@
 from inspect import signature, Parameter
 from abc import abstractmethod
 from functools import wraps
+from time import perf_counter
+
+
+def identity(x):
+    """Returns the input, as is."""
+    return x
+
+
+class Timer:
+    """A simple timer that can be used as a context manager, or started and stopped
+    manually.
+
+    >>> from time import sleep
+    >>> timer = Timer()
+    >>> timer.start()  # doctest: +SKIP
+    >>> sleep(0.01)
+    >>> timer.elapsed()  # doctest: +SKIP
+    0.010056478977203369
+    >>> timer.stop()
+
+    You can also use a context block, and specify an ``egress`` function that will be
+    called on the elapsed seconds:
+
+    >>> with Timer(lambda x: print(f"{int(x / (60 * 60 * 24))} days elapsed")) as timer:
+    ...     sleep(0.01)
+    ...     timer.elapsed()
+    0 days elapsed
+
+    The same timer instance can be reused across context blocks:
+
+    >>> timer = Timer(lambda x: int(x / 3600))
+    >>> with timer:
+    ...     sleep(0.01)
+    ...     t = timer.elapsed()
+    >>> t
+    0
+    >>> with timer:
+    ...     sleep(0.01)
+
+    Asking for the elapsed time of a timer that isn't running tells you so, instead of
+    failing obscurely:
+
+    >>> Timer().elapsed()  # doctest: +ELLIPSIS
+    Traceback (most recent call last):
+      ...
+    ValueError: The timer is not running....
+
+    Note that elapsed time is measured with ``time.perf_counter``, a monotonic clock, so
+    it is unaffected by system clock adjustments. This also means ``start_time`` is not a
+    wall-clock timestamp; it is only meaningful as a reference point for a difference.
+    """
+
+    def __init__(self, egress=identity):
+        self.egress = egress
+        self.start_time = None
+
+    def start(self):
+        self.start_time = perf_counter()
+        return self
+
+    def stop(self, *args, **kwargs):
+        self.start_time = None
+
+    def elapsed(self):
+        """The (egress-transformed) seconds since the timer was started.
+
+        Raises ``ValueError`` if the timer is not running.
+        """
+        if self.start_time is None:
+            raise ValueError(
+                "The timer is not running, so it has no elapsed time. "
+                "Start it with timer.start(), or use a context block "
+                "(`with timer: ...` or `with Timer() as timer: ...`)."
+            )
+        return self.egress(perf_counter() - self.start_time)
+
+    __enter__ = start
+    __exit__ = stop
 
 
 class TypeValidationError(TypeError):