Make resampler infer the input sampling period and buffer length (#136)

The input sampling period is calculated by counting the total received
samples, and dividing the total resampling time until the internal
buffer is full by the total received samples.

To store all the source properties (sampling period, sampling start,
total received samples) a new class is used (SourceProperties) and this
information is passed to the resampling function, so it can have more
information about the source to perform a proper resampling.

This calculation is performed once and then remains constant, but at
some point we could add a timer to re-calculate it.

This also improves slightly the documentation and validation of the
ResamplingConfig class.

The add_timeseries() methods now also takes a name as a way to identify
different sources in log messages (the actor uses the channel name as
timeseries name).

We also update the resampling buffer length based on the input sampling
period, resampling period, and max data age in periods, so it can hold
all the past data without wasting any space.

Fixes #55.

Author: llucax

RELEASE_NOTES.md

@@ -6,12 +6,35 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with --> 
+* The resampler now takes a `name` argument for `add_timeseries()`. This is only used for logging purposes.
+
+* The resampler and resampling actor now takes a `ResamplerConfig` object in the constructor instead of the individual values.
+
+* The resampler and resampling actor can emit errors or warnings if the buffer needed to resample is too big. If it is bigger than `ResamplingConfig.max_buffer_len`, the buffer will be truncated to that length, so the resampling can lose accuracy.
+
+* The `ResamplingFunction` now takes different arguments:
+
+  * `resampling_period_s` was removed.
+  * `resampler_config` is the configuration of the resampler calling the resampling function.
+  * `source_properties` is the properties of the source being resampled.
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* The resampler and resampling actor can now take a few new options via the new `ResamplerConfig` object:
+
+  * `warn_buffer_len`: The minimum length of the resampling buffer that will emit a warning.
+  * `max_buffer_len`: The maximum length of the resampling buffer.
+
+* The resampler now infers the input sampling rate of sources and use a buffer size according to it.
+
+  This information can be consulted via `resampler.get_source_properties(source)`. The buffer size is now calculated so it can store all the needed samples requested via the combination of `resampling_period_s`, `max_data_age_in_periods` and the calculated `input_sampling_period_s`.
+
+  If we are upsampling, one sample could be enough for back-filling, but we store `max_data_age_in_periods` using `input_sampling_period_s` as period, so the resampling functions can do more complex inter/extrapolation if they need to.
+
+  If we are downsampling, we want a buffer that can hold `max_data_age_in_periods * resampling_period_s` seconds of data, and we have one sample every `input_sampling_period_s`, so we use a buffer length of: `max_data_age_in_periods * resampling_period_s / input_sampling_period_s`
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+* Fixed logger creationg for some modules.
+
+  Some modules didn't create the logger properly so there was no way to configure them using the standard logger configuration system. Because of this, it might have happened that some log messages were never showed, or some message that the user didn't want to get were emitted anyway.

benchmarks/timeseries/resampling.py

@@ -8,11 +8,17 @@
 from typing import Sequence
 
 from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries._resampling import ResamplerConfig, _ResamplingHelper
+from frequenz.sdk.timeseries._resampling import (
+    ResamplerConfig,
+    SourceProperties,
+    _ResamplingHelper,
+)
 
 
 def nop(  # pylint: disable=unused-argument
-    samples: Sequence[Sample], resampling_period_s: float
+    samples: Sequence[Sample],
+    resampler_config: ResamplerConfig,
+    source_properties: SourceProperties,
 ) -> float:
     """Return 0.0."""
     return 0.0
@@ -21,12 +27,15 @@ def nop(  # pylint: disable=unused-argument
 def _benchmark_resampling_helper(resamples: int, samples: int) -> None:
     """Benchmark the resampling helper."""
     helper = _ResamplingHelper(
+        "benchmark",
         ResamplerConfig(
             resampling_period_s=1.0,
             max_data_age_in_periods=3.0,
             resampling_function=nop,
             initial_buffer_len=samples * 3,
-        )
+            warn_buffer_len=samples * 3 + 2,
+            max_buffer_len=samples * 3 + 3,
+        ),
     )
     now = datetime.now(timezone.utc)
 

examples/resampling.py

@@ -105,7 +105,9 @@ async def run() -> None:  # pylint: disable=too-many-locals
     average_chan = Broadcast[Sample]("average")
 
     second_stage_resampler = Resampler(ResamplerConfig(resampling_period_s=3.0))
-    second_stage_resampler.add_timeseries(average_chan.new_receiver(), _print_sample)
+    second_stage_resampler.add_timeseries(
+        "avg", average_chan.new_receiver(), _print_sample
+    )
 
     average_sender = average_chan.new_sender()
     # Needed until channels Senders raises exceptions on errors

src/frequenz/sdk/_data_handling/time_series.py

@@ -15,7 +15,7 @@
 Key = TypeVar("Key")
 Value = TypeVar("Value")
 
-logger = logging.Logger(__name__)
+logger = logging.getLogger(__name__)
 
 SYMBOL_SEGMENT_SEPARATOR = "_"
 

src/frequenz/sdk/_data_ingestion/component_info.py

@@ -10,7 +10,7 @@
 from ..microgrid import ComponentGraph
 from ..microgrid.component import ComponentCategory
 
-logger = logging.Logger(__name__)
+logger = logging.getLogger(__name__)
 
 
 @dataclass

src/frequenz/sdk/_data_ingestion/formula_calculator.py

@@ -36,7 +36,7 @@
     METRIC_PV_PROD,
 )
 
-logger = logging.Logger(__name__)
+logger = logging.getLogger(__name__)
 
 
 @dataclass(frozen=True)

src/frequenz/sdk/_data_ingestion/load_historic_data.py

@@ -22,7 +22,7 @@
 import pyarrow.parquet as pq
 from tqdm import tqdm
 
-logger = logging.Logger(__name__)
+logger = logging.getLogger(__name__)
 
 # directory path to all component data of a particular site
 HISTDATA_DIR = "/data"

src/frequenz/sdk/_data_ingestion/microgrid_data.py

@@ -27,7 +27,7 @@
 from .formula_calculator import FormulaCalculator
 from .gen_component_receivers import gen_component_receivers
 
-logger = logging.Logger(__name__)
+logger = logging.getLogger(__name__)
 
 CONFIG_FILE_FORMULA_PREFIX = "formula_"
 

src/frequenz/sdk/actor/_decorator.py

@@ -14,7 +14,7 @@
 import logging
 from typing import Any, Generic, Optional, Type, TypeVar
 
-logger = logging.Logger(__name__)
+logger = logging.getLogger(__name__)
 
 OT = TypeVar("OT")
 

src/frequenz/sdk/actor/_resampling.py

@@ -19,7 +19,7 @@
 from ._data_sourcing import ComponentMetricRequest
 from ._decorator import actor
 
-logger = logging.Logger(__name__)
+logger = logging.getLogger(__name__)
 
 
 @actor
@@ -85,7 +85,7 @@ async def sink_adapter(sample: Sample) -> None:
             if not await sender.send(sample):
                 raise Exception(f"Error while sending with sender {sender}", sender)
 
-        self._resampler.add_timeseries(receiver, sink_adapter)
+        self._resampler.add_timeseries(request_channel_name, receiver, sink_adapter)
 
     async def _process_resampling_requests(self) -> None:
         """Process resampling data requests."""