Rewrite the resampler class to be more reusable (#101)

First we introduce a timeseries `Source` (async iterator of `Sample`)
and `Sink` (an async function taking a `Sample`) types. The idea is for
all classes in the `timeseries` package to be based on these types.

On advantage of using a function as a `Sink` (instead of a channel) is
that for cases where we don't really need a channel (like if we only
need to store samples in a buffer), there is no added overhead (peak
shaving for example could benefit from this).

The `GroupResampler` and `Resampler` classes are not very reusable, or
at least require quite a bit of work from users to use them and they
need to take care of handling timers and calling methods at the right
time for them to be useful.

The new `Resampler` class has its own task(s) to handle all the
resampling, users only need to subscribe to new timeseries by passing a
timeseries `Source` and `Sink`. The `Sink` will be called each time a
new sample is produced.

A new `frequenz.sdk.timeseries.resampling` module is added because there
are more classes only used by the `Resampler` that might not be needed
by anyone just wanting to use `frequenz.sdk.timeseries`.

This PR also introduces the use of
[`async-solipsism`](https://pypi.org/project/async-solipsism/) (see #70)
for testing, so tests using timers can be mocked in a way that they
don't really need to wait for timers and time comparisons can be done
precisely. It works pretty well in combination with
[`time_machine`](https://pypi.org/project/time-machine/).

Fixes #56.

Author: llucax

RELEASE_NOTES.md

@@ -6,11 +6,28 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with --> 
+- `frequenz.sdk.timseries`:
+  - The resample classes in the `frequenz.sdk.timseries` were removed. Use the new `frequenz.sdk.timseries.resampling.Resampler` class instead.
+  - The `ResamplingFunction` was moved to the new module `frequenz.sdk.timseries.resampling`.
+
+- `frequenz.sdk.actor.ComponentMetricsResamplingActor`:
+  - The constructor now requires to pass all arguments as keywords.
+  - The following constructor arguments were renamed to make them more clear:
+    - `subscription_sender` -> `data_sourcing_request_sender`
+    - `subscription_receiver` -> `resampling_request_receiver`
+
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+- New in `frequenz.sdk.timeseries`:
+
+  - `Source` and `Sink` types to work generically with streaming timeseries.
+
+- New `frequenz.sdk.timeseries.resampling` module with:
+  - `Resampler` class that works on timseries `Source` and `Sink`.
+  - `ResamplingFunction` (moved from `frequenz.sdk.timeseries`).
+  - `ResamplingError` and `SourceStoppedError` exceptions.
+  - `average` function (the default resampling function).
 
 ## Bug Fixes
 

examples/resampling.py

@@ -4,9 +4,10 @@
 """Frequenz Python SDK resampling example."""
 
 import asyncio
+from datetime import datetime, timezone
 
 from frequenz.channels import Broadcast
-from frequenz.channels.util import MergeNamed
+from frequenz.channels.util import Merge
 
 from frequenz.sdk import microgrid
 from frequenz.sdk.actor import (
@@ -16,11 +17,29 @@
     DataSourcingActor,
 )
 from frequenz.sdk.microgrid.component import ComponentCategory, ComponentMetricId
+from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries._resampling import Resampler, Sink, Source
 
 HOST = "microgrid.sandbox.api.frequenz.io"
 PORT = 61060
 
 
+async def _calculate_average(source: Source, sink: Sink) -> None:
+    avg: float = 0.0
+    count: int = 0
+    async for sample in source:
+        print(f"Received sample to average at {sample.timestamp}: {sample.value}")
+        count += 1
+        if sample.value is None:
+            continue
+        avg = avg * (count - 1) / count + sample.value / count
+        await sink(Sample(datetime.now(timezone.utc), avg))
+
+
+async def _print_sample(sample: Sample) -> None:
+    print(f"\nResampled average at {sample.timestamp}: {sample.value}\n")
+
+
 async def run() -> None:  # pylint: disable=too-many-locals
     """Run main functions that initializes and creates everything."""
     await microgrid.initialize(HOST, PORT)
@@ -32,9 +51,9 @@ async def run() -> None:  # pylint: disable=too-many-locals
     data_source_request_sender = data_source_request_channel.new_sender()
     data_source_request_receiver = data_source_request_channel.new_receiver()
 
-    resampling_actor_request_channel = Broadcast[ComponentMetricRequest]("resample")
-    resampling_actor_request_sender = resampling_actor_request_channel.new_sender()
-    resampling_actor_request_receiver = resampling_actor_request_channel.new_receiver()
+    resampling_request_channel = Broadcast[ComponentMetricRequest]("resample")
+    resampling_request_sender = resampling_request_channel.new_sender()
+    resampling_request_receiver = resampling_request_channel.new_receiver()
 
     # Instantiate a data sourcing actor
     _data_sourcing_actor = DataSourcingActor(
@@ -44,9 +63,9 @@ async def run() -> None:  # pylint: disable=too-many-locals
     # Instantiate a resampling actor
     _resampling_actor = ComponentMetricsResamplingActor(
         channel_registry=channel_registry,
-        subscription_sender=data_source_request_sender,
-        subscription_receiver=resampling_actor_request_receiver,
-        resampling_period_s=1.0,
+        data_sourcing_request_sender=data_source_request_sender,
+        resampling_request_receiver=resampling_request_receiver,
+        resampling_period_s=1,
     )
 
     components = await microgrid.get().api_client.components()
@@ -56,7 +75,9 @@ async def run() -> None:  # pylint: disable=too-many-locals
         if comp.category == ComponentCategory.BATTERY
     ]
 
-    # Create subscription requests for each time series id
+    print(f"Found {len(battery_ids)} batteries: {battery_ids}")
+
+    # Create subscription requests for each battery's SoC
     subscription_requests = [
         ComponentMetricRequest(
             namespace="resampling",
@@ -69,24 +90,41 @@ async def run() -> None:  # pylint: disable=too-many-locals
 
     # Send the subscription requests
     await asyncio.gather(
-        *[
-            resampling_actor_request_sender.send(request)
-            for request in subscription_requests
-        ]
+        *[resampling_request_sender.send(request) for request in subscription_requests]
     )
 
     # Merge sample receivers for each subscription into one receiver
-    merged_receiver = MergeNamed(
-        **{
-            req.get_channel_name(): channel_registry.new_receiver(
-                req.get_channel_name()
-            )
+    merged_receiver = Merge(
+        *[
+            channel_registry.new_receiver(req.get_channel_name())
             for req in subscription_requests
-        }
+        ]
     )
 
-    async for channel_name, msg in merged_receiver:
-        print(f"{channel_name}: {msg}")
+    # Create a channel to calculate an average for all the data
+    average_chan = Broadcast[Sample]("average")
+
+    second_stage_resampler = Resampler(resampling_period_s=3.0)
+    second_stage_resampler.add_timeseries(average_chan.new_receiver(), _print_sample)
+
+    average_sender = average_chan.new_sender()
+    # Needed until channels Senders raises exceptions on errors
+    async def sink_adapter(sample: Sample) -> None:
+        assert await average_sender.send(sample)
+
+    print("Starting...")
+
+    try:
+        # This will run until it is interrupted (with Ctrl-C for example)
+        await asyncio.gather(
+            _calculate_average(merged_receiver, sink_adapter),
+            second_stage_resampler.resample(),
+        )
+    finally:
+        await second_stage_resampler.stop()
 
 
-asyncio.run(run())
+try:
+    asyncio.run(run())
+except KeyboardInterrupt:
+    print("Bye!")

noxfile.py

@@ -13,7 +13,14 @@
 
 FMT_DEPS = ["black", "isort"]
 DOCSTRING_DEPS = ["pydocstyle", "darglint"]
-PYTEST_DEPS = ["pytest", "pytest-cov", "pytest-mock", "pytest-asyncio", "time-machine"]
+PYTEST_DEPS = [
+    "pytest",
+    "pytest-cov",
+    "pytest-mock",
+    "pytest-asyncio",
+    "time-machine",
+    "async-solipsism",
+]
 MYPY_DEPS = ["mypy", "pandas-stubs", "grpc-stubs"]
 
 

pyproject.toml

@@ -100,10 +100,13 @@ src_paths = ["src", "examples", "tests"]
 asyncio_mode = "auto"
 required_plugins = [ "pytest-asyncio", "pytest-mock" ]
 
-[[tools.mypy.overrides]]
 [[tool.mypy.overrides]]
 module = [
     "grpc.aio",
     "grpc.aio.*"
 ]
 ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "async_solipsism"
+ignore_missing_imports = true