add time_utils.get_recent_clock_time_window() function

flexmeasures/utils/tests/test_time_utils.py

@@ -6,6 +6,7 @@
 from flexmeasures.utils.time_utils import (
     server_now,
     naturalized_datetime_str,
+    get_most_recent_clocktime_window,
 )
 
 
@@ -41,3 +42,37 @@ def test_naturalized_datetime_str(
         h_ago = None
     print(h_ago)
     assert naturalized_datetime_str(h_ago, now=now) == exp_result
+
+
+@pytest.mark.parametrize(
+    "window_size,now,exp_start,exp_end",
+    [
+        (
+            5,
+            datetime(2021, 4, 30, 15, 1),
+            datetime(2021, 4, 30, 14, 55),
+            datetime(2021, 4, 30, 15),
+        ),
+        (
+            15,
+            datetime(2021, 4, 30, 3, 36),
+            datetime(2021, 4, 30, 3, 15),
+            datetime(2021, 4, 30, 3, 30),
+        ),
+        (
+            10,
+            datetime(2021, 4, 30, 0, 5),
+            datetime(2021, 4, 29, 23, 50),
+            datetime(2021, 4, 30, 0, 0),
+        ),
+    ],
+)
+def test_recent_clocktime_window(window_size, now, exp_start, exp_end):
+    start, end = get_most_recent_clocktime_window(window_size, now=now)
+    assert start == exp_start
+    assert end == exp_end
+
+
+def test_recent_clocktime_window_invalid_window():
+    with pytest.raises(AssertionError):
+        get_most_recent_clocktime_window(25, now=datetime(2021, 4, 30, 3, 36))

flexmeasures/utils/time_utils.py

@@ -1,5 +1,5 @@
 from datetime import datetime, timedelta
-from typing import List, Union, Optional
+from typing import List, Union, Tuple, Optional
 
 from flask import current_app
 from flask_security.core import current_user
@@ -193,6 +193,30 @@ def get_most_recent_hour() -> datetime:
     return now.replace(minute=now.minute - (now.minute % 60), second=0, microsecond=0)
 
 
+def get_most_recent_clocktime_window(
+    window_size_in_minutes: int, now: Optional[datetime] = None
+) -> Tuple[datetime, datetime]:
+    """
+    Calculate a recent time window, choosing start and end minute so that
+    a full hour can be filled with such windows, e.g.:
+
+    Calling this function at 15:01:xx with window size 5 -> (14:55:00, 15:00:00)
+    Calling this function at 03:36:xx with window size 15 -> (03:15:00, 03:30:00)
+    """
+    assert 60 % window_size_in_minutes == 0
+    if now is None:
+        now = datetime.now(tz=pytz.utc)
+    last_full_minute = now.replace(second=0, microsecond=0) - timedelta(minutes=1)
+    last_round_minute = last_full_minute.minute - (
+        last_full_minute.minute % window_size_in_minutes
+    )
+    begin_time = last_full_minute.replace(minute=last_round_minute) - timedelta(
+        minutes=window_size_in_minutes
+    )
+    end_time = begin_time + timedelta(minutes=window_size_in_minutes)
+    return begin_time, end_time
+
+
 def get_default_start_time() -> datetime:
     return get_most_recent_quarter() - timedelta(days=1)