feat(services): add sequential parameter to find_cheapest_schedule

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.
This commit is contained in:
Julian Pawlowski 2026-04-19 14:17:32 +00:00
parent 0162394263
commit 31fca73ccd
8 changed files with 420 additions and 4 deletions

View file

@ -973,6 +973,11 @@ find_cheapest_schedule:
max: 120 max: 120
unit_of_measurement: min unit_of_measurement: min
mode: box mode: box
sequential:
required: false
default: false
selector:
boolean:
search_scope: search_scope:
required: false required: false
selector: selector:

View file

@ -92,6 +92,7 @@ FIND_CHEAPEST_SCHEDULE_SERVICE_SCHEMA = vol.Schema(
vol.Optional("min_price_level"): vol.In([lvl.lower() for lvl in PRICE_LEVEL_ORDER]), vol.Optional("min_price_level"): vol.In([lvl.lower() for lvl in PRICE_LEVEL_ORDER]),
vol.Optional("include_comparison_details", default=False): cv.boolean, vol.Optional("include_comparison_details", default=False): cv.boolean,
vol.Optional("use_base_unit", default=False): cv.boolean, vol.Optional("use_base_unit", default=False): cv.boolean,
vol.Optional("sequential", default=False): cv.boolean,
vol.Optional("smooth_outliers", default=True): cv.boolean, vol.Optional("smooth_outliers", default=True): cv.boolean,
vol.Optional("allow_relaxation", default=True): cv.boolean, vol.Optional("allow_relaxation", default=True): cv.boolean,
vol.Optional("duration_flexibility_minutes"): vol.All(vol.Coerce(int), vol.Range(min=0, max=120)), vol.Optional("duration_flexibility_minutes"): vol.All(vol.Coerce(int), vol.Range(min=0, max=120)),
@ -230,9 +231,14 @@ def _attempt_schedule(
tasks: list[dict[str, Any]], tasks: list[dict[str, Any]],
gap_intervals: int, gap_intervals: int,
smooth_outliers: bool, smooth_outliers: bool,
sequential: bool = False,
) -> tuple[list[dict[str, Any]], list[str], list[dict[str, Any]]]: ) -> tuple[list[dict[str, Any]], list[str], list[dict[str, Any]]]:
"""Attempt to schedule tasks with specific filter parameters. """Attempt to schedule tasks with specific filter parameters.
When sequential=True, tasks are placed in declaration order and each task's
search window begins after the previous task's end + gap. When False
(default), tasks are sorted longest-first for optimal greedy packing.
Returns: Returns:
(assignments, unscheduled_names, filtered_price_info) (assignments, unscheduled_names, filtered_price_info)
@ -247,18 +253,38 @@ def _attempt_schedule(
if not search_data: if not search_data:
return [], [t["name"] for t in tasks], filtered return [], [t["name"] for t in tasks], filtered
# Greedy assignment: longest task first # Task ordering: declaration order when sequential, longest-first otherwise
tasks_sorted = sorted(tasks, key=lambda t: t["duration_intervals"], reverse=True) tasks_ordered = list(tasks) if sequential else sorted(tasks, key=lambda t: t["duration_intervals"], reverse=True)
available = [True] * len(search_data) available = [True] * len(search_data)
assignments: list[dict[str, Any]] = [] assignments: list[dict[str, Any]] = []
unscheduled: list[str] = [] unscheduled: list[str] = []
for task in tasks_sorted: # In sequential mode, track the earliest allowed start index for the next task
sequential_min_idx = 0
sequential_chain_broken = False
for task in tasks_ordered:
dur_intervals = task["duration_intervals"] dur_intervals = task["duration_intervals"]
# In sequential mode, if the chain is broken (previous task failed),
# all remaining tasks are also unscheduled
if sequential and sequential_chain_broken:
unscheduled.append(task["name"])
continue
# In sequential mode, restrict search to intervals at or after the
# minimum start index by marking earlier slots as unavailable
if sequential and sequential_min_idx > 0:
for k in range(min(sequential_min_idx, len(search_data))):
available[k] = False
window = _find_cheapest_window_in_pool(search_data, dur_intervals, available) window = _find_cheapest_window_in_pool(search_data, dur_intervals, available)
if window is None: if window is None:
unscheduled.append(task["name"]) unscheduled.append(task["name"])
if sequential:
sequential_chain_broken = True
continue continue
start_idx, end_idx = window start_idx, end_idx = window
@ -273,6 +299,10 @@ def _attempt_schedule(
for k in range(start_idx, gap_end): for k in range(start_idx, gap_end):
available[k] = False available[k] = False
# In sequential mode, advance the minimum start for the next task
if sequential:
sequential_min_idx = gap_end
assignments.append( assignments.append(
{ {
"name": task["name"], "name": task["name"],
@ -297,6 +327,7 @@ async def handle_find_cheapest_schedule(call: ServiceCall) -> ServiceResponse:
include_comparison_details: bool = call.data.get("include_comparison_details", False) include_comparison_details: bool = call.data.get("include_comparison_details", False)
smooth_outliers: bool = call.data.get("smooth_outliers", True) smooth_outliers: bool = call.data.get("smooth_outliers", True)
allow_relaxation: bool = call.data.get("allow_relaxation", True) allow_relaxation: bool = call.data.get("allow_relaxation", True)
sequential: bool = call.data.get("sequential", False)
duration_flexibility_minutes: int | None = call.data.get("duration_flexibility_minutes") duration_flexibility_minutes: int | None = call.data.get("duration_flexibility_minutes")
# Validate task names are unique (before any expensive operations) # Validate task names are unique (before any expensive operations)
@ -372,10 +403,11 @@ async def handle_find_cheapest_schedule(call: ServiceCall) -> ServiceResponse:
) )
_LOGGER.info( _LOGGER.info(
"%s called: %d tasks, gap=%dmin, range=%s to %s", "%s called: %d tasks, gap=%dmin, sequential=%s, range=%s to %s",
service_label, service_label,
len(tasks), len(tasks),
gap_minutes, gap_minutes,
sequential,
search_start, search_start,
search_end, search_end,
) )
@ -411,6 +443,7 @@ async def handle_find_cheapest_schedule(call: ServiceCall) -> ServiceResponse:
tasks=tasks, tasks=tasks,
gap_intervals=gap_intervals, gap_intervals=gap_intervals,
smooth_outliers=smooth_outliers, smooth_outliers=smooth_outliers,
sequential=sequential,
) )
all_scheduled = len(unscheduled) == 0 all_scheduled = len(unscheduled) == 0
level_filter_active = min_price_level is not None or max_price_level is not None level_filter_active = min_price_level is not None or max_price_level is not None
@ -437,6 +470,7 @@ async def handle_find_cheapest_schedule(call: ServiceCall) -> ServiceResponse:
tasks=tasks, tasks=tasks,
gap_intervals=gap_intervals, gap_intervals=gap_intervals,
smooth_outliers=smooth_outliers, smooth_outliers=smooth_outliers,
sequential=sequential,
) )
if len(a) > len(best_assignments): if len(a) > len(best_assignments):
best_assignments, best_unscheduled, best_filtered = a, u, f best_assignments, best_unscheduled, best_filtered = a, u, f
@ -474,6 +508,7 @@ async def handle_find_cheapest_schedule(call: ServiceCall) -> ServiceResponse:
tasks=reduced, tasks=reduced,
gap_intervals=gap_intervals, gap_intervals=gap_intervals,
smooth_outliers=smooth_outliers, smooth_outliers=smooth_outliers,
sequential=sequential,
) )
if len(a) > len(best_assignments): if len(a) > len(best_assignments):
best_assignments, best_unscheduled, best_filtered = a, u, f best_assignments, best_unscheduled, best_filtered = a, u, f
@ -584,6 +619,7 @@ async def handle_find_cheapest_schedule(call: ServiceCall) -> ServiceResponse:
"must_finish_by": must_finish_by_dt.isoformat() if must_finish_by_dt else None, "must_finish_by": must_finish_by_dt.isoformat() if must_finish_by_dt else None,
"currency": currency, "currency": currency,
"price_unit": price_unit, "price_unit": price_unit,
"sequential": sequential,
"all_tasks_scheduled": all_scheduled, "all_tasks_scheduled": all_scheduled,
"reason": reason, "reason": reason,
"relaxation_applied": relaxation_applied, "relaxation_applied": relaxation_applied,

View file

@ -2027,6 +2027,10 @@
"name": "Pause zwischen Aufgaben (Minuten)", "name": "Pause zwischen Aufgaben (Minuten)",
"description": "Mindestpause in Minuten zwischen aufeinanderfolgenden Aufgaben. Wird auf 15 Minuten aufgerundet. Standard: 0 (keine Pause)." "description": "Mindestpause in Minuten zwischen aufeinanderfolgenden Aufgaben. Wird auf 15 Minuten aufgerundet. Standard: 0 (keine Pause)."
}, },
"sequential": {
"name": "Sequenzielle Reihenfolge",
"description": "Aufgaben in der Reihenfolge planen, in der sie in der Aufgabenliste stehen. Jede Aufgabe startet nach dem Ende der vorherigen (plus Pause). Nutze dies für abhängige Geräte wie Waschmaschine → Trockner. Standard: deaktiviert (Aufgaben werden nach Dauer sortiert für optimale Verteilung)."
},
"search_scope": { "search_scope": {
"name": "Suchbereich (Shortcut)", "name": "Suchbereich (Shortcut)",
"description": "Kurzwahl für häufige Suchbereiche. Überschreibt alle anderen Zeitbereich-Optionen. today/tomorrow = ganzer Kalendertag, remaining_today = jetzt bis Mitternacht, next_24h/next_48h = rollierendes Fenster ab jetzt." "description": "Kurzwahl für häufige Suchbereiche. Überschreibt alle anderen Zeitbereich-Optionen. today/tomorrow = ganzer Kalendertag, remaining_today = jetzt bis Mitternacht, next_24h/next_48h = rollierendes Fenster ab jetzt."

View file

@ -2027,6 +2027,10 @@
"name": "Gap Between Tasks (minutes)", "name": "Gap Between Tasks (minutes)",
"description": "Minimum gap in minutes between consecutive scheduled tasks. Rounded up to 15 minutes. Default: 0 (no gap)." "description": "Minimum gap in minutes between consecutive scheduled tasks. Rounded up to 15 minutes. Default: 0 (no gap)."
}, },
"sequential": {
"name": "Sequential Ordering",
"description": "Schedule tasks in the order they appear in the task list. Each task starts after the previous one ends (plus gap). Use this for dependent appliances like washing machine → dryer. Default: disabled (tasks are sorted by duration for optimal packing)."
},
"search_scope": { "search_scope": {
"name": "Search Scope", "name": "Search Scope",
"description": "Shorthand for common search ranges. Overrides all other time range options. today / tomorrow = full calendar day, remaining_today = now until midnight, next_24h / next_48h = rolling window from now." "description": "Shorthand for common search ranges. Overrides all other time range options. today / tomorrow = full calendar day, remaining_today = now until midnight, next_24h / next_48h = rolling window from now."

View file

@ -2027,6 +2027,10 @@
"name": "Pause mellom oppgaver (minutter)", "name": "Pause mellom oppgaver (minutter)",
"description": "Minimum pause i minutter mellom paafoeglende planlagte oppgaver. Avrundes opp til 15 minutter. Standard: 0 (ingen pause)." "description": "Minimum pause i minutter mellom paafoeglende planlagte oppgaver. Avrundes opp til 15 minutter. Standard: 0 (ingen pause)."
}, },
"sequential": {
"name": "Sekvensiell rekkefølge",
"description": "Planlegg oppgaver i rekkefølgen de står i oppgavelisten. Hver oppgave starter etter at den forrige er ferdig (pluss pause). Bruk dette for avhengige apparater som vaskemaskin → tørketrommel. Standard: deaktivert (oppgaver sorteres etter varighet for optimal fordeling)."
},
"search_scope": { "search_scope": {
"name": "Soekeomfang (snarvei)", "name": "Soekeomfang (snarvei)",
"description": "Snarvei for vanlige soekeomraader. Overstyrer alle andre tidsalternativer. today/tomorrow = hele kalenderdagen, remaining_today = naa til midnatt, next_24h/next_48h = rullende vindu fra naa." "description": "Snarvei for vanlige soekeomraader. Overstyrer alle andre tidsalternativer. today/tomorrow = hele kalenderdagen, remaining_today = naa til midnatt, next_24h/next_48h = rullende vindu fra naa."

View file

@ -2027,6 +2027,10 @@
"name": "Tussenpose tussen taken (minuten)", "name": "Tussenpose tussen taken (minuten)",
"description": "Minimale tussenpose in minuten tussen opeenvolgende ingeplande taken. Afgerond omhoog tot 15 minuten. Standaard: 0 (geen tussenpose)." "description": "Minimale tussenpose in minuten tussen opeenvolgende ingeplande taken. Afgerond omhoog tot 15 minuten. Standaard: 0 (geen tussenpose)."
}, },
"sequential": {
"name": "Sequentiële volgorde",
"description": "Plan taken in de volgorde waarin ze in de takenlijst staan. Elke taak start na het einde van de vorige (plus tussenpose). Gebruik dit voor afhankelijke apparaten zoals wasmachine → droger. Standaard: uitgeschakeld (taken worden gesorteerd op duur voor optimale verdeling)."
},
"search_scope": { "search_scope": {
"name": "Zoekbereik (snelkoppeling)", "name": "Zoekbereik (snelkoppeling)",
"description": "Snelkoppeling voor veelgebruikte zoekbereiken. Overschrijft alle andere tijdopties. today/tomorrow = volledige kalenderdag, remaining_today = nu tot middernacht, next_24h/next_48h = rolling venster vanaf nu." "description": "Snelkoppeling voor veelgebruikte zoekbereiken. Overschrijft alle andere tijdopties. today/tomorrow = volledige kalenderdag, remaining_today = nu tot middernacht, next_24h/next_48h = rolling venster vanaf nu."

View file

@ -2027,6 +2027,10 @@
"name": "Paus mellan uppgifter (minuter)", "name": "Paus mellan uppgifter (minuter)",
"description": "Minsta paus i minuter mellan paa varandra foeoljande schemalagda uppgifter. Avrundas uppaat till 15 minuter. Standard: 0 (ingen paus)." "description": "Minsta paus i minuter mellan paa varandra foeoljande schemalagda uppgifter. Avrundas uppaat till 15 minuter. Standard: 0 (ingen paus)."
}, },
"sequential": {
"name": "Sekventiell ordning",
"description": "Schemalägg uppgifter i den ordning de står i uppgiftslistan. Varje uppgift startar efter att den föregående är klar (plus paus). Använd detta för beroende apparater som tvättmaskin → torktumlare. Standard: inaktiverad (uppgifter sorteras efter varaktighet för optimal fördelning)."
},
"search_scope": { "search_scope": {
"name": "Soekumfaang (genvaeg)", "name": "Soekumfaang (genvaeg)",
"description": "Genvaeg foer vanliga soekomraaden. Aasidosaetter alla andra tidsalternativ. today/tomorrow = hela kalenderdagen, remaining_today = nu till midnatt, next_24h/next_48h = rullande foenster fraen nu." "description": "Genvaeg foer vanliga soekomraaden. Aasidosaetter alla andra tidsalternativ. today/tomorrow = hela kalenderdagen, remaining_today = nu till midnatt, next_24h/next_48h = rullande foenster fraen nu."

View file

@ -0,0 +1,355 @@
"""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)