fix(plan): split low-power charge windows at dawn, not window end

A charge window built from a single long cheap-rate period spanning
sunrise was treated as one window by find_charge_rate's PV-overlap
check (#4373), which abandons low power charging for the whole window
the moment any PV is forecast in it - including the still-dark hours
before sunrise (#4557).

Add calc_dawn: classify each plan_interval_minutes slot as PV-light or
PV-dark and split charge windows at that boundary in find_charge_window
(pv_light_dark param), reusing the same mechanism already used for
export windows' alternate_rate_boundary. The dark portion now stays
low power throughout; the light portion keeps today's behaviour.

Dawn is the first bucket that crosses LOW_POWER_PV_LIGHT_FRACTION of
the peak PV forecast anywhere in the horizon - a fraction of the
forecast's own peak rather than a fixed Watts figure, so the same
threshold is meaningful whether the system is 2kWp or 15kWp. Within a
calendar day it's a one-way latch: once crossed, later buckets stay
light even if PV genuinely dips back under the threshold (a cloud
passing), so an intermittently cloudy morning can't chop the window
into several flip-flopping pieces. The latch resets at each day
boundary, which also makes it behave correctly at extreme latitudes -
a polar-night day never crosses (stays all-dark) and a polar-day day
crosses from the first bucket (stays all-light).

No dusk-side equivalent needed: the PV-overlap check is a forward sum
from "now" to window end, recomputed every loop, so it already
self-corrects once a light patch is behind it - the false-trigger-
then-never-corrects failure mode this fix addresses only exists in
the dawn direction.

Verified against the #4557 reporter's own debug.yaml throughout - the
real 03:00-10:00 window splits cleanly at the actual crossing point,
using that household's own forecast peak.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: chalfontchubby

apps/predbat/const.py

@@ -52,6 +52,16 @@
 # which increases the cost of the plan over the full rate charge the planner costed the window at.
 LOW_POWER_PV_THRESHOLD = 0.1
 
+# Fraction of the peak forecast PV power above which a plan_interval_minutes bucket is classed as
+# "light" rather than "dark" when deciding where to split a charge window (calc_dawn). A charge window
+# otherwise built from a single long cheap-rate period spanning sunrise would apply LOW_POWER_PV_THRESHOLD
+# across the whole thing and abandon low power charging even for the still-dark hours before the sun is
+# up (#4557) - splitting at dawn keeps the dark portion as its own window, genuinely PV-free, so it stays
+# throttled. A fraction of that forecast's own peak, rather than a fixed Watts figure, scales with the
+# site - a fixed threshold picked for a typical system would be noise-level for a large array and
+# unreachable for a small one.
+LOW_POWER_PV_LIGHT_FRACTION = 0.1
+
 INVERTER_TEST = False  # Run inverter control self test
 
 # Create an array of times in the day in 5-minute intervals

apps/predbat/fetch.py

@@ -20,7 +20,19 @@
 
 from datetime import datetime, timedelta
 from utils import minutes_to_time, str2time, dp1, dp2, dp3, dp4, time_string_to_stamp, minute_data, get_now_from_cumulative, MinuteArray
-from const import MINUTE_WATT, PREDICT_STEP, TIME_FORMAT, PREDBAT_MODE_OPTIONS, PREDBAT_MODE_CONTROL_SOC, PREDBAT_MODE_CONTROL_CHARGEDISCHARGE, PREDBAT_MODE_CONTROL_CHARGE, PREDBAT_MODE_MONITOR, LOAD_FORECAST_HISTORY_MAX_DAYS, PREDBAT_MAX_CARS
+from const import (
+    MINUTE_WATT,
+    PREDICT_STEP,
+    TIME_FORMAT,
+    PREDBAT_MODE_OPTIONS,
+    PREDBAT_MODE_CONTROL_SOC,
+    PREDBAT_MODE_CONTROL_CHARGEDISCHARGE,
+    PREDBAT_MODE_CONTROL_CHARGE,
+    PREDBAT_MODE_MONITOR,
+    LOAD_FORECAST_HISTORY_MAX_DAYS,
+    PREDBAT_MAX_CARS,
+    LOW_POWER_PV_LIGHT_FRACTION,
+)
 from predbat_metrics import metrics
 from futurerate import FutureRate
 from axle import fetch_axle_sessions, load_axle_slot, fetch_axle_active
@@ -1028,8 +1040,10 @@ def fetch_sensor_data(self, save=True):
 
         # Find charging windows
         if self.rate_import:
+            pv_light_dark = self.calc_dawn() if self.set_charge_low_power else {}
+
             # Find charging window
-            self.low_rates, lowest, highest = self.rate_scan_window(self.rate_import, 5, self.rate_import_cost_threshold, False, alt_rates=self.rate_export)
+            self.low_rates, lowest, highest = self.rate_scan_window(self.rate_import, 5, self.rate_import_cost_threshold, False, alt_rates=self.rate_export, pv_light_dark=pv_light_dark)
             self.log("Low Import rate found rates in range {}{} to {}{}".format(lowest, curr, highest, curr))
             # Update threshold automatically
             if self.rate_low_threshold == 0 and highest >= self.rate_min:
@@ -1539,9 +1553,80 @@ def rate_replicate(self, rates, rate_io={}, is_import=True, is_gas=False):
 
         return rates, replicated_rates
 
-    def find_charge_window(self, rates, minute, threshold_rate, find_high, alt_rates={}):
+    def calc_dawn(self):
+        """
+        Find dawn in self.pv_forecast_minute and return a pv_light_dark dict classifying each minute
+        as light (1, at/after dawn) or dark (0, before it). Used by find_charge_window to split a
+        charge window at the light/dark boundary rather than abandoning low power charging for a whole
+        window just because its tail overlaps the sun - see #4557.
+
+        Classified per plan_interval_minutes bucket (averaged), not per raw minute - a threshold
+        compared minute to minute would let ordinary forecast noise near the cutoff (e.g. a patchy dawn
+        hovering around the threshold) retrigger the split repeatedly and chop the window into several
+        small pieces. Bucketing makes the classification a step function that can change at most once
+        per bucket boundary, at the same granularity the plan already displays.
+
+        Within each calendar day, light is a one-way latch: the first bucket that crosses the threshold
+        confirms dawn, and every later bucket that day stays light too, even if PV genuinely dips back
+        under the threshold later (a cloud passing). This is a dawn detector, not a tracker of every
+        rise and fall - it only cares about finding the first dawn each day, so an intermittently
+        cloudy morning can't chop the window into several flip-flopping pieces. The latch resets at
+        each day boundary so the next day's dawn is found independently rather than one early crossing
+        holding "light" for the rest of the multi-day forecast horizon - this also makes a polar-night
+        day correctly stay all-dark (dawn never crosses) and a polar-day day correctly stay all-light
+        (crossed from the very first bucket).
+
+        The threshold itself is LOW_POWER_PV_LIGHT_FRACTION of the peak PV forecast anywhere in
+        self.pv_forecast_minute, not a fixed Watts figure, so it scales with the site rather than being
+        picked for a "typical" system size.
+
+        Built from whatever PV forecast is already in self.pv_forecast_minute, which at the point this
+        is called from fetch_sensor_data is up to one cycle stale (refreshed later this same loop by
+        fetch_pv_forecast()) - fine for a forecast that doesn't meaningfully change minute to minute.
+        """
+        pv_light_dark = {}
+        if not self.pv_forecast_minute:
+            return pv_light_dark
+
+        peak_pv = max(self.pv_forecast_minute.values())
+        if peak_pv <= 0:
+            return {pv_minute: 0 for pv_minute in self.pv_forecast_minute}
+
+        light_threshold = peak_pv * LOW_POWER_PV_LIGHT_FRACTION
+        interval = self.plan_interval_minutes
+        bucket_sums = {}
+        bucket_counts = {}
+        for pv_minute, pv in self.pv_forecast_minute.items():
+            bucket = pv_minute // interval
+            bucket_sums[bucket] = bucket_sums.get(bucket, 0.0) + pv
+            bucket_counts[bucket] = bucket_counts.get(bucket, 0) + 1
+        bucket_crossed = {bucket: (1 if (bucket_sums[bucket] / bucket_counts[bucket]) >= light_threshold else 0) for bucket in bucket_sums}
+
+        bucket_light = {}
+        after_dawn = False
+        current_day = None
+        for bucket in sorted(bucket_crossed):
+            bucket_day = (bucket * interval) // (24 * 60)
+            if bucket_day != current_day:
+                after_dawn = False
+                current_day = bucket_day
+            if bucket_crossed[bucket]:
+                after_dawn = True
+            bucket_light[bucket] = 1 if after_dawn else 0
+
+        pv_light_dark = {pv_minute: bucket_light[pv_minute // interval] for pv_minute in self.pv_forecast_minute}
+        return pv_light_dark
+
+    def find_charge_window(self, rates, minute, threshold_rate, find_high, alt_rates={}, pv_light_dark={}):
         """
         Find the charging windows based on the low rate threshold (percent below average)
+
+        pv_light_dark, when scanning for charge (not find_high) windows, is a minute-indexed dict of 0/1
+        marking whether PV forecast is at/after dawn ("light") or not ("dark") - see calc_dawn.
+        A transition between the two forces a window split, so a charge window that would otherwise
+        span sunrise (e.g. a single long cheap-rate period) is instead built as separate dark and
+        light windows - see #4557, where low power charging was defeated for the whole window,
+        including the still-dark hours, just because the window's tail overlapped PV later on.
         """
         rate_low_start = -1
         rate_low_end = -1
@@ -1550,6 +1635,8 @@ def find_charge_window(self, rates, minute, threshold_rate, find_high, alt_rates
         rate_low_count = 0
         alternate_rate_boundary = False
         alt_rate_last = None
+        pv_boundary = False
+        pv_light_dark_last = None
 
         # Work out alternate rate threshold
         alt_rate_max = max(alt_rates.values()) if alt_rates else 0
@@ -1563,6 +1650,13 @@ def find_charge_window(self, rates, minute, threshold_rate, find_high, alt_rates
             if (alt_rate is not None) and (alt_rate_last is not None) and (abs(alt_rate - alt_rate_last) >= alt_rate_threshold):
                 # Create alternate rate boundary if the rate is different
                 alternate_rate_boundary = True
+            pv_state = pv_light_dark.get(minute, None)
+            if (rate_low_start >= 0) and (pv_state is not None) and (pv_light_dark_last is not None) and (pv_state != pv_light_dark_last):
+                # Create a PV boundary when the light/dark classification changes since the window
+                # started. Gated on rate_low_start so a transition crossed only during the pre-window
+                # search phase (e.g. scanning past last night's sunset before this window is even
+                # found) doesn't leave a stale boundary that breaks the window the moment it starts.
+                pv_boundary = True
             if minute in rates:
                 rate = rates[minute]
                 if ((not find_high) and (rate <= threshold_rate)) or (find_high and (rate >= threshold_rate) and (rate > 0)) or (minute in self.manual_all_times) or (rate_low_start in self.manual_all_times):
@@ -1587,6 +1681,10 @@ def find_charge_window(self, rates, minute, threshold_rate, find_high, alt_rates
                         # Export slot can never be bigger than 4 hours
                         rate_low_end = minute
                         break
+                    if (not find_high) and (rate_low_start >= 0) and ((minute - rate_low_start) >= self.plan_interval_minutes) and pv_boundary:
+                        # Split a charge window where PV forecast crosses the light/dark threshold
+                        rate_low_end = minute
+                        break
                     if rate_low_start < 0:
                         rate_low_start = minute
                         rate_low_end = stop_at
@@ -1606,6 +1704,7 @@ def find_charge_window(self, rates, minute, threshold_rate, find_high, alt_rates
                     break
             minute += 5
             alt_rate_last = alt_rate
+            pv_light_dark_last = pv_state
 
         if rate_low_count:
             rate_low_average = dp2(rate_low_average / rate_low_count)
@@ -1876,7 +1975,7 @@ def rate_min_forward_calc(self, rates):
 
         return rate_min_forward
 
-    def rate_scan_window(self, rates, rate_low_min_window, threshold_rate, find_high, return_raw=False, alt_rates={}):
+    def rate_scan_window(self, rates, rate_low_min_window, threshold_rate, find_high, return_raw=False, alt_rates={}, pv_light_dark={}):
         """
         Scan for the next high/low rate window
         """
@@ -1887,7 +1986,7 @@ def rate_scan_window(self, rates, rate_low_min_window, threshold_rate, find_high
         upcoming_period = self.minutes_now + 4 * 60
 
         while True:
-            rate_low_start, rate_low_end, rate_low_average = self.find_charge_window(rates, minute, threshold_rate, find_high, alt_rates=alt_rates)
+            rate_low_start, rate_low_end, rate_low_average = self.find_charge_window(rates, minute, threshold_rate, find_high, alt_rates=alt_rates, pv_light_dark=pv_light_dark)
             window = {}
             window["start"] = rate_low_start
             window["end"] = rate_low_end