3.2.6 - Add ability to control default time source

Also add 'saturated' check for circular float buffers

.github/workflows/python-package.yml

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

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2023 Vaughn Kottler
+Copyright (c) 2024 Vaughn Kottler
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.4
-    hash=80fddc32b8f7fd652a35b06893040c4f
+    hash=2a33e265f89ea4a52c3bafa9befb4f7c
     =====================================
 -->
 
-# vcorelib ([3.2.5](https://pypi.org/project/vcorelib/))
+# vcorelib ([3.2.6](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: 5
+patch: 6

pyproject.toml

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

tests/math/analysis/test_average.py

@@ -13,6 +13,8 @@ def test_moving_average_basic():
     """Test basic functionality of the moving average."""
 
     average = MovingAverage(10)
+    assert not average.saturated
+
     assert average(10.0) == approx(1.0)
     assert average(10.0) == approx(2.0)
     assert average(10.0) == approx(3.0)
@@ -22,7 +24,11 @@ def test_moving_average_basic():
     assert average(10.0) == approx(7.0)
     assert average(10.0) == approx(8.0)
     assert average(10.0) == approx(9.0)
+    assert not average.saturated
+
     assert average(10.0) == approx(10.0)
+    assert average.saturated
+
     assert average(0.0) == approx(9.0)
     average.resize(100)
     assert average(100.0) == approx(1.0)

tests/math/analysis/test_buffer.py

@@ -0,0 +1,26 @@
+"""
+Test the 'math.analysis.buffer' module.
+"""
+
+# module under test
+from vcorelib.math.analysis.buffer import FloatBuffer
+
+
+def test_float_buffer_basic():
+    """Test basic interactions with a circular buffer."""
+
+    buffer = FloatBuffer()
+    assert not buffer.saturated
+
+    buffer(1.0)
+    assert not buffer.saturated
+
+    for _ in range(buffer.depth - 2):
+        buffer(1.0)
+    assert not buffer.saturated
+
+    buffer(1.0)
+    assert buffer.saturated
+
+    buffer.reset()
+    assert not buffer.saturated

tests/math/analysis/test_weighted.py

@@ -20,3 +20,10 @@ def test_weighted_average_basic():
     expected = (10.0 / 6.0) + (1.0 * (5.0 / 6.0))
 
     assert average.average() == expected
+
+    assert not average.saturated
+
+    average.reset()
+    for _ in range(average.depth):
+        average(1.0)
+    assert average.saturated

tests/math/test_time.py

@@ -3,7 +3,49 @@
 """
 
 # module under test
-from vcorelib.math import byte_count_str, nano_str, rate_str, seconds_str
+from vcorelib.math import (
+    BILLION,
+    byte_count_str,
+    default_time_ns,
+    nano_str,
+    rate_str,
+    seconds_str,
+    simulated_time,
+)
+
+
+def test_simulated_time_basic():
+    """
+    Test that we can take control over the global time source as a managed
+    context.
+    """
+
+    start = default_time_ns()
+
+    with simulated_time(start_ns=0) as sim_time:
+        assert default_time_ns() == 0
+        sim_time.step()
+        assert default_time_ns() == 1
+        sim_time.step(-1)
+        assert default_time_ns() == 0
+
+        sim_time.step_s(1.0)
+        assert default_time_ns() == BILLION
+
+        sim_time.step_s(-1.0)
+        assert default_time_ns() == 0
+
+    with simulated_time(100, start_ns=0) as sim_time:
+        assert default_time_ns() == 0
+        sim_time.step()
+        assert default_time_ns() == 100
+        sim_time.step(-2)
+        assert default_time_ns() == -100
+
+    with simulated_time() as sim_time:
+        assert default_time_ns() >= start
+
+    assert default_time_ns() >= start
 
 
 def test_nano_str_basic():

vcorelib/__init__.py

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

vcorelib/math/__init__.py

@@ -22,6 +22,7 @@
     nano_str,
     rate_str,
     seconds_str,
+    simulated_time,
 )
 from vcorelib.math.unit import KIBI_UNITS, SI_UNITS, UnitSystem, unit_traverse
 
@@ -38,6 +39,7 @@
     "default_time_ns",
     "nano_str",
     "seconds_str",
+    "simulated_time",
     "rate_str",
     "LoggerType",
     "Timer",

vcorelib/math/analysis/average.py

@@ -79,6 +79,11 @@ def reset(self, initial: float = 0.0) -> None:
         self.min = initial
         self._initialized = False
 
+    @property
+    def saturated(self) -> bool:
+        """Determine if the buffer is saturated with elements yet."""
+        return self.buffer.saturated
+
     def resize(self, depth: int, initial: float = 0.0) -> None:
         """Set a new depth for this moving average and reset the value."""
 

vcorelib/math/analysis/buffer.py

@@ -19,6 +19,7 @@ def __init__(
         self._index: int = 0
         self.data: _List[float] = []
         self.depth: int = depth
+        self.elements: int = 0
 
         self.reset(initial=initial)
 
@@ -35,11 +36,22 @@ def __call__(self, value: float) -> float:
 
         self._index += 1
 
+        # Keep track of how full the buffer is.
+        if self.elements < self.depth:
+            self.elements += 1
+
         return oldest
 
+    @property
+    def saturated(self) -> bool:
+        """Determine if the buffer is saturated with elements yet."""
+        return self.elements == self.depth
+
     def reset(self, initial: float = 0.0) -> None:
         """Reset the buffer."""
+
         self.data = [initial for _ in range(self.depth)]
+        self.elements = 0
 
     def resize(self, depth: int, initial: float = 0.0) -> None:
         """Set a new depth for this buffer average and reset the values."""

vcorelib/math/analysis/weighted.py

@@ -22,6 +22,11 @@ def depth(self) -> int:
         """This average's depth."""
         return self.signals.depth
 
+    @property
+    def saturated(self) -> bool:
+        """Determine if the buffer is saturated with elements yet."""
+        return self.signals.saturated and self.weights.saturated
+
     def reset(self) -> None:
         """Reset the average."""
         self.signals.reset()

vcorelib/math/keeper.py

@@ -0,0 +1,75 @@
+"""
+A module implementing a simple time-keeper interface.
+"""
+
+# built-in
+from contextlib import contextmanager
+from time import time_ns as _time_ns
+from typing import Callable, Iterator
+
+# internal
+from vcorelib.math.constants import to_nanos
+
+TimeSource = Callable[[], int]
+
+
+class SimulatedTime:
+    """A simple simulated-time interface."""
+
+    def __init__(self, step_dt_ns: int, start_ns: int = 0) -> None:
+        """Initialize this instance."""
+
+        self._start_ns = start_ns
+        self._step = 0
+
+        assert step_dt_ns >= 1, step_dt_ns
+        self._step_dt_ns = step_dt_ns
+
+    def step(self, count: int = 1) -> None:
+        """Step forward (or backwards) in time."""
+        self._step += count
+
+    def step_s(self, time_s: float) -> None:
+        """Step some number of seconds."""
+        self.step(to_nanos(time_s) // self._step_dt_ns)
+
+    def __call__(self) -> int:
+        """Get the current simulated time."""
+        return self._start_ns + (self._step * self._step_dt_ns)
+
+
+class TimeKeeper:
+    """A simple nanosecond time keeping interface."""
+
+    def __init__(self, source: TimeSource = _time_ns) -> None:
+        """Initialize this instance."""
+
+        self.source = source
+
+    @contextmanager
+    def simulated(
+        self, step_dt_ns: int = 1, start_ns: int = None
+    ) -> Iterator[SimulatedTime]:
+        """Take over time resolution with a simulated time instance."""
+
+        # Use a realistic starting timestamp value if one isn't provided.
+        if start_ns is None:
+            start_ns = self()
+
+        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
+
+    def __call__(self) -> int:
+        """Get time."""
+        return self.source()
+
+
+TIME = TimeKeeper()

vcorelib/math/time.py

@@ -10,7 +10,6 @@
 from logging import LoggerAdapter as _LoggerAdapter
 from math import floor as _floor
 from time import perf_counter_ns as _perf_counter_ns
-from time import time_ns as _time_ns
 from typing import Any as _Any
 from typing import Dict as _Dict
 from typing import Iterator as _Iterator
@@ -19,6 +18,8 @@
 
 # internal
 from vcorelib.math.constants import to_nanos
+from vcorelib.math.keeper import SimulatedTime as _SimulatedTime
+from vcorelib.math.keeper import TIME as _TIME
 from vcorelib.math.unit import KIBI_UNITS as _KIBI_UNITS
 from vcorelib.math.unit import SI_UNITS as _SI_UNITS
 from vcorelib.math.unit import UnitSystem as _UnitSystem
@@ -27,7 +28,19 @@
 
 def default_time_ns() -> int:
     """Get a timestamp value using a default method."""
-    return _time_ns()
+    return _TIME()
+
+
+@contextmanager
+def simulated_time(
+    step_dt_ns: int = 1, start_ns: int = None
+) -> _Iterator[_SimulatedTime]:
+    """Take control over the default time source as a managed context."""
+
+    with _TIME.simulated(
+        step_dt_ns=step_dt_ns, start_ns=start_ns
+    ) as simulated:
+        yield simulated
 
 
 def seconds_str(seconds: int) -> _Tuple[str, int]: