Merge pull request #143 from vkottler/dev/3.2.7

3.2.7 - New global time control interfaces

.github/workflows/python-package.yml

@@ -70,7 +70,7 @@ jobs:
 
       - run: |
           mk python-release owner=vkottler \
-                            repo=vcorelib version=3.2.6
+                            repo=vcorelib version=3.2.7
         if: |
           matrix.python-version == '3.11'
           && matrix.system == 'ubuntu-latest'

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.4
-    hash=2a33e265f89ea4a52c3bafa9befb4f7c
+    hash=175c456c69021cd0f3288321acf73e39
     =====================================
 -->
 
-# vcorelib ([3.2.6](https://pypi.org/project/vcorelib/))
+# vcorelib ([3.2.7](https://pypi.org/project/vcorelib/))
 
 [![python](https://img.shields.io/pypi/pyversions/vcorelib.svg)](https://pypi.org/project/vcorelib/)
 ![Build Status](https://github.com/vkottler/vcorelib/workflows/Python%20Package/badge.svg)

local/variables/package.yaml

@@ -1,4 +1,4 @@
 ---
 major: 3
 minor: 2
-patch: 6
+patch: 7

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "vcorelib"
-version = "3.2.6"
+version = "3.2.7"
 description = "A collection of core Python utilities."
 readme = "README.md"
 requires-python = ">=3.11"

tests/math/test_time.py

@@ -5,11 +5,14 @@
 # module under test
 from vcorelib.math import (
     BILLION,
+    SimulatedTime,
     byte_count_str,
     default_time_ns,
     nano_str,
     rate_str,
+    restore_time_source,
     seconds_str,
+    set_simulated_source,
     simulated_time,
 )
 
@@ -20,6 +23,10 @@ def test_simulated_time_basic():
     context.
     """
 
+    sim = SimulatedTime(1)
+    set_simulated_source(sim)
+    restore_time_source()
+
     start = default_time_ns()
 
     with simulated_time(start_ns=0) as sim_time:

vcorelib/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.4
-# hash=6f119f3118ec4c02f71649c29826d64d
+# hash=6c63e42f58901f5bb2b0dd07a76fa949
 # =====================================
 
 """
@@ -10,7 +10,7 @@
 
 DESCRIPTION = "A collection of core Python utilities."
 PKG_NAME = "vcorelib"
-VERSION = "3.2.6"
+VERSION = "3.2.7"
 
 # vcorelib-specific content.
 DEFAULT_INDENT = 2

vcorelib/math/__init__.py

@@ -13,20 +13,27 @@
 from vcorelib.math.analysis.rate import RateTracker
 from vcorelib.math.analysis.rate.limiter import RateLimiter
 from vcorelib.math.constants import BILLION, MILLION, from_nanos, to_nanos
+from vcorelib.math.keeper import SimulatedTime
 from vcorelib.math.time import (
     TIMER,
     LoggerType,
     Timer,
     byte_count_str,
     default_time_ns,
+    metrics_time_ns,
     nano_str,
     rate_str,
+    restore_time_source,
     seconds_str,
+    set_simulated_source,
     simulated_time,
 )
 from vcorelib.math.unit import KIBI_UNITS, SI_UNITS, UnitSystem, unit_traverse
 
 __all__ = [
+    "metrics_time_ns",
+    "restore_time_source",
+    "set_simulated_source",
     "MILLION",
     "BILLION",
     "from_nanos",
@@ -51,4 +58,5 @@
     "FloatBuffer",
     "MovingSum",
     "WeightedAverage",
+    "SimulatedTime",
 ]

vcorelib/math/keeper.py

@@ -45,6 +45,11 @@ def __init__(self, source: TimeSource = _time_ns) -> None:
         """Initialize this instance."""
 
         self.source = source
+        self.orig = self.source
+
+    def restore(self) -> None:
+        """Restore the original time source."""
+        self.source = self.orig
 
     @contextmanager
     def simulated(
@@ -59,13 +64,12 @@ def simulated(
         sim_time = SimulatedTime(step_dt_ns, start_ns=start_ns)
 
         # Update 'source' to power simulated time resolution.
-        orig = self.source
         self.source = sim_time
 
         try:
             yield sim_time
         finally:
-            self.source = orig
+            self.restore()
 
     def __call__(self) -> int:
         """Get time."""

vcorelib/math/time.py

@@ -31,6 +31,21 @@ def default_time_ns() -> int:
     return _TIME()
 
 
+def metrics_time_ns() -> int:
+    """Get a timestamp suitable for runtime performance metrics."""
+    return _perf_counter_ns()
+
+
+def set_simulated_source(source: _SimulatedTime) -> None:
+    """Set a simulated time source for the global clock."""
+    _TIME.source = source
+
+
+def restore_time_source() -> None:
+    """Restore the original global time source."""
+    _TIME.restore()
+
+
 @contextmanager
 def simulated_time(
     step_dt_ns: int = 1, start_ns: int = None
@@ -137,11 +152,11 @@ def measure_ns(self) -> _Iterator[int]:
 
         curr = self.curr
         self.curr += 1
-        start = _perf_counter_ns()
+        start = metrics_time_ns()
         try:
             yield curr
         finally:
-            self.data[curr] = _perf_counter_ns() - start
+            self.data[curr] = metrics_time_ns() - start
 
     def result(self, token: int) -> int:
         """Get the timer result."""