Added a simple example for using TimeSeriesResamplingActor

Signed-off-by: Jakub Kozlowicz <[email protected]>

Author: jkozlowicz

examples/sdk_resampling_example.py

@@ -0,0 +1,146 @@
+"""Frequenz Python SDK resampling example.
+
+Copyright
+Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+License
+MIT
+"""
+
+import asyncio
+import random
+from datetime import datetime
+from typing import Dict
+
+import pytz
+from frequenz.channels import Broadcast, MergeNamed, Receiver, Select, Sender
+
+from frequenz.sdk.core.timeseries import TimeSeriesId
+from frequenz.sdk.data_ingestion.resampling.component_metric_resampler import Sample
+from frequenz.sdk.data_ingestion.resampling.data_provider import DataProvider
+from frequenz.sdk.data_ingestion.resampling.subscriptions import (
+    Subscription,
+    SubscriptionRequest,
+    SubscriptionResponse,
+)
+from frequenz.sdk.data_ingestion.resampling.timeseries_resampling_actor import (
+    TimeSeriesResamplingActor,
+)
+
+
+async def data_sender(sender: Sender[Sample]) -> None:
+    """Send a message to the channel every 0.2 seconds.
+
+    Args:
+        sender: channel to be used to send random component data
+    """
+    while True:
+        value = random.uniform(1.0, 100.0)
+        await sender.send(Sample(timestamp=datetime.now(tz=pytz.UTC), value=value))
+        await asyncio.sleep(random.uniform(0.1, 0.5))
+
+
+class DummyDataProvider(DataProvider):
+    """Dummy Data Provider for presentation purposes."""
+
+    def __init__(self) -> None:
+        """Create an instance and initialize receivers."""
+        self._receivers: Dict[Subscription, Receiver[Sample]] = {}
+
+    async def subscribe(self, subscription: Subscription) -> SubscriptionResponse:
+        """Pass a subscription to the data source and return a subscription response.
+
+        Args:
+            subscription: object of the subscription (time_series_id, start_time)
+
+        Returns:
+            response containing a sample receiver for the time series id provided in
+            the subscription object
+        """
+        receiver = self._receivers.get(subscription)
+        if receiver is None:
+            channel = Broadcast[Sample](f"Source {subscription.time_series_id}")
+            asyncio.create_task(data_sender(channel.get_sender()))
+            receiver = channel.get_receiver()
+            self._receivers[subscription] = receiver
+        return SubscriptionResponse(receiver=receiver, subscription=subscription)
+
+
+async def run() -> None:
+    """Run main functions that initializes and creates everything."""
+    data_provider: DataProvider = DummyDataProvider()
+
+    # Create a channel for receiving subscription responses from the resampling actor
+    subscription_response_channel = Broadcast[SubscriptionResponse](
+        "Subscription Response Channel"
+    )
+    subscription_response_sender = subscription_response_channel.get_sender()
+    subscription_response_receiver = subscription_response_channel.get_receiver()
+
+    # Create a channel for sending subscription requests to the resampling actor
+    subscription_request_channel = Broadcast[SubscriptionRequest](
+        "Subscription Request Channel"
+    )
+    subscription_request_sender = subscription_request_channel.get_sender()
+
+    # Instantiate a resampling actor
+    _resampling_actor = TimeSeriesResamplingActor(
+        data_provider=data_provider,
+        resampling_period_s=1.0,
+        subscription_receiver=subscription_request_channel.get_receiver(),
+    )
+
+    # Declare time series ids to subscribe to for data
+    time_series_ids = [
+        TimeSeriesId(component_id=123, metric_id="active_power"),
+        TimeSeriesId(component_id=99, metric_id="active_power"),
+    ]
+
+    # Create subscription requests for each time series id and provide a sender that
+    # the resampling actor should use to send back subscription responses
+    subscription_requests = [
+        SubscriptionRequest(
+            sender=subscription_response_sender,
+            subscription=Subscription(time_series_id=time_series_id),
+        )
+        for time_series_id in time_series_ids
+    ]
+
+    # Send the subscription requests
+    await asyncio.gather(
+        *[
+            subscription_request_sender.send(request)
+            for request in subscription_requests
+        ]
+    )
+
+    # Store sample receivers for each subscription
+    sample_receivers: Dict[Subscription, Receiver[Sample]] = {}
+
+    while True:
+        select = Select(
+            subscription_response_receiver=subscription_response_receiver,
+            sample_receiver=MergeNamed(
+                **{
+                    str(subscription.time_series_id): receiver
+                    for subscription, receiver in sample_receivers.items()
+                }
+            ),
+        )
+        while await select.ready():
+            # Whenever a subscription response is received from the resampling actor,
+            # it contains a receiver used to consume data for that subscription.
+            # The subscription needs to be stored and the inner while loop restarted
+            # in order to regenerate data receivers passed to the Select
+            if msg := select.subscription_response_receiver:
+                response: SubscriptionResponse = msg.inner
+                sample_receivers[response.subscription] = response.receiver
+                break
+
+            if msg := select.sample_receiver:
+                if msg.inner is not None:
+                    time_series_id, sample = msg.inner
+                    print(f"{time_series_id} => {sample}")
+
+
+asyncio.run(run())