mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-07-27 17:26:48 +00:00
fix(services): prevent plan_charging overcharge from segment-constraint bridging
apply_segment_constraints could add far more grid energy than requested when max_cycles_per_day or min_charge_duration_minutes forced bridging across expensive gaps between cheap intervals. With max_cycles_per_day=1, isolated cheap intervals were merged into one continuous segment by filling every gap in between, without ever trimming the surplus back down, resulting in achieved_soc_percent far above 100%. Add a post-bridging trim step that removes segment-edge intervals (highest price first) until total grid energy matches the requested target again, while still respecting max_cycles_per_day/min_charge_duration_minutes and never dropping below the target itself (fixed-power mode's expected last-interval rounding overshoot is preserved). A related edge case is also fixed: trimming could previously remove an interval required to satisfy a must_reach_by deadline, silently flipping deadline_met to False even though the overall energy target was still reached. Deadline-critical intervals are now passed through as protected_starts and are never removed during trimming. Fixes #167 Impact: plan_charging no longer overcharges the battery/EV past the requested target SoC when max_cycles_per_day or min_charge_duration_minutes is set, and must_reach_by deadlines are honored even when those constraints require bridging across expensive price gaps.
This commit is contained in:
Julian Pawlowski
parent
e94130953f
commit
b8e40bfa3b
4 changed files with 430 additions and 98 deletions
|
|
@ -218,6 +218,245 @@ def _add_interval_if_available(
|
|||
return True
|
||||
|
||||
|
||||
def _constraints_satisfied(
|
||||
intervals: list[dict[str, Any]],
|
||||
*,
|
||||
max_cycles_per_day: int | None,
|
||||
min_charge_duration_minutes: int | None,
|
||||
interval_minutes: int,
|
||||
) -> bool:
|
||||
"""Check whether current interval selection satisfies active constraints."""
|
||||
grouped_segments = group_intervals_into_segments(intervals)
|
||||
|
||||
if max_cycles_per_day and len(grouped_segments) > max_cycles_per_day:
|
||||
return False
|
||||
|
||||
if min_charge_duration_minutes:
|
||||
required_intervals = max(1, math.ceil(min_charge_duration_minutes / interval_minutes))
|
||||
if any(segment["interval_count"] < required_intervals for segment in grouped_segments):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def _extend_for_min_duration(
|
||||
selected_map: dict[str, dict[str, Any]],
|
||||
*,
|
||||
candidate_map: dict[str, dict[str, Any]],
|
||||
candidates_sorted: list[dict[str, Any]],
|
||||
candidate_index: dict[str, int],
|
||||
minimum_power_w: int,
|
||||
charging_efficiency: float,
|
||||
interval_minutes: int,
|
||||
min_charge_duration_minutes: int,
|
||||
warnings: list[str],
|
||||
) -> None:
|
||||
"""Extend short segments by adding contiguous neighbor intervals."""
|
||||
required_intervals = max(1, math.ceil(min_charge_duration_minutes / interval_minutes))
|
||||
progress = True
|
||||
|
||||
while progress:
|
||||
progress = False
|
||||
selected_intervals = sorted(selected_map.values(), key=_interval_start)
|
||||
segments = group_intervals_into_segments(selected_intervals)
|
||||
|
||||
for segment in segments:
|
||||
if segment["interval_count"] >= required_intervals:
|
||||
continue
|
||||
|
||||
while segment["interval_count"] < required_intervals:
|
||||
first = segment["intervals"][0]["startsAt"]
|
||||
last = segment["intervals"][-1]["startsAt"]
|
||||
first_index = candidate_index[first]
|
||||
last_index = candidate_index[last]
|
||||
|
||||
prev_interval = candidates_sorted[first_index - 1] if first_index > 0 else None
|
||||
next_interval = candidates_sorted[last_index + 1] if last_index + 1 < len(candidates_sorted) else None
|
||||
|
||||
options: list[dict[str, Any]] = []
|
||||
if (
|
||||
prev_interval is not None
|
||||
and _interval_start(candidate_map[first]) - _interval_start(prev_interval)
|
||||
== timedelta(minutes=interval_minutes)
|
||||
and prev_interval["startsAt"] not in selected_map
|
||||
):
|
||||
options.append(prev_interval)
|
||||
|
||||
if (
|
||||
next_interval is not None
|
||||
and _interval_start(next_interval) - _interval_start(candidate_map[last])
|
||||
== timedelta(minutes=interval_minutes)
|
||||
and next_interval["startsAt"] not in selected_map
|
||||
):
|
||||
options.append(next_interval)
|
||||
|
||||
if not options:
|
||||
warnings.append("min_charge_duration_unreachable")
|
||||
break
|
||||
|
||||
cheapest = min(options, key=lambda interval: (_sort_price(interval), _interval_start(interval)))
|
||||
added = _add_interval_if_available(
|
||||
selected_map,
|
||||
candidate_map,
|
||||
cheapest["startsAt"],
|
||||
power_w=minimum_power_w,
|
||||
charging_efficiency=charging_efficiency,
|
||||
interval_minutes=interval_minutes,
|
||||
)
|
||||
if not added:
|
||||
break
|
||||
|
||||
progress = True
|
||||
selected_intervals = sorted(selected_map.values(), key=_interval_start)
|
||||
segment = next(
|
||||
seg
|
||||
for seg in group_intervals_into_segments(selected_intervals)
|
||||
if first in {iv["startsAt"] for iv in seg["intervals"]}
|
||||
)
|
||||
|
||||
|
||||
def _merge_for_max_cycles(
|
||||
selected_map: dict[str, dict[str, Any]],
|
||||
*,
|
||||
candidate_map: dict[str, dict[str, Any]],
|
||||
candidates_sorted: list[dict[str, Any]],
|
||||
candidate_index: dict[str, int],
|
||||
minimum_power_w: int,
|
||||
charging_efficiency: float,
|
||||
interval_minutes: int,
|
||||
max_cycles_per_day: int,
|
||||
warnings: list[str],
|
||||
) -> None:
|
||||
"""Bridge cheapest gaps until the cycle limit is satisfied."""
|
||||
while True:
|
||||
selected_intervals = sorted(selected_map.values(), key=_interval_start)
|
||||
segments = group_intervals_into_segments(selected_intervals)
|
||||
if len(segments) <= max_cycles_per_day:
|
||||
break
|
||||
|
||||
best_gap: tuple[float, list[dict[str, Any]]] | None = None
|
||||
for left, right in pairwise(segments):
|
||||
left_end_index = candidate_index[left["intervals"][-1]["startsAt"]]
|
||||
right_start_index = candidate_index[right["intervals"][0]["startsAt"]]
|
||||
gap = candidates_sorted[left_end_index + 1 : right_start_index]
|
||||
|
||||
if not gap:
|
||||
continue
|
||||
if any(interval["startsAt"] in selected_map for interval in gap):
|
||||
continue
|
||||
if any(
|
||||
_interval_start(gap[index + 1]) - _interval_start(gap[index]) != timedelta(minutes=interval_minutes)
|
||||
for index in range(len(gap) - 1)
|
||||
):
|
||||
continue
|
||||
|
||||
penalty = sum(_sort_price(interval) for interval in gap)
|
||||
if best_gap is None or penalty < best_gap[0]:
|
||||
best_gap = (penalty, gap)
|
||||
|
||||
if best_gap is None:
|
||||
warnings.append("max_cycles_unreachable")
|
||||
break
|
||||
|
||||
for interval in best_gap[1]:
|
||||
_add_interval_if_available(
|
||||
selected_map,
|
||||
candidate_map,
|
||||
interval["startsAt"],
|
||||
power_w=minimum_power_w,
|
||||
charging_efficiency=charging_efficiency,
|
||||
interval_minutes=interval_minutes,
|
||||
)
|
||||
|
||||
|
||||
def _collect_removable_edge_indices(
|
||||
selected_intervals: list[dict[str, Any]],
|
||||
*,
|
||||
total_grid_energy: float,
|
||||
target_grid_energy_kwh: float,
|
||||
max_cycles_per_day: int | None,
|
||||
min_charge_duration_minutes: int | None,
|
||||
interval_minutes: int,
|
||||
protected_starts: frozenset[str] | None,
|
||||
) -> list[int]:
|
||||
"""Return edge interval indices that can be removed while keeping constraints valid."""
|
||||
removable_indices: list[int] = []
|
||||
segments = group_intervals_into_segments(selected_intervals)
|
||||
|
||||
for segment in segments:
|
||||
first_start = segment["intervals"][0]["startsAt"]
|
||||
last_start = segment["intervals"][-1]["startsAt"]
|
||||
|
||||
for edge_start in (first_start, last_start):
|
||||
if protected_starts is not None and edge_start in protected_starts:
|
||||
continue
|
||||
|
||||
edge_index = next(
|
||||
(index for index, interval in enumerate(selected_intervals) if interval["startsAt"] == edge_start),
|
||||
None,
|
||||
)
|
||||
if edge_index is None or edge_index in removable_indices:
|
||||
continue
|
||||
|
||||
candidate = selected_intervals[edge_index]
|
||||
new_total = total_grid_energy - float(candidate["grid_energy_kwh"])
|
||||
if new_total + _INTERVAL_TOLERANCE < target_grid_energy_kwh:
|
||||
continue
|
||||
|
||||
new_selection = selected_intervals[:edge_index] + selected_intervals[edge_index + 1 :]
|
||||
if not new_selection:
|
||||
continue
|
||||
if not _constraints_satisfied(
|
||||
new_selection,
|
||||
max_cycles_per_day=max_cycles_per_day,
|
||||
min_charge_duration_minutes=min_charge_duration_minutes,
|
||||
interval_minutes=interval_minutes,
|
||||
):
|
||||
continue
|
||||
|
||||
removable_indices.append(edge_index)
|
||||
|
||||
return removable_indices
|
||||
|
||||
|
||||
def _trim_to_target_energy(
|
||||
selected_map: dict[str, dict[str, Any]],
|
||||
*,
|
||||
target_grid_energy_kwh: float,
|
||||
max_cycles_per_day: int | None,
|
||||
min_charge_duration_minutes: int | None,
|
||||
interval_minutes: int,
|
||||
protected_starts: frozenset[str] | None = None,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
"""Trim excess energy from selection by removing expensive edge intervals first.
|
||||
|
||||
Intervals whose ``startsAt`` is listed in ``protected_starts`` (for example, intervals
|
||||
required to satisfy a ``must_reach_by`` deadline) are never removed, even if that means
|
||||
the target energy cannot be fully reached through trimming alone.
|
||||
"""
|
||||
selected_intervals = sorted(selected_map.values(), key=_interval_start)
|
||||
total_grid_energy = sum(float(interval["grid_energy_kwh"]) for interval in selected_intervals)
|
||||
|
||||
while selected_intervals and total_grid_energy > target_grid_energy_kwh + _INTERVAL_TOLERANCE:
|
||||
removable_indices = _collect_removable_edge_indices(
|
||||
selected_intervals,
|
||||
total_grid_energy=total_grid_energy,
|
||||
target_grid_energy_kwh=target_grid_energy_kwh,
|
||||
max_cycles_per_day=max_cycles_per_day,
|
||||
min_charge_duration_minutes=min_charge_duration_minutes,
|
||||
interval_minutes=interval_minutes,
|
||||
protected_starts=protected_starts,
|
||||
)
|
||||
if not removable_indices:
|
||||
break
|
||||
|
||||
best_index = max(removable_indices, key=lambda index: _sort_price(selected_intervals[index]))
|
||||
total_grid_energy -= float(selected_intervals[best_index]["grid_energy_kwh"])
|
||||
del selected_intervals[best_index]
|
||||
|
||||
return {interval["startsAt"]: interval for interval in selected_intervals}
|
||||
|
||||
|
||||
def apply_segment_constraints(
|
||||
schedule: dict[str, Any],
|
||||
candidate_intervals: list[dict[str, Any]],
|
||||
|
|
@ -225,9 +464,15 @@ def apply_segment_constraints(
|
|||
charging_efficiency: float,
|
||||
min_charge_duration_minutes: int | None = None,
|
||||
max_cycles_per_day: int | None = None,
|
||||
target_grid_energy_kwh: float | None = None,
|
||||
protected_starts: frozenset[str] | None = None,
|
||||
interval_minutes: int = 15,
|
||||
) -> tuple[dict[str, Any], list[str]]:
|
||||
"""Extend/bridge selected intervals to satisfy segment duration and cycle constraints."""
|
||||
"""Extend/bridge selected intervals to satisfy segment duration and cycle constraints.
|
||||
|
||||
``protected_starts`` marks intervals (by ``startsAt``) that must never be removed while
|
||||
trimming to ``target_grid_energy_kwh``, e.g. intervals required to meet a deadline.
|
||||
"""
|
||||
warnings: list[str] = []
|
||||
selected_map = {interval["startsAt"]: dict(interval) for interval in schedule["intervals"]}
|
||||
candidate_map = {interval["startsAt"]: interval for interval in candidate_intervals}
|
||||
|
|
@ -236,103 +481,40 @@ def apply_segment_constraints(
|
|||
minimum_power_w = int(schedule["minimum_power_w"])
|
||||
|
||||
if min_charge_duration_minutes:
|
||||
required_intervals = max(1, math.ceil(min_charge_duration_minutes / interval_minutes))
|
||||
progress = True
|
||||
while progress:
|
||||
progress = False
|
||||
selected_intervals = sorted(selected_map.values(), key=_interval_start)
|
||||
segments = group_intervals_into_segments(selected_intervals)
|
||||
for segment in segments:
|
||||
if segment["interval_count"] >= required_intervals:
|
||||
continue
|
||||
while segment["interval_count"] < required_intervals:
|
||||
first = segment["intervals"][0]["startsAt"]
|
||||
last = segment["intervals"][-1]["startsAt"]
|
||||
first_index = candidate_index[first]
|
||||
last_index = candidate_index[last]
|
||||
|
||||
prev_interval = candidates_sorted[first_index - 1] if first_index > 0 else None
|
||||
next_interval = (
|
||||
candidates_sorted[last_index + 1] if last_index + 1 < len(candidates_sorted) else None
|
||||
)
|
||||
|
||||
prev_contiguous = False
|
||||
next_contiguous = False
|
||||
if prev_interval is not None:
|
||||
prev_contiguous = _interval_start(candidate_map[first]) - _interval_start(
|
||||
prev_interval
|
||||
) == timedelta(minutes=interval_minutes)
|
||||
if next_interval is not None:
|
||||
next_contiguous = _interval_start(next_interval) - _interval_start(
|
||||
candidate_map[last]
|
||||
) == timedelta(minutes=interval_minutes)
|
||||
|
||||
options: list[dict[str, Any]] = []
|
||||
if prev_interval is not None and prev_contiguous and prev_interval["startsAt"] not in selected_map:
|
||||
options.append(prev_interval)
|
||||
if next_interval is not None and next_contiguous and next_interval["startsAt"] not in selected_map:
|
||||
options.append(next_interval)
|
||||
if not options:
|
||||
warnings.append("min_charge_duration_unreachable")
|
||||
break
|
||||
|
||||
cheapest = min(options, key=lambda interval: (_sort_price(interval), _interval_start(interval)))
|
||||
added = _add_interval_if_available(
|
||||
selected_map,
|
||||
candidate_map,
|
||||
cheapest["startsAt"],
|
||||
power_w=minimum_power_w,
|
||||
charging_efficiency=charging_efficiency,
|
||||
interval_minutes=interval_minutes,
|
||||
)
|
||||
if not added:
|
||||
break
|
||||
progress = True
|
||||
selected_intervals = sorted(selected_map.values(), key=_interval_start)
|
||||
segment = next(
|
||||
seg
|
||||
for seg in group_intervals_into_segments(selected_intervals)
|
||||
if first in {iv["startsAt"] for iv in seg["intervals"]}
|
||||
)
|
||||
_extend_for_min_duration(
|
||||
selected_map,
|
||||
candidate_map=candidate_map,
|
||||
candidates_sorted=candidates_sorted,
|
||||
candidate_index=candidate_index,
|
||||
minimum_power_w=minimum_power_w,
|
||||
charging_efficiency=charging_efficiency,
|
||||
interval_minutes=interval_minutes,
|
||||
min_charge_duration_minutes=min_charge_duration_minutes,
|
||||
warnings=warnings,
|
||||
)
|
||||
|
||||
if max_cycles_per_day:
|
||||
while True:
|
||||
selected_intervals = sorted(selected_map.values(), key=_interval_start)
|
||||
segments = group_intervals_into_segments(selected_intervals)
|
||||
if len(segments) <= max_cycles_per_day:
|
||||
break
|
||||
_merge_for_max_cycles(
|
||||
selected_map,
|
||||
candidate_map=candidate_map,
|
||||
candidates_sorted=candidates_sorted,
|
||||
candidate_index=candidate_index,
|
||||
minimum_power_w=minimum_power_w,
|
||||
charging_efficiency=charging_efficiency,
|
||||
interval_minutes=interval_minutes,
|
||||
max_cycles_per_day=max_cycles_per_day,
|
||||
warnings=warnings,
|
||||
)
|
||||
|
||||
best_gap: tuple[float, list[dict[str, Any]]] | None = None
|
||||
for left, right in pairwise(segments):
|
||||
left_end_index = candidate_index[left["intervals"][-1]["startsAt"]]
|
||||
right_start_index = candidate_index[right["intervals"][0]["startsAt"]]
|
||||
gap = candidates_sorted[left_end_index + 1 : right_start_index]
|
||||
if not gap:
|
||||
continue
|
||||
if any(interval["startsAt"] in selected_map for interval in gap):
|
||||
continue
|
||||
if any(
|
||||
_interval_start(gap[index + 1]) - _interval_start(gap[index]) != timedelta(minutes=interval_minutes)
|
||||
for index in range(len(gap) - 1)
|
||||
):
|
||||
continue
|
||||
penalty = sum(_sort_price(interval) for interval in gap)
|
||||
if best_gap is None or penalty < best_gap[0]:
|
||||
best_gap = (penalty, gap)
|
||||
|
||||
if best_gap is None:
|
||||
warnings.append("max_cycles_unreachable")
|
||||
break
|
||||
|
||||
for interval in best_gap[1]:
|
||||
_add_interval_if_available(
|
||||
selected_map,
|
||||
candidate_map,
|
||||
interval["startsAt"],
|
||||
power_w=minimum_power_w,
|
||||
charging_efficiency=charging_efficiency,
|
||||
interval_minutes=interval_minutes,
|
||||
)
|
||||
if target_grid_energy_kwh is not None:
|
||||
selected_map = _trim_to_target_energy(
|
||||
selected_map,
|
||||
target_grid_energy_kwh=target_grid_energy_kwh,
|
||||
max_cycles_per_day=max_cycles_per_day,
|
||||
min_charge_duration_minutes=min_charge_duration_minutes,
|
||||
interval_minutes=interval_minutes,
|
||||
protected_starts=protected_starts,
|
||||
)
|
||||
|
||||
selected_intervals = sorted(selected_map.values(), key=_interval_start)
|
||||
segments = group_intervals_into_segments(selected_intervals)
|
||||
|
|
|
|||
|
|
@ -576,12 +576,23 @@ def _attempt_plan(
|
|||
if schedule["unallocated_grid_energy_kwh"] > 1e-6:
|
||||
return None, "energy_unreachable"
|
||||
|
||||
# Deadline-critical intervals (selected to satisfy must_reach_soc by the deadline) must
|
||||
# never be dropped by segment-constraint trimming, or deadline_met could silently become
|
||||
# False even though the overall energy target is still satisfied.
|
||||
protected_starts = (
|
||||
frozenset(interval["startsAt"] for interval in schedule["pre_deadline"]["intervals"])
|
||||
if "pre_deadline" in schedule
|
||||
else None
|
||||
)
|
||||
|
||||
schedule, warnings = apply_segment_constraints(
|
||||
schedule,
|
||||
candidates,
|
||||
charging_efficiency=ctx.charging_efficiency,
|
||||
min_charge_duration_minutes=ctx.min_charge_duration_minutes,
|
||||
max_cycles_per_day=ctx.max_cycles_per_day,
|
||||
target_grid_energy_kwh=effective_energy_needed_grid_kwh,
|
||||
protected_starts=protected_starts,
|
||||
interval_minutes=INTERVAL_MINUTES,
|
||||
)
|
||||
scheduled_intervals = build_soc_progression_from_schedule(
|
||||
|
|
|
|||
|
|
@ -203,6 +203,60 @@ async def test_plan_charging_can_meet_deadline_before_peak_period(monkeypatch: p
|
|||
assert response["deadline"]["achieved_soc_kwh"] >= 4.0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_plan_charging_max_cycles_does_not_break_deadline(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Cycle merging/trimming must never discard intervals required to meet a deadline.
|
||||
|
||||
Regression test for a bug where ``apply_segment_constraints`` could remove the
|
||||
deadline-critical interval while trimming excess energy added by cycle bridging,
|
||||
silently turning ``deadline_met`` false even though the overall target was reached.
|
||||
"""
|
||||
intervals = _make_intervals([0.80, 0.95, 0.95, 0.05])
|
||||
fake_tuple = _build_fake_entry_and_coordinator(intervals)
|
||||
|
||||
monkeypatch.setattr(charging_module, "get_entry_and_data", lambda _hass, _entry_id: fake_tuple)
|
||||
monkeypatch.setattr(charging_module, "resolve_home_timezone", lambda _coord, _home_id: "UTC")
|
||||
monkeypatch.setattr(
|
||||
charging_module,
|
||||
"resolve_search_range",
|
||||
lambda _call_data, _now, _home_tz: (
|
||||
datetime(2026, 1, 1, 0, 0, tzinfo=UTC),
|
||||
datetime(2026, 1, 1, 1, 0, tzinfo=UTC),
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(charging_module, "get_display_unit_factor", lambda _entry: 1)
|
||||
monkeypatch.setattr(charging_module, "get_display_unit_string", lambda _entry, _currency: "EUR/kWh")
|
||||
monkeypatch.setattr(charging_module.dt_util, "now", lambda: datetime(2026, 1, 1, 0, 0, tzinfo=UTC))
|
||||
|
||||
call = SimpleNamespace(
|
||||
hass=object(),
|
||||
data={
|
||||
"battery_capacity_kwh": 10.0,
|
||||
"current_soc_percent": 20.0,
|
||||
"target_soc_percent": 40.0,
|
||||
"must_reach_soc_percent": 30.0,
|
||||
"must_reach_by": datetime(2026, 1, 1, 0, 15, tzinfo=UTC),
|
||||
"max_charge_power_w": 4000,
|
||||
"max_cycles_per_day": 1,
|
||||
"charging_efficiency": 1.0,
|
||||
"use_base_unit": True,
|
||||
"allow_relaxation": False,
|
||||
},
|
||||
)
|
||||
|
||||
response = cast("dict[str, Any]", await handle_plan_charging(cast("ServiceCall", call)))
|
||||
|
||||
assert response["intervals_found"] is True
|
||||
assert response["charging"]["schedule"]["segment_count"] == 1
|
||||
assert response["deadline"]["deadline_met"] is True
|
||||
assert response["deadline"]["achieved_soc_kwh"] == 3.0
|
||||
assert response["battery"]["target_met"] is True
|
||||
assert response["battery"]["achieved_soc_kwh"] == 4.0
|
||||
|
||||
scheduled = cast("list[dict[str, Any]]", response["charging"]["schedule"]["intervals"])
|
||||
assert scheduled[0]["price"] == 0.8 # the deadline-critical interval must survive trimming
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_plan_charging_can_filter_by_profitability(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Economic filtering should keep only profitable intervals when reserve_for_discharge is enabled."""
|
||||
|
|
@ -292,7 +346,7 @@ async def test_plan_charging_respects_min_charge_duration(monkeypatch: pytest.Mo
|
|||
|
||||
@pytest.mark.asyncio
|
||||
async def test_plan_charging_respects_max_cycles_per_day(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Multiple cheap isolated intervals should be bridged to satisfy max cycle limits."""
|
||||
"""Cycle merging must not overcharge beyond the requested target energy."""
|
||||
intervals = _make_intervals([0.10, 0.80, 0.11, 0.90, 0.12, 0.95, 0.50, 0.60])
|
||||
fake_tuple = _build_fake_entry_and_coordinator(intervals)
|
||||
|
||||
|
|
@ -327,7 +381,10 @@ async def test_plan_charging_respects_max_cycles_per_day(monkeypatch: pytest.Mon
|
|||
|
||||
assert response["intervals_found"] is True
|
||||
assert response["charging"]["schedule"]["segment_count"] == 1
|
||||
assert response["charging"]["total_energy_kwh"] == 3.0
|
||||
assert response["battery"]["achieved_soc_kwh"] == 5.0
|
||||
assert response["battery"]["target_met"] is True
|
||||
|
||||
scheduled = cast("list[dict[str, Any]]", response["charging"]["schedule"]["intervals"])
|
||||
assert [interval["price"] for interval in scheduled[:5]] == [0.1, 0.8, 0.11, 0.9, 0.12]
|
||||
assert [interval["price"] for interval in scheduled] == [0.1, 0.8, 0.11]
|
||||
assert response["warnings"] is None
|
||||
|
|
|
|||
|
|
@ -6,7 +6,10 @@ from datetime import UTC, datetime, timedelta
|
|||
from zoneinfo import ZoneInfo
|
||||
|
||||
from custom_components.tibber_prices.services.charging.deadline_solver import resolve_deadline
|
||||
from custom_components.tibber_prices.services.charging.power_scheduler import build_power_schedule
|
||||
from custom_components.tibber_prices.services.charging.power_scheduler import (
|
||||
apply_segment_constraints,
|
||||
build_power_schedule,
|
||||
)
|
||||
|
||||
|
||||
def _make_intervals(prices: list[float]) -> list[dict[str, object]]:
|
||||
|
|
@ -51,6 +54,85 @@ def test_stepped_mode_uses_smallest_sufficient_step() -> None:
|
|||
assert sorted(interval["power_w"] for interval in result["intervals"]) == [2000, 4000, 4000]
|
||||
|
||||
|
||||
def test_apply_segment_constraints_bridges_and_trims_to_target() -> None:
|
||||
"""Bridging isolated cheap intervals must not leave more energy than requested.
|
||||
|
||||
Direct unit-level regression for the ``max_cycles_per_day`` overcharge bug: bridging
|
||||
fills the gaps between segments with extra (non-essential) intervals, and the
|
||||
subsequent trim step must remove exactly that surplus again.
|
||||
"""
|
||||
candidates = _make_intervals([0.10, 0.80, 0.11, 0.90, 0.12, 0.95, 0.50, 0.60])
|
||||
schedule = build_power_schedule(candidates, 3.0, max_charge_power_w=4000, charging_efficiency=1.0)
|
||||
assert schedule["total_grid_energy_kwh"] == 3.0
|
||||
assert len(schedule["segments"]) == 3 # three isolated cheap intervals
|
||||
|
||||
schedule, warnings = apply_segment_constraints(
|
||||
schedule,
|
||||
candidates,
|
||||
charging_efficiency=1.0,
|
||||
max_cycles_per_day=1,
|
||||
target_grid_energy_kwh=3.0,
|
||||
interval_minutes=15,
|
||||
)
|
||||
|
||||
assert warnings == []
|
||||
assert len(schedule["segments"]) == 1
|
||||
assert schedule["total_grid_energy_kwh"] == 3.0
|
||||
assert [interval["total"] for interval in schedule["intervals"]] == [0.10, 0.80, 0.11]
|
||||
|
||||
|
||||
def test_apply_segment_constraints_never_trims_below_target() -> None:
|
||||
"""Trimming must never drop the plan below the required energy.
|
||||
|
||||
Fixed-power mode cannot reduce the last interval's power, so it can legitimately
|
||||
overshoot the target by up to one interval's energy. This is expected physical
|
||||
behavior (not constraint bridging) and must survive trimming unchanged.
|
||||
"""
|
||||
candidates = _make_intervals([0.10, 0.20])
|
||||
schedule = build_power_schedule(candidates, 1.5, max_charge_power_w=4000, charging_efficiency=1.0)
|
||||
assert schedule["total_grid_energy_kwh"] == 2.0 # rounded up from 1.5
|
||||
|
||||
schedule, warnings = apply_segment_constraints(
|
||||
schedule,
|
||||
candidates,
|
||||
charging_efficiency=1.0,
|
||||
target_grid_energy_kwh=1.5,
|
||||
interval_minutes=15,
|
||||
)
|
||||
|
||||
assert warnings == []
|
||||
assert schedule["total_grid_energy_kwh"] == 2.0 # both intervals kept
|
||||
assert len(schedule["intervals"]) == 2
|
||||
|
||||
|
||||
def test_apply_segment_constraints_respects_protected_starts() -> None:
|
||||
"""Protected intervals (e.g. deadline-critical) must survive trimming.
|
||||
|
||||
Even when a protected interval is the most expensive edge of the merged segment,
|
||||
trimming must skip it and remove the next-best removable edge instead.
|
||||
"""
|
||||
candidates = _make_intervals([0.80, 0.95, 0.95, 0.05])
|
||||
protected_start = candidates[0]["startsAt"]
|
||||
|
||||
schedule = build_power_schedule(candidates, 2.0, max_charge_power_w=4000, charging_efficiency=1.0)
|
||||
assert [interval["total"] for interval in schedule["intervals"]] == [0.80, 0.05]
|
||||
|
||||
schedule, warnings = apply_segment_constraints(
|
||||
schedule,
|
||||
candidates,
|
||||
charging_efficiency=1.0,
|
||||
max_cycles_per_day=1,
|
||||
target_grid_energy_kwh=2.0,
|
||||
protected_starts=frozenset({protected_start}),
|
||||
interval_minutes=15,
|
||||
)
|
||||
|
||||
assert warnings == []
|
||||
assert schedule["total_grid_energy_kwh"] == 2.0
|
||||
starts = {interval["startsAt"] for interval in schedule["intervals"]}
|
||||
assert protected_start in starts
|
||||
|
||||
|
||||
def test_resolve_deadline_next_peak_period() -> None:
|
||||
"""Deadline helper should resolve the next future peak period start."""
|
||||
now = datetime(2026, 1, 1, 0, 0, tzinfo=UTC)
|
||||
|
|
|
|||
Loading…
Reference in a new issue