Add tests for the PeriodicWindowProfiler

Author: matthias-wende-frequenz

tests/timeseries/test_timeseries_avg.py

@@ -0,0 +1,76 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the timeseries averager."""
+
+import asyncio
+from datetime import datetime, timedelta
+from typing import List
+# import numpy as np
+import logging
+# from frequenz.channels import Broadcast, Sender
+#
+# from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries.timeseries_avg import TimeseriesAvg
+
+from test_moving_window import init_moving_window, push_lm_data
+
+# TODO Tests
+# * moving intervals into ts timeseries_avg
+# * test the average calculation for all cases
+# * test with all zero init_moving_window
+# * run tests with different sample periods
+
+async def create_ts_avg(data: List[float], weights: List[float] | None = None) -> TimeseriesAvg:
+    window, sender = init_moving_window(len(data))
+    await push_lm_data(sender, data)
+
+    return TimeseriesAvg(
+        moving_window = window, # 28 days moving window
+        distance=timedelta(seconds=5),
+        weights = weights,
+    )
+
+def test_interval_shifting(ts_avg: TimeseriesAvg):
+    """
+    Test if a interval is properly shifted into a moving window
+
+    We consider the following case:
+    
+    moving_window = [1,2,2,1,1,1,2,2,1]
+    the window starts at Jan. 01 2023 at 00:00:00
+    """
+    # Test if the timestamp is not shifted
+    ts = datetime(2023, 1, 1, 0, 0, 1)
+    index_not_shifted = ts_avg._shift_ts_to_window(ts)
+    assert(index_not_shifted == 1)
+
+    # Test if a timestamp in the window is shifted to the first appearance of the window
+    ts = datetime(2023, 1, 1, 0, 0, 6)
+    index_shifted = ts_avg._shift_ts_to_window(ts)
+    assert(index_shifted == 1)
+
+    # Test if a timestamp outside the window is shifted
+    ts = datetime(2023, 1, 1, 0, 0, 11)
+    index_shifted = ts_avg._shift_ts_to_window(ts)
+    assert(index_shifted == 1)
+
+
+async def test_ts_avg(ts_avg: TimeseriesAvg):
+    window, sender = init_moving_window(10)
+    await push_lm_data(sender,[1, 2, 3, 4, 5, 6, 7, 8, 9, 1])
+
+    ts_avg = TimeseriesAvg(
+        moving_window = window, # 28 days moving window
+        distance=timedelta(days=7),
+        weights=[0.1, 0.2, 0.3, 0.5]
+    )
+
+async def tests():
+    """ Temporary function until this is started by pytest."""
+    ts_avg = await create_ts_avg([1,2,2,1,1,1,2,2,1])
+    test_interval_shifting(ts_avg)
+    # await test_ts_avg()
+
+logging.basicConfig(level=logging.DEBUG)
+asyncio.run(tests())