diff --git a/apps/predbat/plan.py b/apps/predbat/plan.py index 7f537d16b..357aff869 100644 --- a/apps/predbat/plan.py +++ b/apps/predbat/plan.py @@ -907,6 +907,50 @@ def in_charge_window(self, charge_window, minute_abs): window_n += 1 return -1 + def plan_fragmentation(self, charge_window, charge_limit, export_window, export_limits): + """Count the contiguous active (charge/export) segments in a plan. + + A slot is active if it discharges the battery (export limit < 99, i.e. not freeze/off) or charges it + (charge target above the reserve floor). Time-adjacent active slots of the same mode form one segment; a + time gap or a change of mode (charge<->export) starts a new segment. A cleaner plan has fewer segments, + so this is used as a tie-break to prefer a single export block over a fragmented staircase when the cost + is otherwise equal. + """ + intervals = [] + for window, limit in zip(export_window, export_limits): + if limit < 99: + intervals.append((window["start"], window["end"], "export")) + for window, limit in zip(charge_window, charge_limit): + if limit > self.reserve: + intervals.append((window["start"], window["end"], "charge")) + + intervals.sort(key=lambda item: item[0]) + + segments = 0 + prev_end = None + prev_mode = None + for start, end, mode in intervals: + if prev_end is None or start > prev_end or mode != prev_mode: + segments += 1 + prev_end = end if prev_end is None else max(prev_end, end) + prev_mode = mode + return segments + + def should_replace_plan(self, metric_prev, metric_new, fragmentation_prev, fragmentation_new): + """Decide whether to adopt the freshly optimised plan over the incumbent. + + The new plan is adopted when it is better by at least metric_min_improvement_plan (the existing + anti-jitter behaviour). On a near-tie within that band it is also adopted when it is no worse on cost + and strictly less fragmented, so a cleaner single export block can replace a locked-in split schedule + without churning the plan for tiny cost changes. Lower metric is better, so improvement is prev - new. + """ + improvement = metric_prev - metric_new + if improvement >= self.metric_min_improvement_plan: + return True + if improvement >= 0 and fragmentation_new < fragmentation_prev: + return True + return False + def calculate_plan(self, recompute=True, debug_mode=False, publish=True): """ Calculate the new plan (best) @@ -1179,14 +1223,18 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True): ) self.log("Previous plan best metric is {} (cost {}) and new plan best metric is {} (cost {})".format(dp2(metric_prev), dp2(cost_prev), dp2(metric), dp2(cost))) - if (metric_prev - metric) < self.metric_min_improvement_plan: + fragmentation_prev = self.plan_fragmentation(charge_window_best_prev, charge_limit_best_prev, export_window_best_prev, export_limits_best_prev) + fragmentation_new = self.plan_fragmentation(self.charge_window_best, self.charge_limit_best, self.export_window_best, self.export_limits_best) + if not self.should_replace_plan(metric_prev, metric, fragmentation_prev, fragmentation_new): self.log("New plan metric is not significantly better (metric_min_improvement_plan {}) than previous plan, using previous plan".format(self.metric_min_improvement_plan)) self.charge_window_best = copy.deepcopy(charge_window_best_prev) self.charge_limit_best = copy.deepcopy(charge_limit_best_prev) self.export_window_best = copy.deepcopy(export_window_best_prev) self.export_limits_best = copy.deepcopy(export_limits_best_prev) - else: + elif (metric_prev - metric) >= self.metric_min_improvement_plan: self.log("New plan metric is significantly better from previous plan, using new plan") + else: + self.log("New plan is a cost-neutral improvement but less fragmented ({} vs {} segments), using new plan".format(fragmentation_new, fragmentation_prev)) # Plan is now valid self.log("Plan valid is now true after recompute was {}".format(self.plan_valid)) diff --git a/apps/predbat/tests/test_plan_tiebreak.py b/apps/predbat/tests/test_plan_tiebreak.py new file mode 100644 index 000000000..126ba0475 --- /dev/null +++ b/apps/predbat/tests/test_plan_tiebreak.py @@ -0,0 +1,88 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# fmt off +# pylint: disable=consider-using-f-string +# pylint: disable=line-too-long +# pylint: disable=attribute-defined-outside-init +"""Tests for the near-tie plan fragmentation tie-break used in calculate_plan(). + +When a freshly optimised plan is not better than the incumbent by metric_min_improvement_plan, +Predbat keeps the incumbent. Under flat export rates that locks in a fragmented (split) export +schedule even though the fresh plan is a cleaner single block. The tie-break lets the cleaner +plan win a near-tie provided it is no worse on cost. +""" + + +def _check(cond, message, failures): + """Record a failure message if the condition is not met.""" + if not cond: + print("ERROR: {}".format(message)) + failures.append(message) + + +def _fragmentation_tests(my_predbat, failures): + """plan_fragmentation counts contiguous active (charge/export) segments split by gaps/mode changes.""" + my_predbat.reserve = 0.4 + + # Three adjacent 30-minute export windows all discharging = one contiguous block = 1 segment. + ew = [{"start": 0, "end": 30}, {"start": 30, "end": 60}, {"start": 60, "end": 90}] + _check(my_predbat.plan_fragmentation([], [], ew, [0.0, 0.0, 0.0]) == 1, "contiguous export block should be 1 segment", failures) + + # Same span but the middle slot is off (100) -> a time gap -> two separate export runs = 2 segments. + _check(my_predbat.plan_fragmentation([], [], ew, [0.0, 100.0, 0.0]) == 2, "export split by an idle slot should be 2 segments", failures) + + # Staircase targets (all discharging, all adjacent) is still one block - differing depth is not a split. + _check(my_predbat.plan_fragmentation([], [], ew, [28.0, 7.0, 6.0]) == 1, "adjacent discharge windows are one segment regardless of depth", failures) + + # Freeze (99) and off (100) are not battery export, so they are inactive -> 0 segments. + _check(my_predbat.plan_fragmentation([], [], ew, [99.0, 100.0, 99.0]) == 0, "freeze/off export windows are inactive", failures) + + # A charge window (target above reserve) adjacent to an export window is a mode change = 2 segments. + cw = [{"start": 0, "end": 30}] + ew2 = [{"start": 30, "end": 60}] + _check(my_predbat.plan_fragmentation(cw, [5.0], ew2, [0.0]) == 2, "adjacent charge then export is a mode change = 2 segments", failures) + + # A charge window at/below reserve is not charging -> inactive. + _check(my_predbat.plan_fragmentation([{"start": 0, "end": 30}], [0.4], [], []) == 0, "charge target at reserve is inactive", failures) + + # The real scenario: split evening (3 runs with gaps) must be more fragmented than the merged block (1 run). + split_ew = [{"start": 1250, "end": 1350}, {"start": 1370, "end": 1410}, {"start": 1425, "end": 1440}] + merged_ew = [{"start": 1250, "end": 1410}] + frag_split = my_predbat.plan_fragmentation([], [], split_ew, [28.0, 7.0, 6.0]) + frag_merged = my_predbat.plan_fragmentation([], [], merged_ew, [0.0]) + _check(frag_split > frag_merged, "split evening ({}) should be more fragmented than merged ({})".format(frag_split, frag_merged), failures) + + +def _tiebreak_decision_tests(my_predbat, failures): + """should_replace_plan: adopt new on a clear win, or on a near-tie if no worse and strictly cleaner.""" + my_predbat.metric_min_improvement_plan = 2.0 + + # Clearly better (gap >= threshold) always adopts the new plan, regardless of fragmentation. + _check(my_predbat.should_replace_plan(-100.0, -103.0, 1, 5) is True, "clear improvement adopts new plan", failures) + + # Real near-tie from the trefor2 case: prev split -224.78, new merged -226.57 (1.79p better), cleaner. + _check(my_predbat.should_replace_plan(-224.78, -226.57, 5, 1) is True, "near-tie cleaner+cheaper plan adopts new", failures) + + # Near-tie but equally fragmented -> keep previous (preserves anti-jitter hysteresis). + _check(my_predbat.should_replace_plan(-224.78, -226.57, 3, 3) is False, "near-tie equal fragmentation keeps previous", failures) + + # Near-tie, new is cleaner but MORE expensive (gap < 0) -> keep previous (never raise cost). + _check(my_predbat.should_replace_plan(-226.57, -224.78, 5, 1) is False, "cleaner but costlier plan is rejected", failures) + + # Near-tie, new is cheaper but MORE fragmented -> keep previous. + _check(my_predbat.should_replace_plan(-224.78, -225.0, 1, 4) is False, "cheaper but more fragmented plan is rejected", failures) + + # Exactly cost-neutral (gap == 0) but cleaner -> adopt new. + _check(my_predbat.should_replace_plan(-200.0, -200.0, 4, 2) is True, "cost-neutral cleaner plan adopts new", failures) + + +def run_plan_tiebreak_tests(my_predbat): + """Run the plan fragmentation tie-break tests. Returns True on failure.""" + print("**** Running plan tie-break tests ****") + failures = [] + _fragmentation_tests(my_predbat, failures) + _tiebreak_decision_tests(my_predbat, failures) + return len(failures) > 0 diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index 2463b19ea..d46a80ba0 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -43,6 +43,7 @@ from tests.test_component_health_status import test_component_health_status from tests.test_optimise_levels import run_optimise_levels_tests from tests.test_trim_export import run_trim_export_tests +from tests.test_plan_tiebreak import run_plan_tiebreak_tests from tests.test_export_commitment import run_export_commitment_tests from tests.test_energydataservice import run_energydataservice_tests from tests.test_iboost import run_iboost_smart_tests @@ -386,6 +387,7 @@ def main(): ("gateway", run_gateway_tests, "GatewayMQTT component tests (protobuf, plan serialization, commands, telemetry)", False), ("optimise_levels", run_optimise_levels_tests, "Optimise levels tests", False), ("trim_export", run_trim_export_tests, "Export trim ordering (buffer from cheapest slot) tests", False), + ("plan_tiebreak", run_plan_tiebreak_tests, "Plan fragmentation near-tie tie-break tests", False), ("export_commitment", run_export_commitment_tests, "Forced-export commitment / anti-flapping tests", False), ("load_ml", test_load_ml, "ML Load Forecaster tests (MLP, training, persistence, validation)", True), # ("optimise_windows", run_optimise_all_windows_tests, "Optimise all windows tests", True),