Resync the resampler on system time changes (take 2)

RELEASE_NOTES.md

@@ -28,3 +28,4 @@
 - Fixed a typing issue that occurs in some cases when composing formulas with constants.
 - Fixed a bug where sending tasks in the data sourcing actor might have not been properly awaited.
 - Updated the logical meter documentation to reflect the latest changes.
+- The resampler will now resync to the system time if it drifts away for more than a resample period.

benchmarks/timeseries/resampling.py

@@ -7,17 +7,14 @@
 from datetime import datetime, timedelta, timezone
 from timeit import timeit
 
-from frequenz.sdk.timeseries import Sample
 from frequenz.sdk.timeseries._quantities import Quantity
-from frequenz.sdk.timeseries._resampling import (
-    ResamplerConfig,
-    SourceProperties,
-    _ResamplingHelper,
-)
+from frequenz.sdk.timeseries._resampling._base_types import SourceProperties
+from frequenz.sdk.timeseries._resampling._config import ResamplerConfig
+from frequenz.sdk.timeseries._resampling._resampler import _ResamplingHelper
 
 
 def nop(  # pylint: disable=unused-argument
-    samples: Sequence[Sample[Quantity]],
+    samples: Sequence[tuple[datetime, Quantity]],
     resampler_config: ResamplerConfig,
     source_properties: SourceProperties,
 ) -> float:
@@ -42,10 +39,12 @@ def _benchmark_resampling_helper(resamples: int, samples: int) -> None:
 
     def _do_work() -> None:
         nonlocal now
+        zero = Quantity.zero()
+        delta = timedelta(seconds=1 / samples)
         for _n_resample in range(resamples):
             for _n_sample in range(samples):
-                now = now + timedelta(seconds=1 / samples)
-                helper.add_sample(Sample(now, Quantity(0.0)))
+                now = now + delta
+                helper.add_sample((now, zero))
             helper.resample(now)
 
     print(timeit(_do_work, number=5))

src/frequenz/sdk/actor/__init__.py

@@ -593,7 +593,7 @@ async def main() -> None:  # (6)!
 [_run]: #the-_run-method
 """
 
-from ..timeseries._resampling import ResamplerConfig
+from ..timeseries._resampling._config import ResamplerConfig
 from ._actor import Actor
 from ._background_service import BackgroundService
 from ._channel_registry import ChannelRegistry

src/frequenz/sdk/actor/_resampling.py

@@ -13,7 +13,9 @@
 from .._internal._asyncio import cancel_and_await
 from ..timeseries import Sample
 from ..timeseries._quantities import Quantity
-from ..timeseries._resampling import Resampler, ResamplerConfig, ResamplingError
+from ..timeseries._resampling._config import ResamplerConfig
+from ..timeseries._resampling._exceptions import ResamplingError
+from ..timeseries._resampling._resampler import Resampler
 from ._actor import Actor
 from ._channel_registry import ChannelRegistry
 from ._data_sourcing import ComponentMetricRequest

src/frequenz/sdk/timeseries/__init__.py

@@ -50,7 +50,10 @@
     Temperature,
     Voltage,
 )
-from ._resampling import ResamplerConfig
+from ._resampling._base_types import SourceProperties
+from ._resampling._config import ResamplerConfig, ResamplingFunction
+from ._resampling._exceptions import ResamplingError, SourceStoppedError
+from ._resampling._wall_clock_timer import WallClockTimerConfig
 
 __all__ = [
     "Bounds",
@@ -59,9 +62,14 @@
     "PeriodicFeatureExtractor",
     "ResamplerConfig",
     "ReceiverFetcher",
+    "ResamplingError",
+    "ResamplingFunction",
     "Sample",
     "Sample3Phase",
+    "SourceProperties",
+    "SourceStoppedError",
     "UNIX_EPOCH",
+    "WallClockTimerConfig",
     #
     # Quantities
     #

src/frequenz/sdk/timeseries/_moving_window.py

@@ -18,7 +18,8 @@
 from ..actor._background_service import BackgroundService
 from ._base_types import UNIX_EPOCH, Sample
 from ._quantities import Quantity
-from ._resampling import Resampler, ResamplerConfig
+from ._resampling._config import ResamplerConfig
+from ._resampling._resampler import Resampler
 from ._ringbuffer import OrderedRingBuffer
 
 _logger = logging.getLogger(__name__)