Move resampler classes to frequenz.sdk.timeseries.resampler

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/sdk/actor/resampling.py

@@ -15,11 +15,8 @@
 
 from frequenz.channels import MergeNamed, Receiver, Select, Sender, Timer
 
-from ..data_ingestion.resampling.component_metric_group_resampler import (
-    ComponentMetricGroupResampler,
-)
-from ..data_ingestion.resampling.component_metric_resampler import ResamplingFunction
 from ..data_pipeline import ComponentMetricRequest, Sample
+from ..timeseries.resampler import ComponentMetricGroupResampler, ResamplingFunction
 from . import ChannelRegistry, actor
 
 logger = logging.Logger(__name__)

src/frequenz/sdk/data_ingestion/resampling/component_metric_resampler.py

@@ -0,0 +1,26 @@
+"""
+Handling of timeseries streams.
+
+A timeseries is a stream (normally an async iterator) of
+[samples][frequenz.sdk.timeseries.Sample].
+
+This module provides tools to operate on timeseries.
+
+Copyright
+Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+License
+MIT
+"""
+
+from .resampler import (
+    ComponentMetricGroupResampler,
+    ComponentMetricResampler,
+    ResamplingFunction,
+)
+
+__all__ = [
+    "ComponentMetricGroupResampler",
+    "ComponentMetricResampler",
+    "ResamplingFunction",
+]

src/frequenz/sdk/timeseries/__init__.py

@@ -1,24 +1,103 @@
 """
-ComponentMetricGroupResampler class that delegates resampling to individual resamplers.
+Timeseries resampler.
 
 Copyright
 Copyright © 2022 Frequenz Energy-as-a-Service GmbH
 
 License
 MIT
 """
+
 import logging
-from datetime import datetime
-from typing import Dict, Generator, Tuple
+from collections import deque
+from datetime import datetime, timedelta
+from typing import Callable, Deque, Dict, Generator, Optional, Sequence, Tuple
 
 import pytz
 
-from ...data_pipeline import Sample
-from .component_metric_resampler import ComponentMetricResampler, ResamplingFunction
+from ..data_pipeline import Sample
 
 logger = logging.Logger(__name__)
 
 
+ResamplingFunction = Callable[[Sequence[Sample], float], float]
+
+
+class ComponentMetricResampler:
+    """Resampler for a single metric of a specific component, e.g. 123_active_power."""
+
+    def __init__(
+        self,
+        resampling_period_s: float,
+        max_data_age_in_periods: float,
+        resampling_function: ResamplingFunction,
+    ) -> None:
+        """Initialize the ComponentMetricResampler.
+
+        Args:
+            resampling_period_s: value describing how often resampling should be
+                performed, in seconds
+            max_data_age_in_periods: max age that samples shouldn't exceed in order
+                to be used in the resampling function
+            resampling_function: function to be applied to a sequence of samples within
+                a resampling period to produce a single output sample
+        """
+        self._resampling_period_s = resampling_period_s
+        self._max_data_age_in_periods: float = max_data_age_in_periods
+        self._buffer: Deque[Sample] = deque()
+        self._resampling_function: ResamplingFunction = resampling_function
+
+    def add_sample(self, sample: Sample) -> None:
+        """Add a new sample.
+
+        Args:
+            sample: sample to be added to the buffer
+        """
+        self._buffer.append(sample)
+
+    def remove_outdated_samples(self, threshold: datetime) -> None:
+        """Remove samples that are older than the provided time threshold.
+
+        It is assumed that items in the buffer are in a sorted order (ascending order
+        by timestamp).
+
+        The removal works by traversing the buffer starting from the oldest sample
+        (smallest timestamp) and comparing sample's timestamp with the threshold.
+        If the sample's threshold is smaller than `threshold`, it means that the
+        sample is outdated and it is removed from the buffer. This continues until
+        the first sample that is with timestamp greater or equal to `threshold` is
+        encountered, then buffer is considered up to date.
+
+        Args:
+            threshold: samples whose timestamp is older than the threshold are
+                considered outdated and should be remove from the buffer
+        """
+        while self._buffer:
+            sample: Sample = self._buffer[0]
+            if sample.timestamp >= threshold:
+                return
+
+            self._buffer.popleft()
+
+    def resample(self) -> Optional[float]:
+        """Resample samples from the buffer and produce a single sample.
+
+        Returns:
+            Samples resampled into a single sample or `None` if the
+                `resampling_function` cannot produce a valid Sample.
+        """
+        # It might be better to provide `now` from the outside so that all
+        # individual resamplers use the same `now`
+        now = datetime.now(tz=pytz.UTC)
+        threshold = now - timedelta(
+            seconds=self._max_data_age_in_periods * self._resampling_period_s
+        )
+        self.remove_outdated_samples(threshold=threshold)
+        if len(self._buffer) == 0:
+            return None
+        return self._resampling_function(self._buffer, self._resampling_period_s)
+
+
 class ComponentMetricGroupResampler:
     """Class that delegates resampling to individual component metric resamplers."""
 

…ling/component_metric_group_resampler.py src/frequenz/sdk/timeseries/resampler.pysrc/frequenz/sdk/data_ingestion/resampling/component_metric_group_resampler.py renamed to src/frequenz/sdk/timeseries/resampler.py src/frequenz/sdk/data_ingestion/resampling/component_metric_group_resampler.py renamed to src/frequenz/sdk/timeseries/resampler.py

@@ -7,16 +7,18 @@
 License
 MIT
 """
+
 from datetime import datetime, timedelta
 from typing import Sequence
 
 import pytz
 import time_machine
 
-from frequenz.sdk.data_ingestion.resampling.component_metric_resampler import (
+from frequenz.sdk.data_pipeline import Sample
+from frequenz.sdk.timeseries.resampler import (
+    ComponentMetricGroupResampler,
     ComponentMetricResampler,
 )
-from frequenz.sdk.data_pipeline import Sample
 
 
 # pylint: disable=unused-argument
@@ -113,3 +115,44 @@ def test_component_metric_resampler_resample_with_outdated_samples() -> None:
     value = resampler.resample()
     assert value is not None
     assert value == sum([value1, value2])
+
+
+@time_machine.travel(datetime.now())
+def test_component_metric_group_resampler() -> None:
+    """Test if resampling is properly delegated to component metric resamplers."""
+    resampler = ComponentMetricGroupResampler(
+        resampling_period_s=0.2,
+        max_data_age_in_periods=5.0,
+        initial_resampling_function=resampling_function_sum,
+    )
+
+    time_series_id_1 = "123_active_power"
+    time_series_id_2 = "99_active_power"
+
+    resampler.add_time_series(time_series_id=time_series_id_1)
+    resampler.add_time_series(time_series_id=time_series_id_2)
+
+    timestamp = datetime.now(tz=pytz.UTC)
+
+    value1 = 5.0
+    value2 = 15.0
+    value3 = 100.0
+    value4 = 999.0
+
+    sample1 = Sample(timestamp - timedelta(seconds=0.5), value=value1)
+    sample2 = Sample(timestamp - timedelta(seconds=0.7), value=value2)
+    sample3 = Sample(timestamp - timedelta(seconds=5.05), value=value3)
+    sample4 = Sample(timestamp - timedelta(seconds=0.99), value=value4)
+
+    resampler.add_sample(time_series_id=time_series_id_1, sample=sample1)
+    resampler.add_sample(time_series_id=time_series_id_1, sample=sample2)
+    resampler.add_sample(time_series_id=time_series_id_2, sample=sample3)
+    resampler.add_sample(time_series_id=time_series_id_2, sample=sample4)
+
+    resampled_samples = dict(resampler.resample())
+
+    assert resampled_samples[time_series_id_1].timestamp >= timestamp
+    assert resampled_samples[time_series_id_1].value == sum([value1, value2])
+
+    assert resampled_samples[time_series_id_2].timestamp >= timestamp
+    assert resampled_samples[time_series_id_2].value == value4