1.1.5 - Add a simple rate-limiter

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.0
-    hash=ae596ec2ce311b8007a08cc354b001dc
+    hash=1142fb532ef7d79ca762297729cf2967
     =====================================
 -->
 
-# vcorelib ([1.1.4](https://pypi.org/project/vcorelib/))
+# vcorelib ([1.1.5](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: 1
 minor: 1
-patch: 4
+patch: 5

pyproject.toml

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

tests/math/analysis/rate/test_limiter.py

@@ -0,0 +1,30 @@
+"""
+Test the 'math.analysis.rate.limiter' module.
+"""
+
+# third-party
+from pytest import approx
+
+# module under test
+from vcorelib.math.analysis.rate.limiter import RateLimiter
+
+
+def test_rate_limiter_basic():
+    """Test that the rate-limiter works correctly."""
+
+    second = int(1e9)
+
+    lim = RateLimiter(second)
+
+    curr = second
+    lim(curr)
+
+    for _ in range(100):
+        curr += second
+        assert lim(curr, lambda: None)
+
+    assert not lim(curr)
+
+    assert lim.rate_hz == approx(1.0)
+
+    assert lim()

vcorelib/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.0
-# hash=82c08388314d339495716f8a42d459dd
+# hash=9c52383580fc08f2f460dfee564cbf32
 # =====================================
 
 """
@@ -10,7 +10,7 @@
 
 DESCRIPTION = "A collection of core Python utilities."
 PKG_NAME = "vcorelib"
-VERSION = "1.1.4"
+VERSION = "1.1.5"
 
 # vcorelib-specific content.
 DEFAULT_INDENT = 2

vcorelib/math/analysis/rate/limiter.py

@@ -0,0 +1,58 @@
+"""
+A module for simplifying limiting the rate of a function call or hot loop.
+"""
+
+# built-in
+from typing import Callable as _Callable
+
+# internal
+from vcorelib.math.analysis.rate import RateTracker
+from vcorelib.math.time import default_time_ns as _default_time_ns
+
+
+class RateLimiter:
+    """A class for limiting the rate of runtime work."""
+
+    def __init__(self, period_ns: int, **kwargs) -> None:
+        """Initialize this rate-limiter."""
+
+        assert period_ns >= 0
+        self.period_ns = period_ns
+        self.prev_time_ns: int = 0
+        self.rate = RateTracker(**kwargs)
+
+    @property
+    def rate_hz(self) -> float:
+        """
+        Get the underlying rate that this limiter is governing. This is useful
+        to determine if the rate limitation is or isn't impacting the rate of
+        some task.
+        """
+        return self.rate.value
+
+    def __call__(
+        self, time_ns: int = None, task: _Callable[[], None] = None
+    ) -> bool:
+        """
+        Query the limiter to determine if the current time would allow a
+        governed task to run.
+        """
+
+        result = False
+
+        # Use a default time if one wasn't provided.
+        if time_ns is None:
+            time_ns = _default_time_ns()
+
+        if time_ns >= self.prev_time_ns + self.period_ns:
+            self.prev_time_ns = time_ns
+            result = True
+
+            # Call the task if provided.
+            if task is not None:
+                task()
+
+            # Update rate tracking.
+            self.rate(time_ns=time_ns)
+
+        return result