diff --git a/src/akkudoktoreos/config/configabc.py b/src/akkudoktoreos/config/configabc.py index fdd6ca7e..09c8eee7 100644 --- a/src/akkudoktoreos/config/configabc.py +++ b/src/akkudoktoreos/config/configabc.py @@ -301,11 +301,14 @@ class TimeWindow(SettingsBaseModel): """ start_day = reference_date.start_of("day") + # Use Pendulum's default (second occurrence) for ambiguous boundaries, + # independently of the fold carried by the reference datetime. start = start_day.replace( hour=self.start_time.hour, minute=self.start_time.minute, second=self.start_time.second, microsecond=self.start_time.microsecond, + fold=1, ) duration_end = start + self.duration @@ -334,9 +337,13 @@ class TimeWindow(SettingsBaseModel): minute=self.end_time.minute, second=self.end_time.second, microsecond=self.end_time.microsecond, + fold=1, ) - return start, max(duration_end, explicit_end) + # Same-timezone datetime ordering ignores fold during the repeated DST + # hour. Pendulum subtraction measures elapsed time between the instants. + end = explicit_end if (explicit_end - duration_end).total_seconds() > 0 else duration_end + return start, end def _matches_start_date(self, start_date: DateTime) -> bool: """Check whether a candidate window start date satisfies all restrictions. @@ -394,10 +401,10 @@ class TimeWindow(SettingsBaseModel): start, end = self._window_start_end(start_day) - if not (start <= date_time < end): + if (date_time - start).total_seconds() < 0 or (end - date_time).total_seconds() <= 0: continue - if duration is not None and date_time + duration > end: + if duration is not None and (date_time + duration - end).total_seconds() > 0: continue return True @@ -454,7 +461,7 @@ class TimeWindow(SettingsBaseModel): latest_start = window_end - duration - if latest_start < window_start: + if (latest_start - window_start).total_seconds() < 0: return None return latest_start @@ -752,7 +759,7 @@ class TimeWindowSequence(SettingsBaseModel, Generic[TimeWindowT]): result = [] current = start_datetime - while current < end_datetime: + while (end_datetime - current).total_seconds() > 0: result.append(1.0 if self.contains(current) else 0.0) current = current.add(seconds=interval_s) @@ -817,7 +824,7 @@ class TimeWindowSequence(SettingsBaseModel, Generic[TimeWindowT]): values: list[float] = [] current = start_datetime - while current < end_datetime: + while (end_datetime - current).total_seconds() > 0: timestamps.append(current) values.append(1.0 if self.contains(current) else 0.0) current = current.add(seconds=interval_s) @@ -1000,7 +1007,7 @@ class ValueTimeWindowSequence(TimeWindowSequence[ValueTimeWindow]): result = [] current = start_datetime - while current < end_datetime: + while (end_datetime - current).total_seconds() > 0: step_value: Optional[float] = None matched = False for window in self.windows: @@ -1091,7 +1098,7 @@ class ValueTimeWindowSequence(TimeWindowSequence[ValueTimeWindow]): result: list[float] = [] current = start_datetime - while current < end_datetime: + while (end_datetime - current).total_seconds() > 0: step_value: Optional[float] = None matched = False diff --git a/tests/test_timewindow_dst.py b/tests/test_timewindow_dst.py new file mode 100644 index 00000000..3af11506 --- /dev/null +++ b/tests/test_timewindow_dst.py @@ -0,0 +1,97 @@ +"""Time window boundaries and grids must distinguish the repeated DST hour.""" + +import numpy as np +import pendulum +import pytest + +from akkudoktoreos.config.configabc import ( + CycleTimeWindowSequence, + TimeWindow, + TimeWindowSequence, + ValueTimeWindow, + ValueTimeWindowSequence, +) + + +def autumn_time(hour, minute=0, fold=1): + return pendulum.datetime(2026, 10, 25, hour, minute, tz="Europe/Berlin", fold=fold) + + +@pytest.mark.parametrize( + ("duration", "expected"), + [ + ("PT1H45M", "2026-10-25T02:30:00+01:00"), + ("PT2H45M", "2026-10-25T02:45:00+01:00"), + ], +) +@pytest.mark.parametrize("fold", [0, 1]) +def test_effective_end_uses_later_instant(duration, expected, fold): + window = TimeWindow.model_validate( + {"start_time": "01:00", "end_time": "02:30", "duration": duration} + ) + + _, end = window._window_start_end(autumn_time(2, fold=fold)) + + assert end.isoformat() == expected + + +@pytest.mark.parametrize( + ("hour", "minute", "fold", "duration", "expected"), + [ + (2, 45, 0, None, True), + (2, 30, 0, None, True), + (2, 30, 1, None, False), + (2, 45, 1, None, False), + (2, 45, 0, 45, True), + (2, 45, 0, 46, False), + ], +) +def test_membership_uses_elapsed_time(hour, minute, fold, duration, expected): + window = TimeWindow.model_validate({"start_time": "01:00", "end_time": "02:30"}) + run_duration = pendulum.duration(minutes=duration) if duration is not None else None + + assert window.contains(autumn_time(hour, minute, fold), run_duration) is expected + + +def test_membership_rejects_first_fold_before_second_fold_start(): + window = TimeWindow.model_validate({"start_time": "02:30", "end_time": "04:00"}) + + assert not window.contains(autumn_time(2, 45, fold=0)) + assert window.contains(autumn_time(2, 45, fold=1)) + + +@pytest.mark.parametrize("minute", [0, 30]) +@pytest.mark.parametrize("start_hour", [0, 2]) +def test_cycle_grid_ends_in_second_fold(start_hour, minute): + sequence = CycleTimeWindowSequence.model_validate( + {"windows": [{"start_time": "00:00", "end_time": "06:00", "value": 0}]} + ) + start = autumn_time(start_hour, fold=0) + end = autumn_time(2, minute, fold=1) + interval = pendulum.duration(minutes=30) + expected = np.ones(int((end - start).total_seconds() / interval.total_seconds())) + + actual = sequence.cycle_to_array(0, start, end, interval) + indices, matrix = sequence.cycles_to_matrix(start, end, interval) + + assert indices == [0] + np.testing.assert_array_equal(actual, expected) + np.testing.assert_array_equal(matrix, expected[np.newaxis, :]) + + +@pytest.mark.parametrize("sequence_type", [TimeWindowSequence, ValueTimeWindowSequence]) +def test_array_and_series_keep_both_folds(sequence_type): + window = ValueTimeWindow.model_validate( + {"start_time": "00:00", "end_time": "06:00", "value": 1} + ) + sequence = sequence_type(windows=[window]) + start = autumn_time(0) + end = autumn_time(2, fold=1) + interval = pendulum.duration(hours=1) + + np.testing.assert_array_equal(sequence.to_array(start, end, interval), np.ones(3)) + series = sequence.to_series(start, end, interval) + np.testing.assert_array_equal(series.to_numpy(), np.ones(3)) + assert [stamp.timestamp() for stamp in series.index] == [ + start.add(hours=i).timestamp() for i in range(3) + ]