Split the resampling module

Future changes require adding more code to this module, which is already
quite big. This commit makes it a package and split the module into a
few sub-modules to make it more manageable.

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

Author: llucax

benchmarks/timeseries/resampling.py

@@ -9,12 +9,9 @@
 
 from frequenz.quantities import Quantity
 
-from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries._resampling import (
-    ResamplerConfig,
-    SourceProperties,
-    _ResamplingHelper,
-)
+from frequenz.sdk.timeseries import ResamplerConfig, Sample
+from frequenz.sdk.timeseries._resampling._base_types import SourceProperties
+from frequenz.sdk.timeseries._resampling._resampler import _ResamplingHelper
 
 
 def nop(  # pylint: disable=unused-argument

src/frequenz/sdk/actor/__init__.py

@@ -597,7 +597,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 ._run_utils import run

src/frequenz/sdk/microgrid/_resampling.py

@@ -13,9 +13,11 @@
 
 from .._internal._asyncio import cancel_and_await
 from .._internal._channels import ChannelRegistry
-from ..actor import Actor
+from ..actor._actor import Actor
 from ..timeseries import Sample
-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 ._data_sourcing import ComponentMetricRequest
 
 _logger = logging.getLogger(__name__)

src/frequenz/sdk/timeseries/__init__.py

@@ -40,7 +40,7 @@
 from ._fuse import Fuse
 from ._moving_window import MovingWindow
 from ._periodic_feature_extractor import PeriodicFeatureExtractor
-from ._resampling import ResamplerConfig
+from ._resampling._config import ResamplerConfig
 
 __all__ = [
     "Bounds",

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 ._resampling import Resampler, ResamplerConfig
+from ._resampling._config import ResamplerConfig
+from ._resampling._resampler import Resampler
 from ._ringbuffer import OrderedRingBuffer
 
 _logger = logging.getLogger(__name__)

src/frequenz/sdk/timeseries/_resampling/__init__.py

@@ -0,0 +1,4 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Timeseries resampling package."""

src/frequenz/sdk/timeseries/_resampling/_base_types.py

@@ -0,0 +1,58 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Resampler base types."""
+
+from collections.abc import AsyncIterator, Callable, Coroutine
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+from frequenz.quantities import Quantity
+
+from .._base_types import Sample
+
+Source = AsyncIterator[Sample[Quantity]]
+"""A source for a timeseries.
+
+A timeseries can be received sample by sample in a streaming way
+using a source.
+"""
+
+Sink = Callable[[Sample[Quantity]], Coroutine[None, None, None]]
+"""A sink for a timeseries.
+
+A new timeseries can be generated by sending samples to a sink.
+
+This should be an `async` callable, for example:
+
+```python
+async some_sink(Sample) -> None:
+    ...
+```
+
+Args:
+    sample (Sample): A sample to be sent out.
+"""
+
+
+@dataclass
+class SourceProperties:
+    """Properties of a resampling source."""
+
+    sampling_start: datetime | None = None
+    """The time when resampling started for this source.
+
+    `None` means it didn't started yet.
+    """
+
+    received_samples: int = 0
+    """Total samples received by this source so far."""
+
+    sampling_period: timedelta | None = None
+    """The sampling period of this source.
+
+    This means we receive (on average) one sample for this source every
+    `sampling_period` time.
+
+    `None` means it is unknown.
+    """

src/frequenz/sdk/timeseries/_resampling/_config.py

@@ -0,0 +1,223 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Resampler configuration."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+from frequenz.quantities import Quantity
+
+from .._base_types import UNIX_EPOCH, QuantityT, Sample
+from ._base_types import SourceProperties
+
+_logger = logging.getLogger(__name__)
+
+
+DEFAULT_BUFFER_LEN_INIT = 16
+"""Default initial buffer length.
+
+Buffers will be created initially with this length, but they could grow or
+shrink depending on the source properties, like sampling rate, to make
+sure all the requested past sampling periods can be stored.
+"""
+
+
+DEFAULT_BUFFER_LEN_MAX = 1024
+"""Default maximum allowed buffer length.
+
+If a buffer length would get bigger than this, it will be truncated to this
+length.
+"""
+
+
+DEFAULT_BUFFER_LEN_WARN = 128
+"""Default minimum buffer length that will produce a warning.
+
+If a buffer length would get bigger than this, a warning will be logged.
+"""
+
+
+ResamplingFunction = Callable[
+    [Sequence[Sample[Quantity]], "ResamplerConfig", "SourceProperties"], float
+]
+"""Resampling function type.
+
+A resampling function produces a new sample based on a list of pre-existing
+samples. It can do "upsampling" when the data rate of the `input_samples`
+period is smaller than the `resampling_period`, or "downsampling" if it is
+bigger.
+
+In general a resampling window is the same as the `resampling_period`, and
+this function might receive input samples from multiple windows in the past to
+enable extrapolation, but no samples from the future (so the timestamp of the
+new sample that is going to be produced will always be bigger than the biggest
+timestamp in the input data).
+
+Args:
+    input_samples (Sequence[Sample]): The sequence of pre-existing samples.
+    resampler_config (ResamplerConfig): The configuration of the resampling
+        calling this function.
+    source_properties (SourceProperties): The properties of the source being
+        resampled.
+
+Returns:
+    new_sample (float): The value of new sample produced after the resampling.
+"""
+
+
+# pylint: disable=unused-argument
+def average(
+    samples: Sequence[Sample[QuantityT]],
+    resampler_config: ResamplerConfig,
+    source_properties: SourceProperties,
+) -> float:
+    """Calculate average of all the provided values.
+
+    Args:
+        samples: The samples to apply the average to. It must be non-empty.
+        resampler_config: The configuration of the resampler calling this
+            function.
+        source_properties: The properties of the source being resampled.
+
+    Returns:
+        The average of all `samples` values.
+    """
+    assert len(samples) > 0, "Average cannot be given an empty list of samples"
+    values = list(
+        sample.value.base_value for sample in samples if sample.value is not None
+    )
+    return sum(values) / len(values)
+
+
+@dataclass(frozen=True)
+class ResamplerConfig:
+    """Resampler configuration."""
+
+    resampling_period: timedelta
+    """The resampling period.
+
+    This is the time it passes between resampled data should be calculated.
+
+    It must be a positive time span.
+    """
+
+    max_data_age_in_periods: float = 3.0
+    """The maximum age a sample can have to be considered *relevant* for resampling.
+
+    Expressed in number of periods, where period is the `resampling_period`
+    if we are downsampling (resampling period bigger than the input period) or
+    the *input sampling period* if we are upsampling (input period bigger than
+    the resampling period).
+
+    It must be bigger than 1.0.
+
+    Example:
+        If `resampling_period` is 3 seconds, the input sampling period is
+        1 and `max_data_age_in_periods` is 2, then data older than 3*2
+        = 6 seconds will be discarded when creating a new sample and never
+        passed to the resampling function.
+
+        If `resampling_period` is 3 seconds, the input sampling period is
+        5 and `max_data_age_in_periods` is 2, then data older than 5*2
+        = 10 seconds will be discarded when creating a new sample and never
+        passed to the resampling function.
+    """
+
+    resampling_function: ResamplingFunction = average
+    """The resampling function.
+
+    This function will be applied to the sequence of relevant samples at
+    a given time. The result of the function is what is sent as the resampled
+    value.
+    """
+
+    initial_buffer_len: int = DEFAULT_BUFFER_LEN_INIT
+    """The initial length of the resampling buffer.
+
+    The buffer could grow or shrink depending on the source properties,
+    like sampling rate, to make sure all the requested past sampling periods
+    can be stored.
+
+    It must be at least 1 and at most `max_buffer_len`.
+    """
+
+    warn_buffer_len: int = DEFAULT_BUFFER_LEN_WARN
+    """The minimum length of the resampling buffer that will emit a warning.
+
+    If a buffer grows bigger than this value, it will emit a warning in the
+    logs, so buffers don't grow too big inadvertently.
+
+    It must be at least 1 and at most `max_buffer_len`.
+    """
+
+    max_buffer_len: int = DEFAULT_BUFFER_LEN_MAX
+    """The maximum length of the resampling buffer.
+
+    Buffers won't be allowed to grow beyond this point even if it would be
+    needed to keep all the requested past sampling periods. An error will be
+    emitted in the logs if the buffer length needs to be truncated to this
+    value.
+
+    It must be at bigger than `warn_buffer_len`.
+    """
+
+    align_to: datetime | None = UNIX_EPOCH
+    """The time to align the resampling period to.
+
+    The resampling period will be aligned to this time, so the first resampled
+    sample will be at the first multiple of `resampling_period` starting from
+    `align_to`. It must be an aware datetime and can be in the future too.
+
+    If `align_to` is `None`, the resampling period will be aligned to the
+    time the resampler is created.
+    """
+
+    def __post_init__(self) -> None:
+        """Check that config values are valid.
+
+        Raises:
+            ValueError: If any value is out of range.
+        """
+        if self.resampling_period.total_seconds() < 0.0:
+            raise ValueError(
+                f"resampling_period ({self.resampling_period}) must be positive"
+            )
+        if self.max_data_age_in_periods < 1.0:
+            raise ValueError(
+                f"max_data_age_in_periods ({self.max_data_age_in_periods}) should be at least 1.0"
+            )
+        if self.warn_buffer_len < 1:
+            raise ValueError(
+                f"warn_buffer_len ({self.warn_buffer_len}) should be at least 1"
+            )
+        if self.max_buffer_len <= self.warn_buffer_len:
+            raise ValueError(
+                f"max_buffer_len ({self.max_buffer_len}) should "
+                f"be bigger than warn_buffer_len ({self.warn_buffer_len})"
+            )
+
+        if self.initial_buffer_len < 1:
+            raise ValueError(
+                f"initial_buffer_len ({self.initial_buffer_len}) should at least 1"
+            )
+        if self.initial_buffer_len > self.max_buffer_len:
+            raise ValueError(
+                f"initial_buffer_len ({self.initial_buffer_len}) is bigger "
+                f"than max_buffer_len ({self.max_buffer_len}), use a smaller "
+                "initial_buffer_len or a bigger max_buffer_len"
+            )
+        if self.initial_buffer_len > self.warn_buffer_len:
+            _logger.warning(
+                "initial_buffer_len (%s) is bigger than warn_buffer_len (%s)",
+                self.initial_buffer_len,
+                self.warn_buffer_len,
+            )
+        if self.align_to is not None and self.align_to.tzinfo is None:
+            raise ValueError(
+                f"align_to ({self.align_to}) should be a timezone aware datetime"
+            )