diff --git a/pyproject.toml b/pyproject.toml index 158ee0c3f..48d5bb4b7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,7 @@ docs-lint = [ ] format = ["black == 23.3.0", "isort == 5.12.0"] nox = ["nox == 2023.4.22", "toml == 0.10.2"] +examples = [ "polars == 0.18.0" ] pytest = [ "pytest == 7.3.1", "pytest-cov == 4.1.0", @@ -68,6 +69,7 @@ pytest = [ # For checking docstring code examples "sybil == 5.0.2", "pylint == 2.17.4", + "frequenz-sdk[examples]", ] mypy = [ "mypy == 1.3.0", diff --git a/src/frequenz/sdk/timeseries/_moving_window.py b/src/frequenz/sdk/timeseries/_moving_window.py index a34bcc015..442b821f2 100644 --- a/src/frequenz/sdk/timeseries/_moving_window.py +++ b/src/frequenz/sdk/timeseries/_moving_window.py @@ -58,50 +58,74 @@ class MovingWindow: ```python from datetime import datetime, timedelta, timezone - resampled_data_recv = Broadcast[Sample]("sample-data").new_receiver() - window = MovingWindow( - size=timedelta(minutes=5), - resampled_data_recv=resampled_data_recv, - input_sampling_period=timedelta(seconds=1), - ) + async def send_mock_data(sender: Sender[Sample]) -> None: + while True: + await sender.send(Sample(datetime.now(tz=timezone.utc), 10.0)) + await asyncio.sleep(1.0) + + async def run() -> None: + resampled_data_channel = Broadcast[Sample]("sample-data") + resampled_data_receiver = resampled_data_channel.new_receiver() + resampled_data_sender = resampled_data_channel.new_sender() + + send_task = asyncio.create_task(send_mock_data(resampled_data_sender)) + + window = MovingWindow( + size=timedelta(minutes=5), + resampled_data_recv=resampled_data_receiver, + input_sampling_period=timedelta(seconds=1), + ) + + time_start = datetime.now(tz=timezone.utc) + time_end = time_start + timedelta(minutes=5) - time_start = datetime.now(tz=timezone.utc) - time_end = time_start + timedelta(minutes=5) + # ... wait for 5 minutes until the buffer is filled + await asyncio.sleep(5) - # ... wait for 5 minutes until the buffer is filled - await asyncio.sleep(5) + # return an numpy array from the window + array = window[time_start:time_end] + # and use it to for example calculate the mean + mean = array.mean() - # return an numpy array from the window - a = window[time_start:time_end] - # and use it to for example calculate the mean - mean = a.mean() + asyncio.run(run()) ``` Example: Create a polars data frame from a `MovingWindow` ```python - # pylint: disable=import-error import polars as pl from datetime import datetime, timedelta, timezone - sample_receiver = Broadcast[Sample]("sample-data").new_receiver() + async def send_mock_data(sender: Sender[Sample]) -> None: + while True: + await sender.send(Sample(datetime.now(tz=timezone.utc), 10.0)) + await asyncio.sleep(1.0) - # create a window that stores two days of data - # starting at 1.1.23 with samplerate=1 - window = MovingWindow( - size=timedelta(days=2), - resampled_data_recv=sample_receiver, - input_sampling_period=timedelta(seconds=1), - ) + async def run() -> None: + resampled_data_channel = Broadcast[Sample]("sample-data") + resampled_data_receiver = resampled_data_channel.new_receiver() + resampled_data_sender = resampled_data_channel.new_sender() + + send_task = asyncio.create_task(send_mock_data(resampled_data_sender)) + + # create a window that stores two days of data + # starting at 1.1.23 with samplerate=1 + window = MovingWindow( + size=timedelta(days=2), + resampled_data_recv=resampled_data_receiver, + input_sampling_period=timedelta(seconds=1), + ) + + # wait for one full day until the buffer is filled + asyncio.sleep(60*60*24) - # wait for one full day until the buffer is filled - asyncio.sleep(60*60*24) + # create a polars series with one full day of data + time_start = datetime(2023, 1, 1, tzinfo=timezone.utc) + time_end = datetime(2023, 1, 2, tzinfo=timezone.utc) + series = pl.Series("Jan_1", window[time_start:time_end]) - # create a polars series with one full day of data - time_start = datetime(2023, 1, 1, tzinfo=timezone.utc) - time_end = datetime(2023, 1, 2, tzinfo=timezone.utc) - s = pl.Series("Jan_1", window[time_start:time_end]) + asyncio.run(run()) ``` """