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.
 
@@ -63,7 +64,7 @@ class MovingWindow:
         resampled_data_recv=resampled_data_recv,
     )
 
-    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
@@ -88,8 +89,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])
     ```
     """
@@ -100,7 +101,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.
@@ -117,7 +118,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,7 +72,7 @@ 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)))