mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-10-09 16:06:40 +00:00
8224c64654
* feat: adapt configuration for multi optimization algorithms Decouple configuration from optimization algorithm parameters. Add to_[algorithm]_param() methods to the configuration that derive optimization algorithm specific parameters from the configuration. Add x-scope tags to the configuration options that describe for which specific algorithms the configuration option is for. The whole device settings are restructured. There are now general settings for the device classes with the afore mentioned to_[algorithm]_param() methods. The general device settings got their own directory `devices/settings`. By this the parameter class also does not have to be a pydantic model which can be used for future optimization/ simulations speed up. Also the parameter class for a device is now part of the device module. This better decouples and also is the natural place for parameters of a device. Besides this feature there are also fixes and improvements: * feat: extend home appliance time window settings and simulation Home appliance can now be configured for multiple runs with per-cycle allowed time windows. The number of remaining cycles to plan is determined at runtime by reading the ``cycles_completed_measurement_key`` from the measurement store. * feat: specialiced CycleTimeWindowSequence for time window sequences Sequence of time windows associated to cycles. This model specializes ``ValueTimeWindowSequence`` so that the ``value`` field of each ``ValueTimeWindow`` encodes the **cycle index** (0-based integer) the window belongs to. Typical use: an appliance that must run ``n`` times per day, each run constrained to a distinct time window. Assign ``value=0`` to windows for the first cycle, ``value=1`` for the second, and so on. Multiple windows may share the same cycle index (their allowed regions are unioned). Windows with ``value=None`` are silently ignored by all cycle-aware methods. * fix: Make test_configmigrate also regard the _ANY_SENTENIEL in key values * chore: Make devices configurations a map instead of a list This makes config paths stable regardless of declaration order and lets each device settings class build its own config path from ``self.device_id`` without needing an external index. Tests are adapted likewise. Devices configurations are automatically migrated from lists to maps. * chore: rename levelized_cost_of_storage_kwh to levelized_cost_of_storage_amt kwh This better fits in the naming scheme and also makes clear the costs are money. Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com> * fix: runtime config update ignored by config file Runtime settings were handed back to pydantic-settings as init settings, which rank below the config file and the environment. Any key already present in EOS.config.json or in the environment silently discarded the update, so a bulk PUT /v1/config returned 200 without applying anything, while the granular PUT /v1/config/{path} endpoint kept working. Add a dedicated runtime settings source ranked directly below the command line arguments and record granular updates there as well, so both endpoints share one store that survives re-evaluation of the settings sources. Environment variables keep precedence over the config file for all keys that were not set at runtime. Also repairs revert_settings() and update(), which passed their data through the same init settings. Closes #1303 * fix: env vars ignored on first config build ConfigEOS.__init__ passed self as first positional argument to _setup, which forwards it to pydantic_settings.BaseSettings.__init__. Its first positional parameter is _case_sensitive, so the environment source matched the upper case variable names against the lower case field names and returned nothing. Environment settings only took effect after the next configuration setup. * docs: changelog for config priority fixes * fix(config): preserve device identities and storage costs during migration * fix(devices): preserve charge-rate typing and public import compatibility * ruff format fix * fix(config): satisfy typed device conversion and migration contracts * docs(config): refresh validated configuration prerequisite schemas --------- Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com> Co-authored-by: Bobby Noelte <b0661n0e17e@gmail.com> Co-authored-by: r0b2g1t <r0b2g1t@users.noreply.github.com>
1998 lines
77 KiB
Python
1998 lines
77 KiB
Python
"""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)",
|
||
)
|