mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-10-29 09:11:57 +00:00
fix: honor DST boundaries in appliance scheduling
Compare elapsed offsets when checking and merging appliance windows so both occurrences of the autumn hour obey the configured boundaries. Preserve calendar durations when finding the earliest valid start. Add optimizer and calendar-day regressions, and consolidate the existing time-window regression files into test_configabc classes.
This commit is contained in:
Normann
Bobby Noelte
@@ -15,6 +15,7 @@ Timezone contract under test:
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
@@ -38,6 +39,10 @@ from akkudoktoreos.config.configabc import ( # noqa — ensure Time is importab
|
||||
ValueTimeWindow,
|
||||
ValueTimeWindowSequence,
|
||||
)
|
||||
from akkudoktoreos.devices.genetic.homeappliance import (
|
||||
HomeAppliance,
|
||||
HomeApplianceParameters,
|
||||
)
|
||||
from akkudoktoreos.utils.datetimeutil import Time, to_datetime, to_duration
|
||||
|
||||
# ===========================================================================
|
||||
@@ -571,6 +576,29 @@ class TestStartTimes:
|
||||
assert result is not None
|
||||
assert result.hour == 8
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("date", "duration", "expected"),
|
||||
[
|
||||
("2026-03-29", "P1D", "2026-03-29T00:00:00+01:00"),
|
||||
("2026-03-29", "PT24H", None),
|
||||
("2026-03-28", "P1D", "2026-03-28T00:00:00+01:00"),
|
||||
("2026-03-28", "PT24H", "2026-03-28T00:00:00+01:00"),
|
||||
("2026-10-25", "P1D", "2026-10-25T00:00:00+02:00"),
|
||||
("2026-10-25", "PT24H", "2026-10-25T00:00:00+02:00"),
|
||||
],
|
||||
)
|
||||
def test_earliest_calendar_day_across_dst(self, date, duration, expected):
|
||||
"""A calendar day fits a local day even when it has fewer than 24 hours."""
|
||||
window = make_tw(start_time="00:00", end_time="00:00")
|
||||
reference = pendulum.from_format(date, "YYYY-MM-DD", tz="Europe/Berlin")
|
||||
run_duration = to_duration(duration)
|
||||
|
||||
earliest = window.earliest_start_time(run_duration, reference)
|
||||
|
||||
assert (earliest.isoformat() if earliest is not None else None) == expected
|
||||
assert window.can_fit_duration(run_duration, reference) is (expected is not None)
|
||||
assert window.contains(reference, run_duration) is (expected is not None)
|
||||
|
||||
def test_latest_naive(self):
|
||||
ref = naive_dt(2024, 6, 15)
|
||||
result = self.w.latest_start_time(pendulum.duration(hours=2), reference_date=ref)
|
||||
@@ -2276,3 +2304,224 @@ class TestCyclesToMatrix:
|
||||
expected,
|
||||
err_msg=f"Row {k} (cycle {cycle_idx}) differs from cycle_to_array (misaligned)",
|
||||
)
|
||||
|
||||
|
||||
class TestTimeWindowBoundaries:
|
||||
"""Explicit window ends, serialization, and cycle scheduling."""
|
||||
|
||||
@pytest.mark.parametrize("model", [TimeWindow, ValueTimeWindow])
|
||||
def test_end_only_window_json_round_trip(self, model):
|
||||
window = model.model_validate({"start_time": "08:00", "end_time": "12:00"})
|
||||
|
||||
serialized = window.model_dump_json()
|
||||
|
||||
assert json.loads(serialized)["duration"] == "PT0S"
|
||||
restored = model.model_validate_json(serialized)
|
||||
assert restored.model_dump() == window.model_dump()
|
||||
assert restored.contains(pendulum.datetime(2026, 10, 26, 9))
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("start_date", "start_time", "end_time", "expected_end"),
|
||||
[
|
||||
# The missing 02:30 on the opening date must not shift Monday's end.
|
||||
("2026-03-29", "22:00", "02:30", "2026-03-30T02:30:00+02:00"),
|
||||
# The explicit end lies after the spring transition.
|
||||
("2026-03-28", "22:00", "06:00", "2026-03-29T06:00:00+02:00"),
|
||||
# The explicit end lies after the autumn transition.
|
||||
("2026-10-24", "22:00", "06:00", "2026-10-25T06:00:00+01:00"),
|
||||
],
|
||||
)
|
||||
def test_overnight_end_uses_clock_time_on_end_date(
|
||||
self, start_date, start_time, end_time, expected_end
|
||||
):
|
||||
window = TimeWindow.model_validate({"start_time": start_time, "end_time": end_time})
|
||||
reference = pendulum.from_format(start_date, "YYYY-MM-DD", tz="Europe/Berlin")
|
||||
|
||||
_, end = window._window_start_end(reference)
|
||||
|
||||
assert end.isoformat() == expected_end
|
||||
assert not window.contains(end)
|
||||
assert window.contains(end.subtract(minutes=1))
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("config", "start", "hours", "active_steps"),
|
||||
[
|
||||
({"start_time": "08:00", "end_time": "12:00"}, "2026-10-26", 24, [8, 9, 10, 11]),
|
||||
# Duration may extend the explicit end.
|
||||
(
|
||||
{"start_time": "08:00", "end_time": "10:00", "duration": "4 hours"},
|
||||
"2026-10-26",
|
||||
24,
|
||||
[8, 9, 10, 11],
|
||||
),
|
||||
({"start_time": "08:00", "end_time": "12:00", "month": 11}, "2026-10-26", 24, []),
|
||||
({"start_time": "08:00", "end_time": "12:00", "day_of_week": 1}, "2026-10-26", 24, []),
|
||||
({"start_time": "08:00", "end_time": "12:00", "date": "2026-10-27"}, "2026-10-26", 24, []),
|
||||
# Include the tail of Friday's window on Saturday.
|
||||
(
|
||||
{"start_time": "22:00", "end_time": "06:00", "day_of_week": 4},
|
||||
"2026-10-24",
|
||||
24,
|
||||
list(range(6)),
|
||||
),
|
||||
# Month restrictions belong to the opening date.
|
||||
(
|
||||
{"start_time": "22:00", "end_time": "06:00", "month": 10},
|
||||
"2026-11-01",
|
||||
24,
|
||||
list(range(6)),
|
||||
),
|
||||
# Long durations can begin more than one calendar day earlier.
|
||||
(
|
||||
{"start_time": "08:00", "duration": "3 days", "date": "2026-10-24"},
|
||||
"2026-10-26",
|
||||
24,
|
||||
list(range(24)),
|
||||
),
|
||||
# Recurring windows must also appear on later horizon days.
|
||||
({"start_time": "08:00", "end_time": "10:00"}, "2026-10-26", 48, [8, 9, 32, 33]),
|
||||
],
|
||||
)
|
||||
def test_cycle_arrays_use_effective_windows(self, config, start, hours, active_steps):
|
||||
window = ValueTimeWindow.model_validate({**config, "value": 0})
|
||||
sequence = CycleTimeWindowSequence(windows=[window])
|
||||
reference = pendulum.from_format(start, "YYYY-MM-DD", tz="UTC")
|
||||
end = reference.add(hours=hours)
|
||||
interval = pendulum.duration(hours=1)
|
||||
expected = np.zeros(hours)
|
||||
expected[active_steps] = 1
|
||||
|
||||
actual = sequence.cycle_to_array(0, reference, end, interval)
|
||||
indices, matrix = sequence.cycles_to_matrix(reference, end, interval)
|
||||
|
||||
assert indices == [0]
|
||||
np.testing.assert_array_equal(actual, expected)
|
||||
np.testing.assert_array_equal(matrix[0], expected)
|
||||
|
||||
@pytest.mark.parametrize(("date", "hours"), [("2026-03-29", 23), ("2026-10-25", 25)])
|
||||
def test_cycle_arrays_across_dst(self, date, hours):
|
||||
window = ValueTimeWindow.model_validate(
|
||||
{"start_time": "00:00", "end_time": "06:00", "value": 0}
|
||||
)
|
||||
sequence = CycleTimeWindowSequence(windows=[window])
|
||||
start = pendulum.from_format(date, "YYYY-MM-DD", tz="Europe/Berlin")
|
||||
end = start.add(days=1)
|
||||
expected = np.zeros(hours)
|
||||
expected[: hours - 18] = 1 # Five elapsed hours in spring; seven in autumn.
|
||||
|
||||
actual = sequence.cycle_to_array(0, start, end, pendulum.duration(hours=1))
|
||||
_, matrix = sequence.cycles_to_matrix(start, end, pendulum.duration(hours=1))
|
||||
|
||||
np.testing.assert_array_equal(actual, expected)
|
||||
np.testing.assert_array_equal(matrix[0], expected)
|
||||
|
||||
def test_end_only_cycle_window_allows_appliance_starts(self):
|
||||
sequence = CycleTimeWindowSequence.model_validate(
|
||||
{"windows": [{"start_time": "08:00", "end_time": "12:00", "value": 0}]}
|
||||
)
|
||||
# Exercise the serialized settings used by the dashboard/API as well.
|
||||
sequence = CycleTimeWindowSequence.model_validate_json(sequence.model_dump_json())
|
||||
appliance = HomeAppliance(
|
||||
parameters=HomeApplianceParameters(
|
||||
device_id="dishwasher",
|
||||
duration_h=2,
|
||||
consumption_wh=2000,
|
||||
num_cycles=1,
|
||||
time_windows=sequence,
|
||||
),
|
||||
prediction_hours=24,
|
||||
optimization_hours=24,
|
||||
)
|
||||
start = pendulum.datetime(2026, 10, 26)
|
||||
appliance.start_allowed = []
|
||||
appliance.start_earliest = []
|
||||
appliance.start_latest = []
|
||||
|
||||
appliance._build_cycle_start_allowed(
|
||||
sequence, start, start.add(days=1), pendulum.duration(hours=1)
|
||||
)
|
||||
|
||||
np.testing.assert_array_equal(np.flatnonzero(appliance.start_allowed[0]), [8, 9, 10])
|
||||
|
||||
|
||||
class TestTimeWindowDST:
|
||||
"""Window boundaries and grids distinguish the repeated DST hour."""
|
||||
|
||||
@staticmethod
|
||||
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(self, duration, expected, fold):
|
||||
window = TimeWindow.model_validate(
|
||||
{"start_time": "01:00", "end_time": "02:30", "duration": duration}
|
||||
)
|
||||
|
||||
_, end = window._window_start_end(self.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(self, 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(self.autumn_time(hour, minute, fold), run_duration) is expected
|
||||
|
||||
def test_membership_rejects_first_fold_before_second_fold_start(self):
|
||||
window = TimeWindow.model_validate({"start_time": "02:30", "end_time": "04:00"})
|
||||
|
||||
assert not window.contains(self.autumn_time(2, 45, fold=0))
|
||||
assert window.contains(self.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(self, start_hour, minute):
|
||||
sequence = CycleTimeWindowSequence.model_validate(
|
||||
{"windows": [{"start_time": "00:00", "end_time": "06:00", "value": 0}]}
|
||||
)
|
||||
start = self.autumn_time(start_hour, fold=0)
|
||||
end = self.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(self, sequence_type):
|
||||
window = ValueTimeWindow.model_validate(
|
||||
{"start_time": "00:00", "end_time": "06:00", "value": 1}
|
||||
)
|
||||
sequence = sequence_type(windows=[window])
|
||||
start = self.autumn_time(0)
|
||||
end = self.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)
|
||||
]
|
||||
|
||||
@@ -7,6 +7,7 @@ from typing import Any
|
||||
from unittest.mock import Mock
|
||||
|
||||
import numpy as np
|
||||
import pendulum
|
||||
import pytest
|
||||
|
||||
from akkudoktoreos.config.configabc import CycleTimeWindowSequence, ValueTimeWindow
|
||||
@@ -182,6 +183,83 @@ class TestExplicitCycleWindow:
|
||||
assert appliance.start_latest[0] == 24 - 3
|
||||
|
||||
|
||||
class TestDSTAllowedStartSlots:
|
||||
"""The optimizer must distinguish both occurrences of the autumn hour."""
|
||||
|
||||
@staticmethod
|
||||
def _appliance(window_field, windows, run_minutes=15):
|
||||
if window_field == "time_windows":
|
||||
windows = [{**window, "value": 0} for window in windows]
|
||||
return HomeAppliance(
|
||||
parameters=HomeApplianceParameters.model_validate(
|
||||
{
|
||||
"device_id": "dishwasher",
|
||||
"load_profile_power_w": [1000],
|
||||
"load_profile_interval_seconds": run_minutes * 60,
|
||||
window_field: {"windows": windows},
|
||||
}
|
||||
),
|
||||
optimization_hours=25,
|
||||
prediction_hours=100,
|
||||
slot_duration_h=0.25,
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("window_field", ["shared_time_windows", "time_windows"])
|
||||
def test_first_fold_starts_before_second_fold_opening_are_rejected(self, window_field):
|
||||
appliance = self._appliance(
|
||||
window_field, [{"start_time": "02:30", "end_time": "04:00"}]
|
||||
)
|
||||
zero = pendulum.datetime(2026, 10, 25, tz="Europe/Berlin")
|
||||
|
||||
allowed = appliance.allowed_start_slots(
|
||||
slot0_datetime=zero, earliest_slot=0, horizon_end_slot=24, cycle_index=0
|
||||
)
|
||||
|
||||
# The second 02:30 is slot 14. The first 02:30/02:45 are slots 10/11
|
||||
# and precede the configured opening despite their matching clock times.
|
||||
assert allowed == list(range(14, 20))
|
||||
|
||||
@pytest.mark.parametrize("window_field", ["shared_time_windows", "time_windows"])
|
||||
def test_first_fold_starts_before_second_fold_closing_are_retained(self, window_field):
|
||||
appliance = self._appliance(
|
||||
window_field, [{"start_time": "01:00", "end_time": "02:30"}]
|
||||
)
|
||||
zero = pendulum.datetime(2026, 10, 25, tz="Europe/Berlin")
|
||||
|
||||
allowed = appliance.allowed_start_slots(
|
||||
slot0_datetime=zero, earliest_slot=0, horizon_end_slot=24, cycle_index=0
|
||||
)
|
||||
|
||||
# Includes a run from the first 02:30 to the first 02:45 (slot 10),
|
||||
# which finishes before the second 02:30 closing boundary.
|
||||
assert allowed == list(range(4, 14))
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("first_duration", "expected"), [("1 hour", []), ("2 hours", [9])]
|
||||
)
|
||||
def test_cycle_window_union_distinguishes_touching_instants_from_clock_times(
|
||||
self, first_duration, expected
|
||||
):
|
||||
appliance = self._appliance(
|
||||
"time_windows",
|
||||
[
|
||||
{"start_time": "01:30", "duration": first_duration},
|
||||
{"start_time": "02:30", "end_time": "03:30"},
|
||||
],
|
||||
run_minutes=90,
|
||||
)
|
||||
zero = pendulum.datetime(2026, 10, 25, tz="Europe/Berlin")
|
||||
|
||||
allowed = appliance.allowed_start_slots(
|
||||
slot0_datetime=zero, earliest_slot=9, horizon_end_slot=15, cycle_index=0
|
||||
)
|
||||
|
||||
# The only candidate runs from the first 02:15 to the second 02:45.
|
||||
# A one-hour first window ends at the first 02:30, leaving a real
|
||||
# one-hour gap; a two-hour first window touches the second opening.
|
||||
assert allowed == expected
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Allowed-start computation: mocked per-cycle windows (CycleTimeWindowSequence)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -1,154 +0,0 @@
|
||||
"""Regression tests for explicit window ends and cycle scheduling."""
|
||||
|
||||
import json
|
||||
|
||||
import numpy as np
|
||||
import pendulum
|
||||
import pytest
|
||||
|
||||
from akkudoktoreos.config.configabc import (
|
||||
CycleTimeWindowSequence,
|
||||
TimeWindow,
|
||||
ValueTimeWindow,
|
||||
)
|
||||
from akkudoktoreos.devices.genetic.homeappliance import (
|
||||
HomeAppliance,
|
||||
HomeApplianceParameters,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("model", [TimeWindow, ValueTimeWindow])
|
||||
def test_end_only_window_json_round_trip(model):
|
||||
window = model.model_validate({"start_time": "08:00", "end_time": "12:00"})
|
||||
|
||||
serialized = window.model_dump_json()
|
||||
|
||||
assert json.loads(serialized)["duration"] == "PT0S"
|
||||
restored = model.model_validate_json(serialized)
|
||||
assert restored.model_dump() == window.model_dump()
|
||||
assert restored.contains(pendulum.datetime(2026, 10, 26, 9))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("start_date", "start_time", "end_time", "expected_end"),
|
||||
[
|
||||
# The missing 02:30 on the opening date must not shift Monday's end.
|
||||
("2026-03-29", "22:00", "02:30", "2026-03-30T02:30:00+02:00"),
|
||||
# The explicit end lies after the spring transition.
|
||||
("2026-03-28", "22:00", "06:00", "2026-03-29T06:00:00+02:00"),
|
||||
# The explicit end lies after the autumn transition.
|
||||
("2026-10-24", "22:00", "06:00", "2026-10-25T06:00:00+01:00"),
|
||||
],
|
||||
)
|
||||
def test_overnight_end_uses_clock_time_on_end_date(start_date, start_time, end_time, expected_end):
|
||||
window = TimeWindow.model_validate({"start_time": start_time, "end_time": end_time})
|
||||
reference = pendulum.from_format(start_date, "YYYY-MM-DD", tz="Europe/Berlin")
|
||||
|
||||
_, end = window._window_start_end(reference)
|
||||
|
||||
assert end.isoformat() == expected_end
|
||||
assert not window.contains(end)
|
||||
assert window.contains(end.subtract(minutes=1))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("config", "start", "hours", "active_steps"),
|
||||
[
|
||||
({"start_time": "08:00", "end_time": "12:00"}, "2026-10-26", 24, [8, 9, 10, 11]),
|
||||
# Duration may extend the explicit end.
|
||||
(
|
||||
{"start_time": "08:00", "end_time": "10:00", "duration": "4 hours"},
|
||||
"2026-10-26",
|
||||
24,
|
||||
[8, 9, 10, 11],
|
||||
),
|
||||
({"start_time": "08:00", "end_time": "12:00", "month": 11}, "2026-10-26", 24, []),
|
||||
({"start_time": "08:00", "end_time": "12:00", "day_of_week": 1}, "2026-10-26", 24, []),
|
||||
({"start_time": "08:00", "end_time": "12:00", "date": "2026-10-27"}, "2026-10-26", 24, []),
|
||||
# Include the tail of Friday's window on Saturday.
|
||||
(
|
||||
{"start_time": "22:00", "end_time": "06:00", "day_of_week": 4},
|
||||
"2026-10-24",
|
||||
24,
|
||||
list(range(6)),
|
||||
),
|
||||
# Month restrictions belong to the opening date.
|
||||
(
|
||||
{"start_time": "22:00", "end_time": "06:00", "month": 10},
|
||||
"2026-11-01",
|
||||
24,
|
||||
list(range(6)),
|
||||
),
|
||||
# Long durations can begin more than one calendar day earlier.
|
||||
(
|
||||
{"start_time": "08:00", "duration": "3 days", "date": "2026-10-24"},
|
||||
"2026-10-26",
|
||||
24,
|
||||
list(range(24)),
|
||||
),
|
||||
# Recurring windows must also appear on later horizon days.
|
||||
({"start_time": "08:00", "end_time": "10:00"}, "2026-10-26", 48, [8, 9, 32, 33]),
|
||||
],
|
||||
)
|
||||
def test_cycle_arrays_use_effective_windows(config, start, hours, active_steps):
|
||||
window = ValueTimeWindow.model_validate({**config, "value": 0})
|
||||
sequence = CycleTimeWindowSequence(windows=[window])
|
||||
reference = pendulum.from_format(start, "YYYY-MM-DD", tz="UTC")
|
||||
end = reference.add(hours=hours)
|
||||
interval = pendulum.duration(hours=1)
|
||||
expected = np.zeros(hours)
|
||||
expected[active_steps] = 1
|
||||
|
||||
actual = sequence.cycle_to_array(0, reference, end, interval)
|
||||
indices, matrix = sequence.cycles_to_matrix(reference, end, interval)
|
||||
|
||||
assert indices == [0]
|
||||
np.testing.assert_array_equal(actual, expected)
|
||||
np.testing.assert_array_equal(matrix[0], expected)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("date", "hours"), [("2026-03-29", 23), ("2026-10-25", 25)])
|
||||
def test_cycle_arrays_across_dst(date, hours):
|
||||
window = ValueTimeWindow.model_validate(
|
||||
{"start_time": "00:00", "end_time": "06:00", "value": 0}
|
||||
)
|
||||
sequence = CycleTimeWindowSequence(windows=[window])
|
||||
start = pendulum.from_format(date, "YYYY-MM-DD", tz="Europe/Berlin")
|
||||
end = start.add(days=1)
|
||||
expected = np.zeros(hours)
|
||||
expected[: hours - 18] = 1 # Five elapsed hours in spring; seven in autumn.
|
||||
|
||||
actual = sequence.cycle_to_array(0, start, end, pendulum.duration(hours=1))
|
||||
_, matrix = sequence.cycles_to_matrix(start, end, pendulum.duration(hours=1))
|
||||
|
||||
np.testing.assert_array_equal(actual, expected)
|
||||
np.testing.assert_array_equal(matrix[0], expected)
|
||||
|
||||
|
||||
def test_end_only_cycle_window_allows_appliance_starts():
|
||||
sequence = CycleTimeWindowSequence.model_validate(
|
||||
{"windows": [{"start_time": "08:00", "end_time": "12:00", "value": 0}]}
|
||||
)
|
||||
# Exercise the serialized settings used by the dashboard/API as well.
|
||||
sequence = CycleTimeWindowSequence.model_validate_json(sequence.model_dump_json())
|
||||
appliance = HomeAppliance(
|
||||
parameters=HomeApplianceParameters(
|
||||
device_id="dishwasher",
|
||||
duration_h=2,
|
||||
consumption_wh=2000,
|
||||
num_cycles=1,
|
||||
time_windows=sequence,
|
||||
),
|
||||
prediction_hours=24,
|
||||
optimization_hours=24,
|
||||
)
|
||||
start = pendulum.datetime(2026, 10, 26)
|
||||
appliance.start_allowed = []
|
||||
appliance.start_earliest = []
|
||||
appliance.start_latest = []
|
||||
|
||||
appliance._build_cycle_start_allowed(
|
||||
sequence, start, start.add(days=1), pendulum.duration(hours=1)
|
||||
)
|
||||
|
||||
np.testing.assert_array_equal(np.flatnonzero(appliance.start_allowed[0]), [8, 9, 10])
|
||||
@@ -1,97 +0,0 @@
|
||||
"""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)
|
||||
]
|
||||
Reference in New Issue
Block a user