Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2026-10-09 07:56:40 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity
Files
55e205c602c78b0bd4a6b863a051c2a0889f62d0
EOS/tests/test_configabc.py
T

1998 lines
77 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Tests for configabc.TimeWindow and TimeWindowSequence.
Timezone contract under test:
* ``start_time`` is always **naive** (no ``tzinfo``).
* ``date`` is inherently timezone-free (a calendar date).
* ``date_time`` / ``reference_date`` passed to ``contains()``,
``earliest_start_time()``, and ``latest_start_time()`` may be
timezone-aware or naive.
* When a timezone-aware datetime is supplied, ``start_time`` is
interpreted as wall-clock time **in that timezone** — no tz
conversion is applied to ``start_time`` itself.
* Constructing a ``TimeWindow`` with a naive ``start_time`` raises
``ValidationError``.
"""
import datetime
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
from typing import cast
import numpy as np
import pandas as pd
import pendulum
import pytest
from pydantic import ValidationError
from akkudoktoreos.config.configabc import ( # noqa — ensure Time is importable
CycleTimeWindowSequence,
)
from akkudoktoreos.config.configabc import TimeWindow
from akkudoktoreos.config.configabc import TimeWindow as _TW_check
from akkudoktoreos.config.configabc import ( # noqa — ensure Time is importable
TimeWindowSequence,
ValueTimeWindow,
ValueTimeWindowSequence,
)
from akkudoktoreos.utils.datetimeutil import Time
# ===========================================================================
# Helpers
# ===========================================================================
def naive_dt(year, month, day, hour=0, minute=0, second=0):
"""Return a truly naive DateTime (no timezone)."""
return pendulum.instance(
datetime.datetime(year, month, day, hour, minute, second)
).naive()
def aware_dt(year, month, day, hour=0, minute=0, second=0, tz="Europe/Berlin"):
"""Return a timezone-aware pendulum DateTime."""
return pendulum.datetime(year, month, day, hour, minute, second, tz=tz)
def make_window(start_h, duration_h, **kwargs):
"""Build a TimeWindow with a naive start_time at ``start_h:00``."""
return TimeWindow.model_validate(dict(
start_time=f"{start_h:02d}:00:00",
duration=f"{duration_h} hours",
**kwargs,
))
# ===========================================================================
# Construction / validation
# ===========================================================================
class TestTimeWindowConstruction:
def test_naive_start_time_accepted(self):
w = make_window(8, 2)
assert w.start_time.tzinfo is None
def test_aware_start_time_stripped_to_naive(self):
"""An aware start_time is silently stripped to naive (to_time may add a tz)."""
w = TimeWindow.model_validate(dict(
start_time=Time(8, 0, 0, tzinfo=pendulum.timezone("Europe/Berlin")),
duration="2 hours",
))
assert w.start_time.tzinfo is None
assert w.start_time.hour == 8
def test_duration_string_parsed(self):
w = make_window(8, 3)
assert w.duration.total_seconds() == 3 * 3600
def test_day_of_week_integer_valid(self):
w = make_window(8, 2, day_of_week=0)
assert w.day_of_week == 0
def test_day_of_week_integer_out_of_range(self):
with pytest.raises(ValidationError):
make_window(8, 2, day_of_week=7)
def test_day_of_week_english_string(self):
w = make_window(8, 2, day_of_week="Monday")
assert w.day_of_week == 0
def test_day_of_week_english_string_case_insensitive(self):
w = make_window(8, 2, day_of_week="friday")
assert w.day_of_week == 4
def test_day_of_week_invalid_string(self):
with pytest.raises(ValidationError, match="Invalid weekday"):
make_window(8, 2, day_of_week="notaday")
def test_day_of_week_localized_german(self):
w = make_window(8, 2, day_of_week="Montag", locale="de")
assert w.day_of_week == 0
# ===========================================================================
# _window_start_end
# ===========================================================================
class TestWindowStartEnd:
def test_naive_reference_date(self):
w = make_window(8, 2)
ref = naive_dt(2024, 6, 15, 10, 0, 0)
start, end = w._window_start_end(ref)
assert start.hour == 8
assert start.minute == 0
assert end.hour == 10
assert end.minute == 0
assert start.timezone is None
def test_aware_reference_date_berlin(self):
w = make_window(8, 2)
ref = aware_dt(2024, 6, 15, 10, 0, 0, tz="Europe/Berlin")
start, end = w._window_start_end(ref)
assert start.hour == 8
assert end.hour == 10
assert str(start.timezone) == "Europe/Berlin"
def test_aware_reference_date_utc(self):
w = make_window(6, 4)
ref = aware_dt(2024, 1, 10, 9, 0, 0, tz="UTC")
start, end = w._window_start_end(ref)
assert start.hour == 6
assert end.hour == 10
assert str(start.timezone) == "UTC"
def test_aware_reference_date_eastern(self):
w = make_window(20, 4)
ref = aware_dt(2024, 6, 15, 21, 0, 0, tz="US/Eastern")
start, end = w._window_start_end(ref)
assert start.hour == 20
assert end.hour == 0 # midnight next day
assert str(start.timezone) == "US/Eastern"
# ===========================================================================
# contains() — naive datetime
# ===========================================================================
class TestContainsNaive:
def setup_method(self, method):
self.w = make_window(8, 2)
def test_inside_window(self):
assert self.w.contains(naive_dt(2024, 6, 15, 9, 0, 0))
def test_at_start(self):
assert self.w.contains(naive_dt(2024, 6, 15, 8, 0, 0))
def test_at_end_exclusive(self):
assert not self.w.contains(naive_dt(2024, 6, 15, 10, 0, 0))
def test_before_window(self):
assert not self.w.contains(naive_dt(2024, 6, 15, 7, 59, 59))
def test_after_window(self):
assert not self.w.contains(naive_dt(2024, 6, 15, 10, 0, 1))
def test_with_fitting_duration(self):
dt = naive_dt(2024, 6, 15, 8, 0, 0)
assert self.w.contains(dt, duration=pendulum.duration(hours=2))
def test_with_duration_too_long(self):
dt = naive_dt(2024, 6, 15, 8, 0, 0)
assert not self.w.contains(dt, duration=pendulum.duration(hours=3))
def test_with_duration_starting_late(self):
dt = naive_dt(2024, 6, 15, 9, 30, 0)
assert not self.w.contains(dt, duration=pendulum.duration(hours=1))
def test_with_duration_exactly_fitting_late(self):
dt = naive_dt(2024, 6, 15, 9, 0, 0)
assert self.w.contains(dt, duration=pendulum.duration(hours=1))
# ===========================================================================
# contains() — aware datetime
# ===========================================================================
class TestContainsAware:
def setup_method(self, method):
self.w = make_window(8, 2)
def test_inside_window_berlin(self):
dt = aware_dt(2024, 6, 15, 9, 0, 0, tz="Europe/Berlin")
assert self.w.contains(dt)
def test_before_window_berlin(self):
dt = aware_dt(2024, 6, 15, 7, 30, 0, tz="Europe/Berlin")
assert not self.w.contains(dt)
def test_after_window_berlin(self):
dt = aware_dt(2024, 6, 15, 10, 30, 0, tz="Europe/Berlin")
assert not self.w.contains(dt)
def test_start_time_is_local_not_utc(self):
# 06:00 UTC is before the 08:00 UTC window → outside
dt_utc = aware_dt(2024, 6, 15, 6, 0, 0, tz="UTC")
assert not self.w.contains(dt_utc)
# 08:30 UTC is inside the 08:00–10:00 UTC window
dt_utc_inside = aware_dt(2024, 6, 15, 8, 30, 0, tz="UTC")
assert self.w.contains(dt_utc_inside)
def test_crossing_midnight(self):
w = make_window(23, 2)
dt_inside = aware_dt(2024, 6, 15, 23, 30, 0, tz="Europe/Berlin")
dt_outside = aware_dt(2024, 6, 15, 22, 59, 0, tz="Europe/Berlin")
assert w.contains(dt_inside)
assert not w.contains(dt_outside)
def test_same_wall_clock_different_tz(self):
"""Naive start_time means 12:00 wall clock in *whatever* tz is passed."""
w = make_window(12, 2)
dt_berlin = aware_dt(2024, 6, 15, 13, 0, 0, tz="Europe/Berlin")
dt_ny = aware_dt(2024, 6, 15, 13, 0, 0, tz="US/Eastern")
assert w.contains(dt_berlin)
assert w.contains(dt_ny)
# ===========================================================================
# contains() — day_of_week and date constraints
# ===========================================================================
class TestContainsConstraints:
def test_day_of_week_match_naive(self):
# 2024-06-17 is a Monday (day_of_week == 0)
w = make_window(8, 4, day_of_week=0)
assert w.contains(naive_dt(2024, 6, 17, 9, 0, 0))
def test_day_of_week_no_match_naive(self):
w = make_window(8, 4, day_of_week=0)
# 2024-06-18 is a Tuesday
assert not w.contains(naive_dt(2024, 6, 18, 9, 0, 0))
def test_day_of_week_match_aware(self):
w = make_window(8, 4, day_of_week=0)
dt = aware_dt(2024, 6, 17, 9, 0, 0, tz="Europe/Berlin")
assert w.contains(dt)
def test_date_constraint_match(self):
w = make_window(8, 4, date=pendulum.date(2024, 6, 17))
assert w.contains(naive_dt(2024, 6, 17, 9, 0, 0))
def test_date_constraint_no_match(self):
w = make_window(8, 4, date=pendulum.date(2024, 6, 17))
assert not w.contains(naive_dt(2024, 6, 18, 9, 0, 0))
def test_date_constraint_aware_datetime(self):
w = make_window(8, 4, date=pendulum.date(2024, 6, 17))
dt = aware_dt(2024, 6, 17, 9, 0, 0, tz="US/Eastern")
assert w.contains(dt)
def test_date_and_day_of_week_both_must_hold(self):
# 2024-06-18 is Tuesday; day_of_week=0 (Monday) → False even on matching date
w = make_window(8, 4, date=pendulum.date(2024, 6, 18), day_of_week=0)
assert not w.contains(naive_dt(2024, 6, 18, 9, 0, 0))
# ===========================================================================
# earliest_start_time / latest_start_time
# ===========================================================================
class TestStartTimes:
def setup_method(self, method):
self.w = make_window(8, 4) # 08:00–12:00
def test_earliest_naive(self):
ref = naive_dt(2024, 6, 15)
result = self.w.earliest_start_time(pendulum.duration(hours=2), reference_date=ref)
assert result is not None
assert result.hour == 8
def test_latest_naive(self):
ref = naive_dt(2024, 6, 15)
result = self.w.latest_start_time(pendulum.duration(hours=2), reference_date=ref)
assert result is not None
assert result.hour == 10 # 12:00 - 2h = 10:00
def test_earliest_aware_berlin(self):
ref = aware_dt(2024, 6, 15, tz="Europe/Berlin")
result = self.w.earliest_start_time(pendulum.duration(hours=1), reference_date=ref)
assert result is not None
assert result.hour == 8
assert str(result.timezone) == "Europe/Berlin"
def test_latest_aware_berlin(self):
ref = aware_dt(2024, 6, 15, tz="Europe/Berlin")
result = self.w.latest_start_time(pendulum.duration(hours=1), reference_date=ref)
assert result is not None
assert result.hour == 11
assert str(result.timezone) == "Europe/Berlin"
def test_duration_too_long_returns_none(self):
ref = naive_dt(2024, 6, 15)
result = self.w.earliest_start_time(pendulum.duration(hours=5), reference_date=ref)
assert result is None
def test_wrong_day_of_week_returns_none(self):
w = make_window(8, 4, day_of_week=0) # Monday only
ref = naive_dt(2024, 6, 18) # Tuesday
assert w.earliest_start_time(pendulum.duration(hours=1), reference_date=ref) is None
def test_wrong_date_returns_none(self):
w = make_window(8, 4, date=pendulum.date(2024, 6, 17))
ref = naive_dt(2024, 6, 18)
assert w.earliest_start_time(pendulum.duration(hours=1), reference_date=ref) is None
def test_earliest_aware_utc(self):
ref = aware_dt(2024, 6, 15, tz="UTC")
result = self.w.earliest_start_time(pendulum.duration(hours=2), reference_date=ref)
assert result is not None
assert result.hour == 8
assert str(result.timezone) == "UTC"
def test_latest_equals_window_end_minus_duration(self):
ref = naive_dt(2024, 6, 15)
result = self.w.latest_start_time(pendulum.duration(hours=4), reference_date=ref)
assert result is not None
assert result.hour == 8 # exactly at window start when duration == window size
def test_latest_duration_leaves_no_room(self):
# duration > window → None
ref = naive_dt(2024, 6, 15)
result = self.w.latest_start_time(pendulum.duration(hours=5), reference_date=ref)
assert result is None
# ===========================================================================
# can_fit_duration / available_duration
# ===========================================================================
class TestFitAndAvailable:
def setup_method(self, method):
self.w = make_window(8, 3) # 08:00–11:00
def test_can_fit_exact(self):
assert self.w.can_fit_duration(pendulum.duration(hours=3), naive_dt(2024, 6, 15))
def test_can_fit_shorter(self):
assert self.w.can_fit_duration(pendulum.duration(hours=1), naive_dt(2024, 6, 15))
def test_cannot_fit_longer(self):
assert not self.w.can_fit_duration(pendulum.duration(hours=4), naive_dt(2024, 6, 15))
def test_available_duration_no_constraint(self):
result = self.w.available_duration(naive_dt(2024, 6, 15))
assert result == pendulum.duration(hours=3)
def test_available_duration_wrong_date(self):
w = make_window(8, 3, date=pendulum.date(2024, 6, 17))
result = w.available_duration(naive_dt(2024, 6, 15))
assert result is None
# ===========================================================================
# TimeWindowSequence
# ===========================================================================
class TestTimeWindowSequence:
def setup_method(self, method):
self.seq = TimeWindowSequence[TimeWindow](
windows=[
make_window(8, 2), # 08:00–10:00
make_window(14, 3), # 14:00–17:00
]
)
def test_contains_first_window(self):
assert self.seq.contains(naive_dt(2024, 6, 15, 9, 0, 0))
def test_contains_second_window(self):
assert self.seq.contains(naive_dt(2024, 6, 15, 15, 0, 0))
def test_contains_gap_between_windows(self):
assert not self.seq.contains(naive_dt(2024, 6, 15, 12, 0, 0))
def test_contains_with_duration_fits_second(self):
dt = naive_dt(2024, 6, 15, 14, 0, 0)
assert self.seq.contains(dt, duration=pendulum.duration(hours=2))
def test_contains_aware(self):
dt = aware_dt(2024, 6, 15, 9, 0, 0, tz="Europe/Berlin")
assert self.seq.contains(dt)
def test_earliest_start_time(self):
ref = naive_dt(2024, 6, 15)
result = self.seq.earliest_start_time(pendulum.duration(hours=1), ref)
assert result is not None
assert result.hour == 8
def test_latest_start_time(self):
ref = naive_dt(2024, 6, 15)
result = self.seq.latest_start_time(pendulum.duration(hours=1), ref)
assert result is not None
assert result.hour == 16 # 17:00 - 1h
def test_available_duration_sum(self):
ref = naive_dt(2024, 6, 15)
result = self.seq.available_duration(ref)
assert result == pendulum.duration(hours=5)
def test_empty_sequence_contains_false(self):
seq = TimeWindowSequence[TimeWindow]()
assert not seq.contains(naive_dt(2024, 6, 15, 9, 0, 0))
def test_empty_sequence_earliest_none(self):
seq = TimeWindowSequence[TimeWindow]()
assert seq.earliest_start_time(pendulum.duration(hours=1), naive_dt(2024, 6, 15)) is None
def test_empty_sequence_available_none(self):
seq = TimeWindowSequence[TimeWindow]()
assert seq.available_duration(naive_dt(2024, 6, 15)) is None
def test_get_applicable_windows(self):
ref = naive_dt(2024, 6, 15)
applicable = self.seq.get_applicable_windows(ref)
assert len(applicable) == 2
def test_find_windows_for_duration_fits_both(self):
ref = naive_dt(2024, 6, 15)
fits = self.seq.find_windows_for_duration(pendulum.duration(hours=1), ref)
assert len(fits) == 2
def test_find_windows_for_duration_fits_only_second(self):
ref = naive_dt(2024, 6, 15)
fits = self.seq.find_windows_for_duration(pendulum.duration(hours=3), ref)
assert len(fits) == 1
assert fits[0].start_time.hour == 14
def test_sort_windows_by_start_time(self):
seq = TimeWindowSequence[TimeWindow](
windows=[make_window(14, 1), make_window(8, 1)]
)
ref = naive_dt(2024, 6, 15)
seq.sort_windows_by_start_time(ref)
assert seq.windows[0].start_time.hour == 8
assert seq.windows[1].start_time.hour == 14
def test_add_and_remove_window(self):
seq = TimeWindowSequence[TimeWindow]()
w = make_window(10, 1)
seq.add_window(w)
assert len(seq) == 1
removed = seq.remove_window(0)
assert removed == w
assert len(seq) == 0
def test_remove_from_empty_raises(self):
seq = TimeWindowSequence[TimeWindow]()
with pytest.raises(IndexError):
seq.remove_window(0)
def test_get_all_possible_start_times(self):
ref = naive_dt(2024, 6, 15)
result = self.seq.get_all_possible_start_times(pendulum.duration(hours=1), ref)
assert len(result) == 2
earliest_hours = sorted(e.hour for e, _, _ in result)
assert earliest_hours == [8, 14]
def test_iter_and_len_and_getitem(self):
assert len(self.seq) == 2
windows = list(self.seq)
assert len(windows) == 2
assert self.seq[0].start_time.hour == 8
# ===========================================================================
# ValueTimeWindow / ValueTimeWindowSequence
# ===========================================================================
class TestValueTimeWindow:
def test_value_stored(self):
w = ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=0.288))
assert w.value == pytest.approx(0.288)
def test_value_default_none(self):
w = ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours"))
assert w.value is None
def test_inherits_aware_start_time_stripped(self):
"""ValueTimeWindow inherits the strip-to-naive behaviour from TimeWindow."""
w = ValueTimeWindow.model_validate(dict(
start_time=Time(8, 0, 0, tzinfo=pendulum.timezone("UTC")),
duration="2 hours",
value=0.1,
))
assert w.start_time.tzinfo is None
assert w.start_time.hour == 8
class TestValueTimeWindowSequence:
def setup_method(self, method):
self.seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=0.25)),
ValueTimeWindow.model_validate(dict(start_time="18:00:00", duration="4 hours", value=0.35)),
]
)
def test_get_value_morning(self):
dt = naive_dt(2024, 6, 15, 9, 0, 0)
assert self.seq.get_value_for_datetime(dt) == pytest.approx(0.25)
def test_get_value_evening(self):
dt = naive_dt(2024, 6, 15, 19, 0, 0)
assert self.seq.get_value_for_datetime(dt) == pytest.approx(0.35)
def test_get_value_outside_all_windows(self):
dt = naive_dt(2024, 6, 15, 13, 0, 0)
assert self.seq.get_value_for_datetime(dt) == pytest.approx(0.0)
def test_get_value_none_value_returns_zero(self):
seq = ValueTimeWindowSequence(
windows=[ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=None))]
)
assert seq.get_value_for_datetime(naive_dt(2024, 6, 15, 9, 0, 0)) == pytest.approx(0.0)
def test_get_value_aware_datetime(self):
dt = aware_dt(2024, 6, 15, 9, 0, 0, tz="Europe/Berlin")
assert self.seq.get_value_for_datetime(dt) == pytest.approx(0.25)
# ===========================================================================
# TimeWindowSequence.to_array
# ===========================================================================
class TestTimeWindowSequenceToArray:
"""Tests for TimeWindowSequence.to_array.
Window layout used throughout:
win1: 08:00–10:00 (2 h)
win2: 14:00–17:00 (3 h)
Grid step = 1 hour unless stated otherwise.
"""
def setup_method(self, method):
self.seq = TimeWindowSequence[TimeWindow](
windows=[
make_window(8, 2), # 08:00–10:00
make_window(14, 3), # 14:00–17:00
]
)
# ------------------------------------------------------------------
# basic correctness
# ------------------------------------------------------------------
def test_basic_1h_steps_naive(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 0).add(hours=24)
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.shape == (24,)
# Window 1: hours 8, 9
assert arr[8] == pytest.approx(1.0)
assert arr[9] == pytest.approx(1.0)
assert arr[10] == pytest.approx(0.0)
# Window 2: hours 14, 15, 16
assert arr[14] == pytest.approx(1.0)
assert arr[15] == pytest.approx(1.0)
assert arr[16] == pytest.approx(1.0)
assert arr[17] == pytest.approx(0.0)
# Gap between windows
assert arr[12] == pytest.approx(0.0)
def test_basic_1h_steps_aware_berlin(self):
start = aware_dt(2024, 6, 15, 0, tz="Europe/Berlin")
end = aware_dt(2024, 6, 16, 0, tz="Europe/Berlin")
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.shape == (24,)
assert arr[8] == pytest.approx(1.0)
assert arr[9] == pytest.approx(1.0)
assert arr[10] == pytest.approx(0.0)
assert arr[14] == pytest.approx(1.0)
assert arr[16] == pytest.approx(1.0)
assert arr[17] == pytest.approx(0.0)
def test_outside_all_windows_all_zeros(self):
# Only 2 hours at midnight — no overlap with any window
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 2)
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert np.all(arr == 0.0)
def test_inside_one_window_all_ones(self):
# Entirely inside window 1 (08:00–10:00)
start = naive_dt(2024, 6, 15, 8)
end = naive_dt(2024, 6, 15, 10)
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert np.all(arr == 1.0)
def test_dtype_is_float64(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.dtype == np.float64
def test_end_is_exclusive(self):
# end == window start → 0 steps inside
start = naive_dt(2024, 6, 15, 6)
end = naive_dt(2024, 6, 15, 8) # exclusive — 08:00 itself not emitted
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.shape == (2,)
assert np.all(arr == 0.0)
# ------------------------------------------------------------------
# sub-hour steps
# ------------------------------------------------------------------
def test_30min_steps(self):
start = naive_dt(2024, 6, 15, 8)
end = naive_dt(2024, 6, 15, 10)
arr = self.seq.to_array(start, end, pendulum.duration(minutes=30))
# Steps: 08:00, 08:30 → both inside [08:00, 10:00)
assert arr.shape == (4,)
assert np.all(arr == 1.0)
def test_15min_steps_boundary(self):
# Steps at 09:45, 10:00, 10:15; only 09:45 inside window
start = naive_dt(2024, 6, 15, 9, 45)
end = naive_dt(2024, 6, 15, 10, 30)
arr = self.seq.to_array(start, end, pendulum.duration(minutes=15))
# align_to_interval=True floors to interval boundary
# interval=15 min; 09:45 is already on a 15-min boundary
assert arr[0] == pytest.approx(1.0) # 09:45 inside win1
assert arr[1] == pytest.approx(0.0) # 10:00 outside (exclusive end)
# ------------------------------------------------------------------
# align_to_interval
# ------------------------------------------------------------------
def test_align_to_interval_false_preserves_start(self):
# Start at 08:10 — not on a whole-hour boundary
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
arr = self.seq.to_array(
start, end, pendulum.duration(hours=1), align_to_interval=False
)
# Steps: 08:10 (inside win1), 09:10 (inside win1), 10:10 (outside)
assert arr.shape == (2,)
assert arr[0] == pytest.approx(1.0)
assert arr[1] == pytest.approx(1.0)
def test_align_to_interval_true_floors_start(self):
# Start at 08:10; floored to 08:00 with 1h interval
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
arr = self.seq.to_array(
start, end, pendulum.duration(hours=1), align_to_interval=True
)
# After flooring: steps 08:00, 09:00, 10:00 → 3 steps
assert arr.shape == (3,)
assert arr[0] == pytest.approx(1.0) # 08:00
assert arr[1] == pytest.approx(1.0) # 09:00
assert arr[2] == pytest.approx(0.0) # 10:00
# ------------------------------------------------------------------
# boundary validation
# ------------------------------------------------------------------
def test_unsupported_boundary_raises(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
with pytest.raises(ValueError, match="boundary"):
self.seq.to_array(start, end, pendulum.duration(hours=1), boundary="strict")
# ------------------------------------------------------------------
# dropna (no effect for binary windows — accepted for compat)
# ------------------------------------------------------------------
def test_dropna_true_no_effect(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
arr_t = self.seq.to_array(start, end, pendulum.duration(hours=1), dropna=True)
arr_f = self.seq.to_array(start, end, pendulum.duration(hours=1), dropna=False)
np.testing.assert_array_equal(arr_t, arr_f)
# ------------------------------------------------------------------
# empty sequence
# ------------------------------------------------------------------
def test_empty_sequence_all_zeros(self):
seq = TimeWindowSequence[TimeWindow]()
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
arr = seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.shape == (4,)
assert np.all(arr == 0.0)
# ------------------------------------------------------------------
# day_of_week and date constraints propagate
# ------------------------------------------------------------------
def test_day_of_week_constraint_respected(self):
# Monday-only window; 2024-06-17 is Monday, 2024-06-18 is Tuesday
seq = TimeWindowSequence[TimeWindow](windows=[make_window(8, 2, day_of_week=0)])
monday_start = naive_dt(2024, 6, 17, 7)
tuesday_start = naive_dt(2024, 6, 18, 7)
end_offset = pendulum.duration(hours=4)
arr_mon = seq.to_array(monday_start, monday_start.add(hours=4),
pendulum.duration(hours=1))
arr_tue = seq.to_array(tuesday_start, tuesday_start.add(hours=4),
pendulum.duration(hours=1))
assert arr_mon[1] == pytest.approx(1.0) # 08:00 Monday — inside
assert np.all(arr_tue == 0.0) # Tuesday — all outside
# ===========================================================================
# TimeWindowSequence.to_series
# ===========================================================================
class TestTimeWindowSequenceToSeries:
"""Tests for TimeWindowSequence.to_series.
Window layout:
win1: 08:00–10:00
win2: 14:00–17:00
"""
def setup_method(self, method):
self.seq = TimeWindowSequence[TimeWindow](
windows=[
make_window(8, 2),
make_window(14, 3),
]
)
def test_basic_1h_steps_naive(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
series = self.seq.to_series(
start, end, pendulum.duration(hours=1)
)
assert isinstance(series, pd.Series)
assert series.shape == (24,)
assert isinstance(series.index, pd.DatetimeIndex)
assert series.iloc[8] == pytest.approx(1.0)
assert series.iloc[9] == pytest.approx(1.0)
assert series.iloc[10] == pytest.approx(0.0)
assert series.iloc[14] == pytest.approx(1.0)
assert series.iloc[15] == pytest.approx(1.0)
assert series.iloc[16] == pytest.approx(1.0)
assert series.iloc[17] == pytest.approx(0.0)
def test_values_match_to_array(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
interval = pendulum.duration(hours=1)
arr = self.seq.to_array(start, end, interval)
series = self.seq.to_series(start, end, interval)
np.testing.assert_array_equal(series.to_numpy(), arr)
def test_dtype_is_float64(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
series = self.seq.to_series(
start, end, pendulum.duration(hours=1)
)
assert series.dtype == np.float64
def test_end_is_exclusive(self):
start = naive_dt(2024, 6, 15, 6)
end = naive_dt(2024, 6, 15, 8)
series = self.seq.to_series(
start, end, pendulum.duration(hours=1)
)
index = cast(pd.DatetimeIndex, series.index)
assert series.shape == (2,)
assert list(index.hour) == [6, 7]
assert np.all(series.to_numpy() == 0.0)
def test_align_to_interval_false_preserves_start(self):
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
series = self.seq.to_series(
start,
end,
pendulum.duration(hours=1),
align_to_interval=False,
)
index = cast(pd.DatetimeIndex, series.index)
assert series.shape == (2,)
assert index[0] == pd.Timestamp(start)
assert index[1] == pd.Timestamp(start.add(hours=1))
assert np.all(series.to_numpy() == 1.0)
def test_align_to_interval_true_floors_start(self):
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
series = self.seq.to_series(
start,
end,
pendulum.duration(hours=1),
align_to_interval=True,
)
index = cast(pd.DatetimeIndex, series.index)
assert series.shape == (3,)
assert list(index.hour) == [8, 9, 10]
assert series.iloc[0] == pytest.approx(1.0)
assert series.iloc[1] == pytest.approx(1.0)
assert series.iloc[2] == pytest.approx(0.0)
def test_aware_datetime_preserves_timezone(self):
start = aware_dt(2024, 6, 15, 0, tz="Europe/Berlin")
end = aware_dt(2024, 6, 15, 4, tz="Europe/Berlin")
series = self.seq.to_series(
start, end, pendulum.duration(hours=1)
)
index = cast(pd.DatetimeIndex, series.index)
assert series.shape == (4,)
assert str(index.tz) == "Europe/Berlin"
def test_unsupported_boundary_raises(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
with pytest.raises(ValueError, match="boundary"):
self.seq.to_series(
start,
end,
pendulum.duration(hours=1),
boundary="strict",
)
def test_empty_sequence_all_zeros(self):
seq = TimeWindowSequence[TimeWindow]()
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
series = seq.to_series(
start, end, pendulum.duration(hours=1)
)
assert series.shape == (4,)
assert np.all(series.to_numpy() == 0.0)
# ===========================================================================
# ValueTimeWindowSequence.to_array
# ===========================================================================
class TestValueTimeWindowSequenceToArray:
"""Tests for ValueTimeWindowSequence.to_array.
Window layout:
win1: 08:00–12:00 value=0.25
win2: 18:00–22:00 value=0.35
"""
def setup_method(self, method):
self.seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=0.25)),
ValueTimeWindow.model_validate(dict(start_time="18:00:00", duration="4 hours", value=0.35)),
]
)
# ------------------------------------------------------------------
# basic correctness
# ------------------------------------------------------------------
def test_basic_1h_steps_values(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.shape == (24,)
# win1: hours 8–11
assert arr[8] == pytest.approx(0.25)
assert arr[11] == pytest.approx(0.25)
assert arr[12] == pytest.approx(0.0)
# win2: hours 18–21
assert arr[18] == pytest.approx(0.35)
assert arr[21] == pytest.approx(0.35)
assert arr[22] == pytest.approx(0.0)
# Gap
assert arr[0] == pytest.approx(0.0)
assert arr[14] == pytest.approx(0.0)
def test_dtype_is_float64(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.dtype == np.float64
def test_zero_outside_all_windows(self):
start = naive_dt(2024, 6, 15, 12)
end = naive_dt(2024, 6, 15, 18)
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert np.all(arr == 0.0)
def test_aware_datetime_berlin(self):
start = aware_dt(2024, 6, 15, 0, tz="Europe/Berlin")
end = aware_dt(2024, 6, 16, 0, tz="Europe/Berlin")
arr = self.seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.shape == (24,)
assert arr[8] == pytest.approx(0.25)
assert arr[18] == pytest.approx(0.35)
assert arr[7] == pytest.approx(0.0)
# ------------------------------------------------------------------
# dropna semantics
# ------------------------------------------------------------------
def test_dropna_false_none_value_emits_nan(self):
seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=None)),
ValueTimeWindow.model_validate(dict(start_time="12:00:00", duration="2 hours", value=0.5)),
]
)
start = naive_dt(2024, 6, 15, 8)
end = naive_dt(2024, 6, 15, 15)
arr = seq.to_array(start, end, pendulum.duration(hours=1), dropna=False)
# Steps: 08, 09 (nan), 10 (0), 11 (0), 12 (0.5), 13 (0.5), 14 (0)
assert arr.shape == (7,)
assert np.isnan(arr[0])
assert np.isnan(arr[1])
assert arr[2] == pytest.approx(0.0)
assert arr[4] == pytest.approx(0.5)
def test_dropna_true_none_value_step_omitted(self):
seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=None)),
ValueTimeWindow.model_validate(dict(start_time="12:00:00", duration="2 hours", value=0.5)),
]
)
start = naive_dt(2024, 6, 15, 8)
end = naive_dt(2024, 6, 15, 15)
arr = seq.to_array(start, end, pendulum.duration(hours=1), dropna=True)
# 08 and 09 dropped (None value), remaining 5 steps: 10,11,12,13,14
assert arr.shape == (5,)
assert arr[0] == pytest.approx(0.0) # 10:00
assert arr[1] == pytest.approx(0.0) # 11:00
assert arr[2] == pytest.approx(0.5) # 12:00
assert arr[3] == pytest.approx(0.5) # 13:00
assert arr[4] == pytest.approx(0.0) # 14:00
def test_dropna_no_none_values_same_result(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 6)
arr_t = self.seq.to_array(start, end, pendulum.duration(hours=1), dropna=True)
arr_f = self.seq.to_array(start, end, pendulum.duration(hours=1), dropna=False)
np.testing.assert_array_equal(arr_t, arr_f)
# ------------------------------------------------------------------
# align_to_interval and boundary
# ------------------------------------------------------------------
def test_align_to_interval_false(self):
# Start at 08:30 — between steps
start = naive_dt(2024, 6, 15, 8, 30)
end = naive_dt(2024, 6, 15, 12, 30)
arr = self.seq.to_array(
start, end, pendulum.duration(hours=1), align_to_interval=False
)
# Steps: 08:30, 09:30, 10:30, 11:30 → all inside win1 [08:00–12:00)
assert arr.shape == (4,)
assert np.all(arr == pytest.approx(0.25))
def test_unsupported_boundary_raises(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
with pytest.raises(ValueError, match="boundary"):
self.seq.to_array(start, end, pendulum.duration(hours=1), boundary="inner")
# ------------------------------------------------------------------
# empty sequence
# ------------------------------------------------------------------
def test_empty_sequence_all_zeros(self):
seq = ValueTimeWindowSequence()
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
arr = seq.to_array(start, end, pendulum.duration(hours=1))
assert arr.shape == (4,)
assert np.all(arr == 0.0)
# ------------------------------------------------------------------
# overlapping windows — first match wins
# ------------------------------------------------------------------
def test_overlapping_windows_first_wins(self):
seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=0.10)),
ValueTimeWindow.model_validate(dict(start_time="09:00:00", duration="4 hours", value=0.99)),
]
)
start = naive_dt(2024, 6, 15, 9)
end = naive_dt(2024, 6, 15, 11)
arr = seq.to_array(start, end, pendulum.duration(hours=1))
# 09:00 and 10:00 are in both windows; first (0.10) must win
assert arr[0] == pytest.approx(0.10)
assert arr[1] == pytest.approx(0.10)
# ===========================================================================
# ValueTimeWindowSequence.to_series
# ===========================================================================
class TestValueTimeWindowSequenceToSeries:
"""Tests for ValueTimeWindowSequence.to_series."""
def setup_method(self, method):
self.seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(
start_time="08:00:00",
duration="4 hours",
value=0.25,
)),
ValueTimeWindow.model_validate(dict(
start_time="18:00:00",
duration="4 hours",
value=0.35,
)),
]
)
def test_basic_1h_steps_values(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
series = self.seq.to_series(
start, end, pendulum.duration(hours=1)
)
assert isinstance(series, pd.Series)
assert isinstance(series.index, pd.DatetimeIndex)
assert series.shape == (24,)
assert series.iloc[8] == pytest.approx(0.25)
assert series.iloc[11] == pytest.approx(0.25)
assert series.iloc[12] == pytest.approx(0.0)
assert series.iloc[18] == pytest.approx(0.35)
assert series.iloc[21] == pytest.approx(0.35)
assert series.iloc[22] == pytest.approx(0.0)
def test_values_match_to_array(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
interval = pendulum.duration(hours=1)
arr = self.seq.to_array(start, end, interval)
series = self.seq.to_series(start, end, interval)
np.testing.assert_array_equal(series.to_numpy(), arr)
def test_dtype_is_float64(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
series = self.seq.to_series(
start, end, pendulum.duration(hours=1)
)
assert series.dtype == np.float64
def test_dropna_false_none_value_emits_nan(self):
seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(
start_time="08:00:00",
duration="2 hours",
value=None,
)),
ValueTimeWindow.model_validate(dict(
start_time="12:00:00",
duration="2 hours",
value=0.5,
)),
]
)
start = naive_dt(2024, 6, 15, 8)
end = naive_dt(2024, 6, 15, 15)
series = seq.to_series(
start,
end,
pendulum.duration(hours=1),
dropna=False,
)
index = cast(pd.DatetimeIndex, series.index)
assert series.shape == (7,)
assert list(index.hour) == [8, 9, 10, 11, 12, 13, 14]
assert np.isnan(series.iloc[0])
assert np.isnan(series.iloc[1])
assert series.iloc[2] == pytest.approx(0.0)
assert series.iloc[3] == pytest.approx(0.0)
assert series.iloc[4] == pytest.approx(0.5)
assert series.iloc[5] == pytest.approx(0.5)
assert series.iloc[6] == pytest.approx(0.0)
def test_dropna_true_none_value_omits_timestamp(self):
seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(
start_time="08:00:00",
duration="2 hours",
value=None,
)),
ValueTimeWindow.model_validate(dict(
start_time="12:00:00",
duration="2 hours",
value=0.5,
)),
]
)
start = naive_dt(2024, 6, 15, 8)
end = naive_dt(2024, 6, 15, 15)
series = seq.to_series(
start,
end,
pendulum.duration(hours=1),
dropna=True,
)
index = cast(pd.DatetimeIndex, series.index)
# 08:00 and 09:00 are omitted completely.
assert series.shape == (5,)
assert list(index.hour) == [10, 11, 12, 13, 14]
np.testing.assert_allclose(
series.to_numpy(),
[0.0, 0.0, 0.5, 0.5, 0.0],
)
def test_dropna_no_none_values_same_result(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 6)
interval = pendulum.duration(hours=1)
series_true = self.seq.to_series(
start, end, interval, dropna=True
)
series_false = self.seq.to_series(
start, end, interval, dropna=False
)
pd.testing.assert_series_equal(series_true, series_false)
def test_aware_datetime_preserves_timezone(self):
start = aware_dt(2024, 6, 15, 0, tz="Europe/Berlin")
end = aware_dt(2024, 6, 15, 4, tz="Europe/Berlin")
series = self.seq.to_series(
start, end, pendulum.duration(hours=1)
)
index = cast(pd.DatetimeIndex, series.index)
assert series.shape == (4,)
assert str(index.tz) == "Europe/Berlin"
def test_align_to_interval_true_floors_start(self):
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
series = self.seq.to_series(
start,
end,
pendulum.duration(hours=1),
align_to_interval=True,
)
index = cast(pd.DatetimeIndex, series.index)
assert series.shape == (3,)
assert list(index.hour) == [8, 9, 10]
assert series.iloc[0] == pytest.approx(0.25)
assert series.iloc[1] == pytest.approx(0.25)
assert series.iloc[2] == pytest.approx(0.25)
def test_align_to_interval_false_preserves_start(self):
start = naive_dt(2024, 6, 15, 8, 30)
end = naive_dt(2024, 6, 15, 12, 30)
series = self.seq.to_series(
start,
end,
pendulum.duration(hours=1),
align_to_interval=False,
)
assert series.shape == (4,)
assert series.index[0] == pd.Timestamp(start)
assert series.iloc[0] == pytest.approx(0.25)
assert np.all(series.to_numpy() == pytest.approx(0.25))
def test_unsupported_boundary_raises(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
with pytest.raises(ValueError, match="boundary"):
self.seq.to_series(
start,
end,
pendulum.duration(hours=1),
boundary="inner",
)
def test_empty_sequence_all_zeros(self):
seq = ValueTimeWindowSequence()
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 4)
series = seq.to_series(
start, end, pendulum.duration(hours=1)
)
assert series.shape == (4,)
assert np.all(series.to_numpy() == 0.0)
def test_overlapping_windows_first_wins(self):
seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(
start_time="08:00:00",
duration="4 hours",
value=0.10,
)),
ValueTimeWindow.model_validate(dict(
start_time="09:00:00",
duration="4 hours",
value=0.99,
)),
]
)
start = naive_dt(2024, 6, 15, 9)
end = naive_dt(2024, 6, 15, 11)
series = seq.to_series(
start, end, pendulum.duration(hours=1)
)
assert series.iloc[0] == pytest.approx(0.10)
assert series.iloc[1] == pytest.approx(0.10)
# ===========================================================================
# align_to_interval — timezone-invariance
#
# These tests reproduce the bug that existed before the wall-clock floor fix.
# The old epoch-arithmetic implementation gave wrong results when the machine's
# local timezone was non-UTC:
# - For naive datetimes, pendulum.instance() attached the *local* timezone,
# so subtracting a UTC epoch shifted the floored start.
# - For aware datetimes, subtracting a UTC epoch converted to UTC first,
# then epoch.add() returned a UTC datetime instead of preserving the
# original timezone.
#
# The `set_other_timezone` fixture (from conftest.py) temporarily changes
# pendulum's local timezone via pendulum.set_local_timezone() and restores it
# after the test. Calling it with no argument picks a non-UTC default
# ("Atlantic/Canary" or "Asia/Singapore"); calling it with "UTC" sets UTC.
#
# Each scenario runs two tests:
# _utc — local tz = UTC (passes even with the old code)
# _nonUTC — local tz = non-UTC (would have FAILED with the old code)
# ===========================================================================
class TestAlignToIntervalTimezoneInvariance:
"""Verify align_to_interval produces identical results regardless of
the machine's local timezone.
Tests are paired: ``_utc`` sets local tz to UTC, ``_non_utc`` sets it to
a non-UTC zone via ``set_other_timezone()``. The pair must produce the
same array — any divergence indicates a timezone-dependent bug.
"""
# ------------------------------------------------------------------
# helpers
# ------------------------------------------------------------------
@staticmethod
def _tws_naive():
"""TimeWindowSequence with one window 08:00–10:00."""
return TimeWindowSequence(windows=[make_window(8, 2)])
@staticmethod
def _tws_naive_start():
return naive_dt(2024, 6, 15, 8, 10)
@staticmethod
def _tws_naive_end():
return naive_dt(2024, 6, 15, 10, 10)
# ------------------------------------------------------------------
# TimeWindowSequence — naive datetime, 1-hour steps
# floor 08:10 → 08:00; expect steps 08:00(1), 09:00(1), 10:00(0)
# ------------------------------------------------------------------
def test_tws_naive_floor_utc(self, set_other_timezone):
set_other_timezone("UTC")
arr = self._tws_naive().to_array(
self._tws_naive_start(), self._tws_naive_end(),
pendulum.duration(hours=1), align_to_interval=True,
)
assert arr.shape == (3,)
assert arr[0] == pytest.approx(1.0)
assert arr[1] == pytest.approx(1.0)
assert arr[2] == pytest.approx(0.0)
def test_tws_naive_floor_non_utc(self, set_other_timezone):
set_other_timezone()
arr = self._tws_naive().to_array(
self._tws_naive_start(), self._tws_naive_end(),
pendulum.duration(hours=1), align_to_interval=True,
)
assert arr.shape == (3,)
assert arr[0] == pytest.approx(1.0)
assert arr[1] == pytest.approx(1.0)
assert arr[2] == pytest.approx(0.0)
def test_tws_series_naive_floor_non_utc(self, set_other_timezone):
set_other_timezone()
series = self._tws_naive().to_series(
self._tws_naive_start(),
self._tws_naive_end(),
pendulum.duration(hours=1),
align_to_interval=True,
)
assert series.shape == (3,)
assert list(series.index.hour) == [8, 9, 10]
assert series.iloc[0] == pytest.approx(1.0)
assert series.iloc[1] == pytest.approx(1.0)
assert series.iloc[2] == pytest.approx(0.0)
# ------------------------------------------------------------------
# TimeWindowSequence — naive datetime, 30-min steps
# floor 08:10 → 08:00; expect steps 08:00(1), 08:30(1), 09:00(1), 09:30(1), 10:00(0)
# ------------------------------------------------------------------
def test_tws_naive_30min_floor_utc(self, set_other_timezone):
set_other_timezone("UTC")
arr = self._tws_naive().to_array(
self._tws_naive_start(), self._tws_naive_end(),
pendulum.duration(minutes=30), align_to_interval=True,
)
assert arr.shape == (5,)
assert np.all(arr[:4] == pytest.approx(1.0))
assert arr[4] == pytest.approx(0.0)
def test_tws_naive_30min_floor_non_utc(self, set_other_timezone):
set_other_timezone()
arr = self._tws_naive().to_array(
self._tws_naive_start(), self._tws_naive_end(),
pendulum.duration(minutes=30), align_to_interval=True,
)
assert arr.shape == (5,)
assert np.all(arr[:4] == pytest.approx(1.0))
assert arr[4] == pytest.approx(0.0)
# ------------------------------------------------------------------
# TimeWindowSequence — aware datetime (Europe/Berlin), 1-hour steps
# floor 08:10 Berlin → 08:00 Berlin; timezone must be preserved
# ------------------------------------------------------------------
def test_tws_aware_floor_utc(self, set_other_timezone):
set_other_timezone("UTC")
seq = self._tws_naive()
start = aware_dt(2024, 6, 15, 8, 10, tz="Europe/Berlin")
end = aware_dt(2024, 6, 15, 10, 10, tz="Europe/Berlin")
arr = seq.to_array(start, end, pendulum.duration(hours=1), align_to_interval=True)
assert arr.shape == (3,)
assert arr[0] == pytest.approx(1.0)
assert arr[1] == pytest.approx(1.0)
assert arr[2] == pytest.approx(0.0)
def test_tws_aware_floor_non_utc(self, set_other_timezone):
set_other_timezone()
seq = self._tws_naive()
start = aware_dt(2024, 6, 15, 8, 10, tz="Europe/Berlin")
end = aware_dt(2024, 6, 15, 10, 10, tz="Europe/Berlin")
arr = seq.to_array(start, end, pendulum.duration(hours=1), align_to_interval=True)
assert arr.shape == (3,)
assert arr[0] == pytest.approx(1.0)
assert arr[1] == pytest.approx(1.0)
assert arr[2] == pytest.approx(0.0)
def test_tws_series_aware_floor_non_utc(self, set_other_timezone):
set_other_timezone()
seq = self._tws_naive()
start = aware_dt(2024, 6, 15, 8, 10, tz="Europe/Berlin")
end = aware_dt(2024, 6, 15, 10, 10, tz="Europe/Berlin")
series = seq.to_series(
start,
end,
pendulum.duration(hours=1),
align_to_interval=True,
)
assert series.shape == (3,)
assert list(series.index.hour) == [8, 9, 10]
assert str(series.index.tz) == "Europe/Berlin"
assert series.iloc[0] == pytest.approx(1.0)
assert series.iloc[1] == pytest.approx(1.0)
assert series.iloc[2] == pytest.approx(0.0)
# ------------------------------------------------------------------
# ValueTimeWindowSequence — naive datetime, 1-hour steps
# floor 08:10 → 08:00; values 0.25 at 08:00, 09:00; 0.0 at 10:00
# ------------------------------------------------------------------
def test_vtws_naive_floor_utc(self, set_other_timezone):
set_other_timezone("UTC")
seq = ValueTimeWindowSequence(windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=0.25))
])
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
arr = seq.to_array(start, end, pendulum.duration(hours=1), align_to_interval=True)
assert arr.shape == (3,)
assert arr[0] == pytest.approx(0.25)
assert arr[1] == pytest.approx(0.25)
assert arr[2] == pytest.approx(0.0)
def test_vtws_naive_floor_non_utc(self, set_other_timezone):
set_other_timezone()
seq = ValueTimeWindowSequence(windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=0.25))
])
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
arr = seq.to_array(start, end, pendulum.duration(hours=1), align_to_interval=True)
assert arr.shape == (3,)
assert arr[0] == pytest.approx(0.25)
assert arr[1] == pytest.approx(0.25)
assert arr[2] == pytest.approx(0.0)
def test_vtws_series_naive_floor_non_utc(self, set_other_timezone):
set_other_timezone()
seq = ValueTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(
start_time="08:00:00",
duration="2 hours",
value=0.25,
))
]
)
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
series = seq.to_series(
start,
end,
pendulum.duration(hours=1),
align_to_interval=True,
)
index = cast(pd.DatetimeIndex, series.index)
assert series.shape == (3,)
assert list(index.hour) == [8, 9, 10]
assert series.iloc[0] == pytest.approx(0.25)
assert series.iloc[1] == pytest.approx(0.25)
assert series.iloc[2] == pytest.approx(0.0)
# ===========================================================================
# CycleTimeWindowSequence
# ===========================================================================
class TestCycleTimeWindowSequence:
"""Tests for CycleTimeWindowSequence.
Window layout:
win1: 08:00–12:00 cycle=0
win2: 14:00–18:00 cycle=1
win3: 20:00–22:00 cycle=2
"""
def setup_method(self, method):
self.seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=0.0)),
ValueTimeWindow.model_validate(dict(start_time="14:00:00", duration="4 hours", value=1.0)),
ValueTimeWindow.model_validate(dict(start_time="20:00:00", duration="2 hours", value=2.0)),
]
)
# ------------------------------------------------------------------
# cycle detection
# ------------------------------------------------------------------
def test_num_cycles(self):
assert self.seq.num_cycles() == 3
def test_num_cycles_ignores_none(self):
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=None)),
ValueTimeWindow.model_validate(dict(start_time="10:00:00", duration="2 hours", value=1.0)),
]
)
assert seq.num_cycles() == 1
def test_num_cycles_non_contiguous(self):
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=2.0)),
ValueTimeWindow.model_validate(dict(start_time="10:00:00", duration="2 hours", value=5.0)),
]
)
assert seq.num_cycles() == 2
# ------------------------------------------------------------------
# cycle_to_array basic correctness
# ------------------------------------------------------------------
def test_cycle0_array(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
arr = self.seq.cycle_to_array(
0, start, end, pendulum.duration(hours=1)
)
assert arr.shape == (24,)
assert arr[8] == pytest.approx(1.0)
assert arr[11] == pytest.approx(1.0)
assert arr[12] == pytest.approx(0.0)
def test_cycle1_array(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
arr = self.seq.cycle_to_array(
1, start, end, pendulum.duration(hours=1)
)
assert arr[14] == pytest.approx(1.0)
assert arr[17] == pytest.approx(1.0)
assert arr[13] == pytest.approx(0.0)
def test_cycle2_array(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
arr = self.seq.cycle_to_array(
2, start, end, pendulum.duration(hours=1)
)
assert arr[20] == pytest.approx(1.0)
assert arr[21] == pytest.approx(1.0)
assert arr[22] == pytest.approx(0.0)
# ------------------------------------------------------------------
# cycle not present
# ------------------------------------------------------------------
def test_cycle_not_present_all_zero(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 6)
arr = self.seq.cycle_to_array(
5, start, end, pendulum.duration(hours=1)
)
assert np.all(arr == 0.0)
# ------------------------------------------------------------------
# aware datetime
# ------------------------------------------------------------------
def test_cycle_array_aware_datetime(self):
start = aware_dt(2024, 6, 15, 0, tz="Europe/Berlin")
end = aware_dt(2024, 6, 16, 0, tz="Europe/Berlin")
arr = self.seq.cycle_to_array(
1, start, end, pendulum.duration(hours=1)
)
assert arr[14] == pytest.approx(1.0)
assert arr[13] == pytest.approx(0.0)
# ------------------------------------------------------------------
# dtype
# ------------------------------------------------------------------
def test_cycle_array_dtype(self):
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 6)
arr = self.seq.cycle_to_array(
0, start, end, pendulum.duration(hours=1)
)
assert arr.dtype == np.float64
# ------------------------------------------------------------------
# dropna propagation
# ------------------------------------------------------------------
def test_cycle_array_dropna_false(self):
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=None)),
ValueTimeWindow.model_validate(dict(start_time="10:00:00", duration="2 hours", value=1.0)),
]
)
start = naive_dt(2024, 6, 15, 8)
end = naive_dt(2024, 6, 15, 13)
arr = seq.cycle_to_array(
1, start, end, pendulum.duration(hours=1), dropna=False
)
assert arr[2] == pytest.approx(1.0)
assert arr[3] == pytest.approx(1.0)
# ===========================================================================
# CycleTimeWindowSequence.cycles_to_matrix
# ===========================================================================
class TestCyclesToMatrix:
"""Tests for CycleTimeWindowSequence.cycles_to_matrix.
The method returns ``(cycle_indices, matrix)`` where:
* ``cycle_indices`` — sorted list of distinct integer cycle indices
(derived from the integer part of each window's ``value``).
* ``matrix`` — shape ``(len(cycle_indices), n_steps)`` float64 array;
``matrix[k, t] == 1.0`` iff step ``t`` falls inside a window whose
cycle index equals ``cycle_indices[k]``, ``0.0`` otherwise.
Alignment contract (same as ``to_array`` and ``cycle_to_array``):
* ``start_datetime`` is floored to the nearest interval boundary in
wall-clock time before building the grid.
* The step count uses ``math.ceil`` so that a partially-covered final
step is included, consistent with ``to_array``'s while-loop.
Window layout used in ``setup_method`` unless overridden:
cycle 0: 08:00–12:00 (4 h)
cycle 1: 14:00–18:00 (4 h)
cycle 2: 20:00–22:00 (2 h)
All tests use 1-hour steps over a 24-hour horizon unless stated otherwise.
"""
def setup_method(self, method):
self.seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=0.0)),
ValueTimeWindow.model_validate(dict(start_time="14:00:00", duration="4 hours", value=1.0)),
ValueTimeWindow.model_validate(dict(start_time="20:00:00", duration="2 hours", value=2.0)),
]
)
self.start = naive_dt(2024, 6, 15, 0)
self.end = naive_dt(2024, 6, 16, 0)
self.interval = pendulum.duration(hours=1)
# ------------------------------------------------------------------
# return-value structure
# ------------------------------------------------------------------
def test_returns_tuple_of_two(self):
result = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
assert isinstance(result, tuple) and len(result) == 2
def test_cycle_indices_is_sorted_list(self):
indices, _ = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
assert indices == sorted(indices)
assert isinstance(indices, list)
def test_cycle_indices_values(self):
indices, _ = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
assert indices == [0, 1, 2]
def test_matrix_shape(self):
indices, matrix = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
assert matrix.shape == (len(indices), 24)
def test_matrix_dtype_float64(self):
_, matrix = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
assert matrix.dtype == np.float64
# ------------------------------------------------------------------
# correctness: cycle-row contents
# ------------------------------------------------------------------
def test_cycle0_row_marks_correct_steps(self):
# Cycle 0: 08:00–12:00 → steps 8, 9, 10, 11
_, matrix = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
assert matrix[0, 7] == pytest.approx(0.0)
assert matrix[0, 8] == pytest.approx(1.0)
assert matrix[0, 11] == pytest.approx(1.0)
assert matrix[0, 12] == pytest.approx(0.0)
def test_cycle1_row_marks_correct_steps(self):
# Cycle 1: 14:00–18:00 → steps 14, 15, 16, 17
_, matrix = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
assert matrix[1, 13] == pytest.approx(0.0)
assert matrix[1, 14] == pytest.approx(1.0)
assert matrix[1, 17] == pytest.approx(1.0)
assert matrix[1, 18] == pytest.approx(0.0)
def test_cycle2_row_marks_correct_steps(self):
# Cycle 2: 20:00–22:00 → steps 20, 21
_, matrix = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
assert matrix[2, 19] == pytest.approx(0.0)
assert matrix[2, 20] == pytest.approx(1.0)
assert matrix[2, 21] == pytest.approx(1.0)
assert matrix[2, 22] == pytest.approx(0.0)
def test_cycle_rows_are_mutually_exclusive(self):
# No step should be 1.0 in more than one row
_, matrix = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
overlap = (matrix > 0.5).sum(axis=0)
assert np.all(overlap <= 1)
def test_steps_outside_all_windows_are_zero_in_all_rows(self):
_, matrix = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
outside = list(range(0, 8)) + [12, 13, 18, 19] + list(range(22, 24))
for t in outside:
assert matrix[:, t].sum() == pytest.approx(0.0), f"step {t} should be all-zero"
# ------------------------------------------------------------------
# None value windows are skipped
# ------------------------------------------------------------------
def test_none_value_windows_skipped(self):
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=None)),
ValueTimeWindow.model_validate(dict(start_time="10:00:00", duration="2 hours", value=1.0)),
]
)
indices, matrix = seq.cycles_to_matrix(self.start, self.end, self.interval)
assert indices == [1]
assert matrix.shape == (1, 24)
# The None window (08:00–10:00) must not bleed into cycle 1's row
assert matrix[0, 8] == pytest.approx(0.0)
assert matrix[0, 10] == pytest.approx(1.0)
assert matrix[0, 11] == pytest.approx(1.0)
def test_all_none_returns_empty(self):
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=None)),
]
)
indices, matrix = seq.cycles_to_matrix(self.start, self.end, self.interval)
assert indices == []
assert matrix.shape == (0, 24)
# ------------------------------------------------------------------
# row ordering: sorted by cycle index regardless of window order
# ------------------------------------------------------------------
def test_row_order_independent_of_window_order(self):
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="20:00:00", duration="2 hours", value=2.0)),
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=0.0)),
ValueTimeWindow.model_validate(dict(start_time="14:00:00", duration="4 hours", value=1.0)),
]
)
indices, matrix = seq.cycles_to_matrix(self.start, self.end, self.interval)
assert indices == [0, 1, 2]
# Row 0 must be cycle 0 (08:00–12:00)
assert matrix[0, 8] == pytest.approx(1.0)
assert matrix[0, 14] == pytest.approx(0.0)
# ------------------------------------------------------------------
# non-contiguous cycle indices
# ------------------------------------------------------------------
def test_non_contiguous_cycle_indices(self):
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="06:00:00", duration="2 hours", value=3.0)),
ValueTimeWindow.model_validate(dict(start_time="16:00:00", duration="2 hours", value=7.0)),
]
)
indices, matrix = seq.cycles_to_matrix(self.start, self.end, self.interval)
assert indices == [3, 7]
assert matrix.shape == (2, 24)
assert matrix[0, 6] == pytest.approx(1.0)
assert matrix[0, 7] == pytest.approx(1.0)
assert matrix[1, 16] == pytest.approx(1.0)
assert matrix[1, 17] == pytest.approx(1.0)
# ------------------------------------------------------------------
# multiple windows for the same cycle index (union)
# ------------------------------------------------------------------
def test_same_cycle_multiple_windows_union(self):
# Cycle 0 appears twice: 06:00–08:00 and 20:00–22:00
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="06:00:00", duration="2 hours", value=0.0)),
ValueTimeWindow.model_validate(dict(start_time="20:00:00", duration="2 hours", value=0.0)),
]
)
indices, matrix = seq.cycles_to_matrix(self.start, self.end, self.interval)
assert indices == [0]
assert matrix[0, 6] == pytest.approx(1.0)
assert matrix[0, 7] == pytest.approx(1.0)
assert matrix[0, 20] == pytest.approx(1.0)
assert matrix[0, 21] == pytest.approx(1.0)
assert matrix[0, 8] == pytest.approx(0.0)
# ------------------------------------------------------------------
# sub-hour steps
# ------------------------------------------------------------------
def test_30min_steps(self):
# Cycle 0: 08:00–12:00 → 8 half-hour steps starting at step 16 (08:00 / 0.5h)
seq = CycleTimeWindowSequence(
windows=[
ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=0.0)),
]
)
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 16, 0)
_, matrix = seq.cycles_to_matrix(start, end, pendulum.duration(minutes=30))
assert matrix.shape == (1, 48)
# Steps 16–23 (08:00–12:00 in 30-min slots)
assert matrix[0, 15] == pytest.approx(0.0)
assert matrix[0, 16] == pytest.approx(1.0)
assert matrix[0, 23] == pytest.approx(1.0)
assert matrix[0, 24] == pytest.approx(0.0)
# ------------------------------------------------------------------
# aware datetime
# ------------------------------------------------------------------
def test_aware_datetime_berlin(self):
start = aware_dt(2024, 6, 15, 0, tz="Europe/Berlin")
end = aware_dt(2024, 6, 16, 0, tz="Europe/Berlin")
indices, matrix = self.seq.cycles_to_matrix(start, end, self.interval)
assert indices == [0, 1, 2]
assert matrix[0, 8] == pytest.approx(1.0)
assert matrix[0, 11] == pytest.approx(1.0)
assert matrix[0, 12] == pytest.approx(0.0)
def test_aware_datetime_utc(self):
start = aware_dt(2024, 6, 15, 0, tz="UTC")
end = aware_dt(2024, 6, 16, 0, tz="UTC")
indices, matrix = self.seq.cycles_to_matrix(start, end, self.interval)
assert matrix[0, 8] == pytest.approx(1.0)
assert matrix[1, 14] == pytest.approx(1.0)
# ------------------------------------------------------------------
# short horizon — window partially or fully outside
# ------------------------------------------------------------------
def test_window_fully_outside_horizon(self):
# Only first 6 hours — all windows are outside
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 6)
indices, matrix = self.seq.cycles_to_matrix(start, end, self.interval)
assert matrix.shape == (3, 6)
assert np.all(matrix == 0.0)
def test_window_partially_inside_horizon_clipped(self):
# Horizon ends at 10:00; cycle 0 window is 08:00–12:00 → only steps 8, 9 inside
start = naive_dt(2024, 6, 15, 0)
end = naive_dt(2024, 6, 15, 10)
indices, matrix = self.seq.cycles_to_matrix(start, end, self.interval)
assert matrix.shape == (3, 10)
assert matrix[0, 8] == pytest.approx(1.0)
assert matrix[0, 9] == pytest.approx(1.0)
assert matrix[0, :8].sum() == pytest.approx(0.0)
# ------------------------------------------------------------------
# empty sequence
# ------------------------------------------------------------------
def test_empty_sequence(self):
seq = CycleTimeWindowSequence()
indices, matrix = seq.cycles_to_matrix(self.start, self.end, self.interval)
assert indices == []
assert matrix.shape == (0, 24)
assert matrix.dtype == np.float64
# ------------------------------------------------------------------
# alignment: misaligned start_datetime is floored (wall-clock floor)
# ------------------------------------------------------------------
def test_misaligned_start_floored_step_count(self):
# start=08:10, end=10:10, interval=1h
# floor(08:10) = 08:00 → steps: 08:00, 09:00, 10:00 = 3 steps
# (10:00 < 10:10, so it is included by ceil)
# Window 08:00–12:00 → all three steps are inside → all 1.0
seq = CycleTimeWindowSequence(
windows=[ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="4 hours", value=0.0))]
)
start = naive_dt(2024, 6, 15, 8, 10)
end = naive_dt(2024, 6, 15, 10, 10)
_, matrix = seq.cycles_to_matrix(start, end, pendulum.duration(hours=1))
assert matrix.shape == (1, 3)
np.testing.assert_array_equal(matrix[0], [1.0, 1.0, 1.0])
def test_misaligned_start_30min_steps(self):
# start=08:15, interval=30min → floor to 08:00
# Window 08:00–10:00 → steps 08:00(1), 08:30(1), 09:00(1), 09:30(1)
seq = CycleTimeWindowSequence(
windows=[ValueTimeWindow.model_validate(dict(start_time="08:00:00", duration="2 hours", value=0.0))]
)
start = naive_dt(2024, 6, 15, 8, 15)
end = naive_dt(2024, 6, 15, 10, 15)
_, matrix = seq.cycles_to_matrix(start, end, pendulum.duration(minutes=30))
# floor(08:15, 30min) = 08:00; steps: 08:00,08:30,09:00,09:30,10:00(ceil)
# 10:00 is outside [08:00,10:00) → 0.0
assert matrix.shape[1] >= 4
np.testing.assert_array_equal(matrix[0, :4], [1.0, 1.0, 1.0, 1.0])
assert matrix[0, 4] == pytest.approx(0.0) # 10:00 outside
# ------------------------------------------------------------------
# consistency with cycle_to_array
# ------------------------------------------------------------------
def test_matrix_rows_match_cycle_to_array(self):
"""Each matrix row must exactly match the corresponding cycle_to_array output.
Both methods now use the same wall-clock floor alignment and math.ceil
step count, so they must produce identical arrays for every cycle.
"""
indices, matrix = self.seq.cycles_to_matrix(self.start, self.end, self.interval)
for k, cycle_idx in enumerate(indices):
expected = self.seq.cycle_to_array(
cycle_idx, self.start, self.end, self.interval
)
np.testing.assert_array_equal(
matrix[k],
expected,
err_msg=f"Row {k} (cycle {cycle_idx}) differs from cycle_to_array",
)
def test_matrix_rows_match_cycle_to_array_misaligned(self):
"""Consistency holds even when start_datetime is not on an interval boundary."""
start = naive_dt(2024, 6, 15, 8, 20)
end = naive_dt(2024, 6, 15, 14, 20)
interval = pendulum.duration(hours=1)
indices, matrix = self.seq.cycles_to_matrix(start, end, interval)
for k, cycle_idx in enumerate(indices):
expected = self.seq.cycle_to_array(cycle_idx, start, end, interval)
np.testing.assert_array_equal(
matrix[k],
expected,
err_msg=f"Row {k} (cycle {cycle_idx}) differs from cycle_to_array (misaligned)",
)
Reference in New Issue View Git Blame Copy Permalink