Add timezone to datetime types

There were inconsistencies in the way the SDK set
datetime variables in the MovingWindow/RingBuffer without
tz/tzinfo and in the Resampler with tz/tzinfo.
This patch sets the timezone UTC for all datetime variables
set in the MovingWindow and RingBuffer to make them consistent
to the datetime variables set in the Resampler.

Signed-off-by: Daniel Zullo <[email protected]>

Author: daniel-zullo-frequenz

benchmarks/timeseries/benchmark_ringbuffer.py

@@ -5,7 +5,7 @@
 
 import random
 import timeit
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, TypeVar
 
 import numpy as np
@@ -23,7 +23,7 @@
 def fill_buffer(days: int, buffer: OrderedRingBuffer[Any]) -> None:
     """Fill the given buffer up to the given amount of days, one sample per minute."""
     random.seed(0)
-    basetime = datetime(2022, 1, 1)
+    basetime = datetime(2022, 1, 1, tzinfo=timezone.utc)
     print("..filling", end="", flush=True)
 
     for day in range(days):
@@ -36,7 +36,7 @@ def fill_buffer(days: int, buffer: OrderedRingBuffer[Any]) -> None:
 
 def test_days(days: int, buffer: OrderedRingBuffer[Any]) -> None:
     """Gets the data for each of the 29 days."""
-    basetime = datetime(2022, 1, 1)
+    basetime = datetime(2022, 1, 1, tzinfo=timezone.utc)
 
     for day in range(days):
         # pylint: disable=unused-variable
@@ -51,7 +51,7 @@ def test_slices(days: int, buffer: OrderedRingBuffer[Any], median: bool) -> None
     Takes a buffer, fills it up and then excessively gets
     the data for each day to calculate the average/median.
     """
-    basetime = datetime(2022, 1, 1)
+    basetime = datetime(2022, 1, 1, tzinfo=timezone.utc)
 
     total = 0.0
 

src/frequenz/sdk/timeseries/_moving_window.py

@@ -9,7 +9,7 @@
 import logging
 import math
 from collections.abc import Sequence
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import SupportsIndex, overload
 
 import numpy as np
@@ -41,9 +41,10 @@ class MovingWindow:
     the point in time that defines the alignment can be outside of the time window.
     Modulo arithmetic is used to move the `window_alignment` timestamp into the
     latest window.
-    If for example the `window_alignment` parameter is set to `datetime(1, 1, 1)`
-    and the window size is bigger than one day then the first element will always
-    be aligned to the midnight. For further information see also the
+    If for example the `window_alignment` parameter is set to
+    `datetime(1, 1, 1, tzinfo=timezone.utc)` and the window size is bigger than
+    one day then the first element will always be aligned to the midnight.
+    For further information see also the
     [`OrderedRingBuffer`][frequenz.sdk.timeseries._ringbuffer.OrderedRingBuffer]
     documentation.
 
@@ -67,7 +68,7 @@ class MovingWindow:
         input_sampling_period=timedelta(seconds=1)
     )
 
-    time_start = datetime.now()
+    time_start = datetime.now(tz=timezone.utc)
     time_end = time_start + timedelta(minutes=5)
 
     # ... wait for 5 minutes until the buffer is filled
@@ -96,8 +97,8 @@ class MovingWindow:
     asyncio.sleep(60*60*24)
 
     # create a polars series with one full day of data
-    time_start = datetime(2023, 1, 1)
-    time_end = datetime(2023, 1, 2)
+    time_start = datetime(2023, 1, 1, tzinfo=timezone.utc)
+    time_end = datetime(2023, 1, 2, tzinfo=timezone.utc)
     s = pl.Series("Jan_1", mv[time_start:time_end])
     ```
     """
@@ -108,7 +109,7 @@ def __init__(  # pylint: disable=too-many-arguments
         resampled_data_recv: Receiver[Sample],
         input_sampling_period: timedelta,
         resampler_config: ResamplerConfig | None = None,
-        window_alignment: datetime = datetime(1, 1, 1),
+        window_alignment: datetime = datetime(1, 1, 1, tzinfo=timezone.utc),
     ) -> None:
         """
         Initialize the MovingWindow.
@@ -121,7 +122,7 @@ def __init__(  # pylint: disable=too-many-arguments
         different sources at different sampling rates or periods, and it can be
         set by specifying the output sampling period parameter so that the user
         can control the granularity of the samples to be stored in the
-        underlying ringbuffer.
+        underlying buffer.
 
         If resampling is not required, the output sampling period parameter
         can be set to None or be equal to the input sampling period in which
@@ -135,7 +136,7 @@ def __init__(  # pylint: disable=too-many-arguments
             resampler_config: The resampler configuration in case resampling is required.
             window_alignment: A datetime object that defines a point in time to which
                 the window is aligned to modulo window size.
-                (default is midnight 01.01.01)
+                (default is midnight 01.01.01 UTC)
                 For further information, consult the class level documentation.
 
         Raises:

src/frequenz/sdk/timeseries/_ringbuffer.py

@@ -8,7 +8,7 @@
 from collections.abc import Iterable
 from copy import deepcopy
 from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Generic, List, SupportsFloat, SupportsIndex, TypeVar, overload
 
 import numpy as np
@@ -50,7 +50,7 @@ def __init__(
         self,
         buffer: FloatArray,
         sampling_period: timedelta,
-        time_index_alignment: datetime = datetime(1, 1, 1),
+        time_index_alignment: datetime = datetime(1, 1, 1, tzinfo=timezone.utc),
     ) -> None:
         """Initialize the time aware ringbuffer.
 
@@ -76,8 +76,8 @@ def __init__(
         self._time_index_alignment: datetime = time_index_alignment
 
         self._gaps: list[Gap] = []
-        self._datetime_newest: datetime = datetime.min
-        self._datetime_oldest: datetime = datetime.max
+        self._datetime_newest: datetime = datetime.min.replace(tzinfo=timezone.utc)
+        self._datetime_oldest: datetime = datetime.max.replace(tzinfo=timezone.utc)
         self._time_range: timedelta = (len(self._buffer) - 1) * sampling_period
 
     @property
@@ -130,7 +130,10 @@ def update(self, sample: Sample) -> None:
         timestamp = self._normalize_timestamp(sample.timestamp)
 
         # Don't add outdated entries
-        if timestamp < self._datetime_oldest and self._datetime_oldest != datetime.max:
+        if (
+            timestamp < self._datetime_oldest
+            and self._datetime_oldest != datetime.max.replace(tzinfo=timezone.utc)
+        ):
             raise IndexError(
                 f"Timestamp {timestamp} too old (cut-off is at {self._datetime_oldest})."
             )
@@ -485,7 +488,7 @@ def __len__(self) -> int:
         Returns:
             The length.
         """
-        if self._datetime_newest == datetime.min:
+        if self._datetime_newest == datetime.min.replace(tzinfo=timezone.utc):
             return 0
 
         start_index = self.datetime_to_index(self._datetime_oldest)

src/frequenz/sdk/timeseries/_serializable_ringbuffer.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 import pickle
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from os.path import exists
 
 from ._ringbuffer import FloatArray, OrderedRingBuffer
@@ -25,7 +25,7 @@ def __init__(
         buffer: FloatArray,
         sampling_period: timedelta,
         path: str,
-        time_index_alignment: datetime = datetime(1, 1, 1),
+        time_index_alignment: datetime = datetime(1, 1, 1, tzinfo=timezone.utc),
     ) -> None:
         """Initialize the time aware ringbuffer.
 

tests/timeseries/test_moving_window.py

@@ -4,7 +4,7 @@
 """Tests for the moving window."""
 
 import asyncio
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Sequence, Tuple
 
 import numpy as np
@@ -23,7 +23,7 @@ async def push_lm_data(sender: Sender[Sample], test_seq: Sequence[float]) -> Non
         sender: Sender for pushing resampled samples to the `MovingWindow`.
         test_seq: The Sequence that is pushed into the `MovingWindow`.
     """
-    start_ts: datetime = datetime(2023, 1, 1)
+    start_ts: datetime = datetime(2023, 1, 1, tzinfo=timezone.utc)
     for i, j in zip(test_seq, range(0, len(test_seq))):
         timestamp = start_ts + timedelta(seconds=j)
         await sender.send(Sample(timestamp, float(i)))
@@ -58,7 +58,7 @@ async def test_access_window_by_timestamp() -> None:
     """Test indexing a window by timestamp"""
     window, sender = init_moving_window(timedelta(seconds=1))
     await push_lm_data(sender, [1])
-    assert np.array_equal(window[datetime(2023, 1, 1)], 1.0)
+    assert np.array_equal(window[datetime(2023, 1, 1, tzinfo=timezone.utc)], 1.0)
 
 
 async def test_access_window_by_int_slice() -> None:
@@ -72,6 +72,6 @@ async def test_access_window_by_ts_slice() -> None:
     """Test accessing a subwindow with a timestamp slice"""
     window, sender = init_moving_window(timedelta(seconds=5))
     await push_lm_data(sender, range(0, 5))
-    time_start = datetime(2023, 1, 1) + timedelta(seconds=3)
+    time_start = datetime(2023, 1, 1, tzinfo=timezone.utc) + timedelta(seconds=3)
     time_end = time_start + timedelta(seconds=2)
     assert np.array_equal(window[time_start:time_end], np.array([3.0, 4.0]))  # type: ignore

tests/timeseries/test_ringbuffer.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import random
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from itertools import cycle, islice
 from typing import Any
 
@@ -30,7 +30,9 @@
             np.empty(shape=(24 * 1800,), dtype=np.float64),
             TWO_HUNDRED_MS,
         ),
-        OrderedRingBuffer([0.0] * 1800, TWO_HUNDRED_MS, datetime(2000, 1, 1)),
+        OrderedRingBuffer(
+            [0.0] * 1800, TWO_HUNDRED_MS, datetime(2000, 1, 1, tzinfo=timezone.utc)
+        ),
     ],
 )
 def test_timestamp_ringbuffer(buffer: OrderedRingBuffer[Any]) -> None:
@@ -46,14 +48,16 @@ def test_timestamp_ringbuffer(buffer: OrderedRingBuffer[Any]) -> None:
     # Push in random order
     # for i in random.sample(range(size), size):
     for i in range(size):
-        buffer.update(Sample(datetime.fromtimestamp(200 + i * resolution), i))
+        buffer.update(
+            Sample(datetime.fromtimestamp(200 + i * resolution, tz=timezone.utc), i)
+        )
 
     # Check all possible window sizes and start positions
     for i in range(0, size, 1000):
         for j in range(1, size - i, 987):
             assert i + j < size
-            start = datetime.fromtimestamp(200 + i * resolution)
-            end = datetime.fromtimestamp(200 + (j + i) * resolution)
+            start = datetime.fromtimestamp(200 + i * resolution, tz=timezone.utc)
+            end = datetime.fromtimestamp(200 + (j + i) * resolution, tz=timezone.utc)
 
             tmp = list(islice(cycle(range(0, size)), i, i + j))
             assert list(buffer.window(start, end)) == list(tmp)
@@ -74,17 +78,17 @@ def test_timestamp_ringbuffer_overwrite(buffer: OrderedRingBuffer[Any]) -> None:
 
     # Push in random order
     for i in random.sample(range(size), size):
-        buffer.update(Sample(datetime.fromtimestamp(200 + i), i))
+        buffer.update(Sample(datetime.fromtimestamp(200 + i, tz=timezone.utc), i))
 
     # Push the same amount twice
     for i in random.sample(range(size), size):
-        buffer.update(Sample(datetime.fromtimestamp(200 + i), i * 2))
+        buffer.update(Sample(datetime.fromtimestamp(200 + i, tz=timezone.utc), i * 2))
 
     # Check all possible window sizes and start positions
     for i in range(size):
         for j in range(1, size - i):
-            start = datetime.fromtimestamp(200 + i)
-            end = datetime.fromtimestamp(200 + j + i)
+            start = datetime.fromtimestamp(200 + i, tz=timezone.utc)
+            end = datetime.fromtimestamp(200 + j + i, tz=timezone.utc)
 
             tmp = islice(cycle(range(0, size * 2, 2)), i, i + j)
             actual: float
@@ -110,28 +114,28 @@ def test_timestamp_ringbuffer_gaps(
 
     # Add initial data
     for i in random.sample(range(size), size):
-        buffer.update(Sample(datetime.fromtimestamp(200 + i), i))
+        buffer.update(Sample(datetime.fromtimestamp(200 + i, tz=timezone.utc), i))
 
     # Request window of the data
     buffer.window(
-        datetime.fromtimestamp(200),
-        datetime.fromtimestamp(202),
+        datetime.fromtimestamp(200, tz=timezone.utc),
+        datetime.fromtimestamp(202, tz=timezone.utc),
     )
 
     # Add entry far in the future
-    buffer.update(Sample(datetime.fromtimestamp(500 + size), 9999))
+    buffer.update(Sample(datetime.fromtimestamp(500 + size, tz=timezone.utc), 9999))
 
     # Expect exception for the same window
     with pytest.raises(IndexError):
         buffer.window(
-            datetime.fromtimestamp(200),
-            datetime.fromtimestamp(202),
+            datetime.fromtimestamp(200, tz=timezone.utc),
+            datetime.fromtimestamp(202, tz=timezone.utc),
         )
 
     # Receive new window without exception
     buffer.window(
-        datetime.fromtimestamp(501),
-        datetime.fromtimestamp(500 + size),
+        datetime.fromtimestamp(501, tz=timezone.utc),
+        datetime.fromtimestamp(500 + size, tz=timezone.utc),
     )
 
 
@@ -149,30 +153,30 @@ def test_timestamp_ringbuffer_missing_parameter(
     buffer: OrderedRingBuffer[Any],
 ) -> None:
     """Test ordered ring buffer."""
-    buffer.update(Sample(datetime(2, 2, 2, 0, 0), 0))
+    buffer.update(Sample(datetime(2, 2, 2, 0, 0, tzinfo=timezone.utc), 0))
 
     # pylint: disable=protected-access
     assert buffer._normalize_timestamp(buffer.gaps[0].start) == buffer.gaps[0].start
 
     # Expecting one gap now, made of all the previous entries of the one just
     # added.
     assert len(buffer.gaps) == 1
-    assert buffer.gaps[0].end == datetime(2, 2, 2)
+    assert buffer.gaps[0].end == datetime(2, 2, 2, tzinfo=timezone.utc)
 
     # Add entry so that a second gap appears
     # pylint: disable=protected-access
-    assert buffer._normalize_timestamp(datetime(2, 2, 2, 0, 7, 31)) == datetime(
-        2, 2, 2, 0, 10
-    )
-    buffer.update(Sample(datetime(2, 2, 2, 0, 7, 31), 0))
+    assert buffer._normalize_timestamp(
+        datetime(2, 2, 2, 0, 7, 31, tzinfo=timezone.utc)
+    ) == datetime(2, 2, 2, 0, 10, tzinfo=timezone.utc)
+    buffer.update(Sample(datetime(2, 2, 2, 0, 7, 31, tzinfo=timezone.utc), 0))
 
     assert buffer.datetime_to_index(
-        datetime(2, 2, 2, 0, 7, 31)
-    ) == buffer.datetime_to_index(datetime(2, 2, 2, 0, 10))
+        datetime(2, 2, 2, 0, 7, 31, tzinfo=timezone.utc)
+    ) == buffer.datetime_to_index(datetime(2, 2, 2, 0, 10, tzinfo=timezone.utc))
     assert len(buffer.gaps) == 2
 
     # import pdb; pdb.set_trace()
-    buffer.update(Sample(datetime(2, 2, 2, 0, 5), 0))
+    buffer.update(Sample(datetime(2, 2, 2, 0, 5, tzinfo=timezone.utc), 0))
     assert len(buffer.gaps) == 1
 
 
@@ -213,7 +217,8 @@ def test_timestamp_ringbuffer_missing_parameter_smoke(
             buffer.update(
                 Sample(
                     datetime.fromtimestamp(
-                        200 + j * buffer.sampling_period.total_seconds()
+                        200 + j * buffer.sampling_period.total_seconds(),
+                        tz=timezone.utc,
                     ),
                     None if missing else j,
                 )
@@ -224,11 +229,11 @@ def test_timestamp_ringbuffer_missing_parameter_smoke(
                 lambda x: Gap(
                     # pylint: disable=protected-access
                     start=buffer._normalize_timestamp(
-                        datetime.fromtimestamp(200 + x[0] * resolution)
+                        datetime.fromtimestamp(200 + x[0] * resolution, tz=timezone.utc)
                     ),
                     # pylint: disable=protected-access
                     end=buffer._normalize_timestamp(
-                        datetime.fromtimestamp(200 + x[1] * resolution)
+                        datetime.fromtimestamp(200 + x[1] * resolution, tz=timezone.utc)
                     ),
                 ),
                 expected_gaps_concrete,
@@ -261,10 +266,10 @@ def test_len_ringbuffer_samples_fit_buffer_size() -> None:
         buffer = OrderedRingBuffer(
             np.empty(shape=len(test_samples), dtype=float),
             sampling_period=timedelta(seconds=1),
-            time_index_alignment=datetime(1, 1, 1),
+            time_index_alignment=datetime(1, 1, 1, tzinfo=timezone.utc),
         )
 
-        start_ts: datetime = datetime(2023, 1, 1)
+        start_ts: datetime = datetime(2023, 1, 1, tzinfo=timezone.utc)
         for index, sample_value in enumerate(test_samples):
             timestamp = start_ts + timedelta(seconds=index)
             buffer.update(Sample(timestamp, float(sample_value)))
@@ -289,10 +294,10 @@ def test_len_ringbuffer_samples_overwrite_buffer() -> None:
         buffer = OrderedRingBuffer(
             np.empty(shape=half_buffer_size, dtype=float),
             sampling_period=timedelta(seconds=1),
-            time_index_alignment=datetime(1, 1, 1),
+            time_index_alignment=datetime(1, 1, 1, tzinfo=timezone.utc),
         )
 
-        start_ts: datetime = datetime(2023, 1, 1)
+        start_ts: datetime = datetime(2023, 1, 1, tzinfo=timezone.utc)
         for index, sample_value in enumerate(test_samples):
             timestamp = start_ts + timedelta(seconds=index)
             buffer.update(Sample(timestamp, float(sample_value)))