mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-05-28 18:43:40 +00:00
31fca73ccd
When sequential: true, tasks are placed in declaration order instead of being sorted by duration. Each task's search window starts after the previous task ends (plus gap_minutes). If a task cannot be placed, all subsequent tasks in the chain are also marked unscheduled. Adds 12 tests covering ordering, chaining, gap enforcement, and chain-breaking behavior. Impact: Users can now schedule dependent appliances (e.g., washing machine → dryer) in a single find_cheapest_schedule call with guaranteed order, instead of chaining two find_cheapest_block calls.
355 lines
12 KiB
Python
355 lines
12 KiB
Python
"""Tests for sequential scheduling feature in find_cheapest_schedule."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any, cast
|
||
|
||
from custom_components.tibber_prices.services.find_cheapest_schedule import (
|
||
FIND_CHEAPEST_SCHEDULE_SERVICE_SCHEMA,
|
||
_attempt_schedule,
|
||
)
|
||
|
||
|
||
def _make_intervals(
|
||
prices: list[float],
|
||
start: datetime | None = None,
|
||
*,
|
||
level: str = "NORMAL",
|
||
) -> list[dict[str, Any]]:
|
||
"""Create contiguous quarter-hour intervals for tests."""
|
||
base = start or datetime(2026, 1, 1, 0, 0, tzinfo=UTC)
|
||
return [
|
||
{
|
||
"startsAt": (base + timedelta(minutes=15 * i)).isoformat(),
|
||
"total": price,
|
||
"level": level,
|
||
}
|
||
for i, price in enumerate(prices)
|
||
]
|
||
|
||
|
||
def _make_tasks(*specs: tuple[str, int]) -> list[dict[str, Any]]:
|
||
"""Create task dicts from (name, duration_intervals) tuples."""
|
||
return [
|
||
{
|
||
"name": name,
|
||
"duration_minutes_requested": dur * 15,
|
||
"duration_minutes": dur * 15,
|
||
"duration_intervals": dur,
|
||
"power_profile": None,
|
||
}
|
||
for name, dur in specs
|
||
]
|
||
|
||
|
||
class TestSequentialSchema:
|
||
"""Schema accepts sequential parameter."""
|
||
|
||
def test_schema_accepts_sequential_true(self) -> None:
|
||
"""Schema should accept sequential: true."""
|
||
result = cast(
|
||
"dict[str, Any]",
|
||
FIND_CHEAPEST_SCHEDULE_SERVICE_SCHEMA(
|
||
{
|
||
"tasks": [{"name": "dishwasher", "duration": timedelta(hours=1)}],
|
||
"sequential": True,
|
||
}
|
||
),
|
||
)
|
||
assert result["sequential"] is True
|
||
|
||
def test_schema_defaults_sequential_false(self) -> None:
|
||
"""Sequential should default to false when omitted."""
|
||
result = cast(
|
||
"dict[str, Any]",
|
||
FIND_CHEAPEST_SCHEDULE_SERVICE_SCHEMA(
|
||
{
|
||
"tasks": [{"name": "dishwasher", "duration": timedelta(hours=1)}],
|
||
}
|
||
),
|
||
)
|
||
assert result["sequential"] is False
|
||
|
||
|
||
class TestSequentialOrdering:
|
||
"""Sequential mode preserves declaration order and chains search windows."""
|
||
|
||
def test_non_sequential_sorts_by_duration(self) -> None:
|
||
"""Default (non-sequential) mode sorts tasks longest-first."""
|
||
# 16 intervals = 4 hours of data
|
||
# Prices: first 8 cheap, last 8 expensive
|
||
prices = [5.0] * 8 + [20.0] * 8
|
||
pool = _make_intervals(prices)
|
||
|
||
# Task A is short (2 intervals), Task B is long (4 intervals)
|
||
tasks = _make_tasks(("short_a", 2), ("long_b", 4))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=False,
|
||
)
|
||
|
||
assert not unscheduled
|
||
assert len(assignments) == 2
|
||
# Greedy longest-first: long_b gets placed first (cheapest window)
|
||
# Assignments are returned in placement order (longest first)
|
||
assert assignments[0]["name"] == "long_b"
|
||
assert assignments[1]["name"] == "short_a"
|
||
|
||
def test_sequential_preserves_declaration_order(self) -> None:
|
||
"""Sequential mode places tasks in the order they appear."""
|
||
# 16 intervals, all same price
|
||
prices = [10.0] * 16
|
||
pool = _make_intervals(prices)
|
||
|
||
# Declare short task first, long task second
|
||
tasks = _make_tasks(("short_a", 2), ("long_b", 4))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
|
||
assert not unscheduled
|
||
assert len(assignments) == 2
|
||
# Sequential: short_a placed first, long_b after
|
||
assert assignments[0]["name"] == "short_a"
|
||
assert assignments[1]["name"] == "long_b"
|
||
|
||
def test_sequential_chains_search_windows(self) -> None:
|
||
"""Each sequential task starts after the previous task's end."""
|
||
# 12 intervals: first 4 are cheap, next 4 medium, last 4 expensive
|
||
prices = [5.0] * 4 + [10.0] * 4 + [20.0] * 4
|
||
pool = _make_intervals(prices)
|
||
|
||
# Two tasks of 3 intervals each
|
||
tasks = _make_tasks(("task_a", 3), ("task_b", 3))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
|
||
assert not unscheduled
|
||
assert len(assignments) == 2
|
||
|
||
# Task A should get the cheapest window (intervals 0-2)
|
||
a_end_last = datetime.fromisoformat(assignments[0]["intervals"][-1]["startsAt"])
|
||
|
||
# Task B must start at or after task A's end
|
||
b_start = datetime.fromisoformat(assignments[1]["intervals"][0]["startsAt"])
|
||
assert b_start >= a_end_last + timedelta(minutes=15)
|
||
|
||
def test_sequential_respects_gap(self) -> None:
|
||
"""Sequential mode enforces gap between tasks."""
|
||
# 16 intervals of uniform price
|
||
prices = [10.0] * 16
|
||
pool = _make_intervals(prices)
|
||
|
||
# 2 tasks of 3 intervals each, with 2-interval (30 min) gap
|
||
tasks = _make_tasks(("washer", 3), ("dryer", 3))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=2,
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
|
||
assert not unscheduled
|
||
assert len(assignments) == 2
|
||
|
||
washer_end = datetime.fromisoformat(assignments[0]["intervals"][-1]["startsAt"]) + timedelta(minutes=15)
|
||
dryer_start = datetime.fromisoformat(assignments[1]["intervals"][0]["startsAt"])
|
||
|
||
# Gap should be at least 30 minutes (2 intervals × 15 min)
|
||
gap = dryer_start - washer_end
|
||
assert gap >= timedelta(minutes=30)
|
||
|
||
def test_sequential_chain_breaks_on_failure(self) -> None:
|
||
"""If a sequential task can't be placed, all later tasks are unscheduled."""
|
||
# Only 6 intervals — not enough for 3 tasks of 3 intervals each
|
||
prices = [10.0] * 6
|
||
pool = _make_intervals(prices)
|
||
|
||
tasks = _make_tasks(("task_a", 3), ("task_b", 3), ("task_c", 3))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
|
||
# Task A and B fit (6 intervals total), task C doesn't
|
||
assert len(assignments) == 2
|
||
assert assignments[0]["name"] == "task_a"
|
||
assert assignments[1]["name"] == "task_b"
|
||
assert unscheduled == ["task_c"]
|
||
|
||
def test_sequential_all_fail_after_first_failure(self) -> None:
|
||
"""If the first task fails in sequential mode, all are unscheduled."""
|
||
# 2 intervals — not enough for any 3-interval task
|
||
prices = [10.0] * 2
|
||
pool = _make_intervals(prices)
|
||
|
||
tasks = _make_tasks(("task_a", 3), ("task_b", 2))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
|
||
# Task A can't fit (needs 3, only 2 available)
|
||
# Task B should also be unscheduled because the chain is broken
|
||
assert len(assignments) == 0
|
||
assert unscheduled == ["task_a", "task_b"]
|
||
|
||
def test_sequential_optimizes_within_window(self) -> None:
|
||
"""Sequential still finds cheapest window within each task's available range."""
|
||
# 12 intervals: pattern cheap-expensive-cheap-expensive...
|
||
# First 6 for task A, second 6 for task B
|
||
# Within each half, there's a cheaper sub-window
|
||
prices = [20.0, 5.0, 5.0, 20.0, 20.0, 20.0, 20.0, 20.0, 5.0, 5.0, 20.0, 20.0]
|
||
pool = _make_intervals(prices)
|
||
|
||
tasks = _make_tasks(("task_a", 2), ("task_b", 2))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
|
||
assert not unscheduled
|
||
assert len(assignments) == 2
|
||
|
||
# Task A should pick the cheapest 2-interval window: indices 1-2 (price 5.0 each)
|
||
a_prices = [iv["total"] for iv in assignments[0]["intervals"]]
|
||
assert a_prices == [5.0, 5.0]
|
||
|
||
# Task B searches from index 2 onward, cheapest is indices 8-9 (price 5.0 each)
|
||
b_prices = [iv["total"] for iv in assignments[1]["intervals"]]
|
||
assert b_prices == [5.0, 5.0]
|
||
|
||
def test_sequential_single_task_same_as_non_sequential(self) -> None:
|
||
"""With a single task, sequential and non-sequential produce the same result."""
|
||
prices = [20.0, 5.0, 5.0, 20.0, 10.0, 10.0]
|
||
pool = _make_intervals(prices)
|
||
tasks = _make_tasks(("only_task", 2))
|
||
|
||
a_seq, u_seq, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
a_non, u_non, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=False,
|
||
)
|
||
|
||
assert not u_seq
|
||
assert not u_non
|
||
assert len(a_seq) == len(a_non) == 1
|
||
assert a_seq[0]["intervals"] == a_non[0]["intervals"]
|
||
|
||
|
||
class TestSequentialThreeTasks:
|
||
"""Sequential scheduling with three tasks (washer → dryer → fold reminder)."""
|
||
|
||
def test_three_tasks_chained(self) -> None:
|
||
"""Three sequential tasks are placed in order with no overlap."""
|
||
# 24 intervals (6 hours) with varying prices
|
||
prices = [15.0, 10.0, 5.0, 5.0, 10.0, 15.0, 20.0, 25.0] * 3
|
||
pool = _make_intervals(prices)
|
||
|
||
tasks = _make_tasks(("washer", 4), ("dryer", 3), ("fold", 1))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=0,
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
|
||
assert not unscheduled
|
||
assert len(assignments) == 3
|
||
assert assignments[0]["name"] == "washer"
|
||
assert assignments[1]["name"] == "dryer"
|
||
assert assignments[2]["name"] == "fold"
|
||
|
||
# Verify no overlap: each task starts after previous ends
|
||
for i in range(1, len(assignments)):
|
||
prev_end = datetime.fromisoformat(assignments[i - 1]["intervals"][-1]["startsAt"]) + timedelta(minutes=15)
|
||
curr_start = datetime.fromisoformat(assignments[i]["intervals"][0]["startsAt"])
|
||
assert curr_start >= prev_end
|
||
|
||
def test_three_tasks_with_gap(self) -> None:
|
||
"""Three sequential tasks respect gap between each pair."""
|
||
prices = [10.0] * 24
|
||
pool = _make_intervals(prices)
|
||
|
||
tasks = _make_tasks(("washer", 4), ("dryer", 3), ("fold", 1))
|
||
|
||
assignments, unscheduled, _ = _attempt_schedule(
|
||
pool,
|
||
max_price_level=None,
|
||
min_price_level=None,
|
||
tasks=tasks,
|
||
gap_intervals=1, # 15 min gap
|
||
smooth_outliers=False,
|
||
sequential=True,
|
||
)
|
||
|
||
assert not unscheduled
|
||
assert len(assignments) == 3
|
||
|
||
# Verify gaps between each pair
|
||
for i in range(1, len(assignments)):
|
||
prev_end = datetime.fromisoformat(assignments[i - 1]["intervals"][-1]["startsAt"]) + timedelta(minutes=15)
|
||
curr_start = datetime.fromisoformat(assignments[i]["intervals"][0]["startsAt"])
|
||
gap = curr_start - prev_end
|
||
assert gap >= timedelta(minutes=15)
|