fix(plan): skip low power charging when the charge window overlaps solar

The planner costs every charge window at the full charge rate - low power
charging is only applied to the final plan (prediction.py gates it on
save in best/best10/test). find_charge_rate then sizes the throttled rate
from charge_left / minutes_left with no knowledge of PV.

In a charge window that overlaps solar production the rate cap in

    battery_draw = -max(min(charge_rate_now_curve_step, ...), 0, -battery_to_max)

stops the PV reaching the battery. The surplus is exported at the export
rate and the charge target is then made up from grid import once the sun
has gone, costing more than the full rate charge that was planned for.

find_charge_rate now returns the max rate when the PV forecast summed
across the remainder of the charge window exceeds LOW_POWER_PV_THRESHOLD
(0.1kWh). Both callers supply that sum: the prediction from a suffix sum
of the PV step array built once per run (only when low power is active,
so the optimiser hot loop is untouched), and execute from
pv_forecast_minute between now and the window end.

No C++ kernel change is needed - the kernel only runs when save is falsy,
which is exactly when charge low power is off, so it already takes the max
rate path and stays in parity.

Tests: model scenarios low_power_pv_* / low_power_dark_* (without the fix
the low power run costs 44p against a 0p full rate plan), execute
scenarios charge_low_power_pv / charge_low_power_pv_trace, and a
find_charge_rate unit test for the threshold either side.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: springfall2008

apps/predbat/const.py

@@ -40,6 +40,12 @@
 MAX_INCREMENT = 240 * 100 * 3 / 1000 / 60
 MINUTE_WATT = 60 * 1000
 
+# PV production (kWh) forecast across the remainder of a charge window above which low power charging is
+# abandoned in favour of the max charge rate. Throttling the charge rate while the sun is shining stops the
+# PV reaching the battery, the surplus is exported cheaply and the target is then made up with grid import,
+# which increases the cost of the plan over the full rate charge the planner costed the window at.
+LOW_POWER_PV_THRESHOLD = 0.1
+
 INVERTER_TEST = False  # Run inverter control self test
 
 # Create an array of times in the day in 5-minute intervals

apps/predbat/execute.py

@@ -130,6 +130,13 @@ def execute_plan(self):
                         inv_target_soc_percent = self.adjust_battery_target_multi(inverter, target_soc, True, False, check=True, isFreezeCharge=is_freeze_charge)
 
                         current_charge_rate = inverter.get_current_charge_rate()
+
+                        # How much PV is still forecast before this charge window closes?
+                        pv_window_kwh = 0.0
+                        if self.set_charge_low_power:
+                            for pv_minute in range(self.minutes_now, window["end"]):
+                                pv_window_kwh += self.pv_forecast_minute.get(pv_minute, 0.0)
+
                         new_charge_rate, new_charge_rate_real = find_charge_rate(
                             self.minutes_now,
                             inverter.soc_kw,
@@ -147,6 +154,7 @@ def execute_plan(self):
                             inverter.battery_temperature,
                             self.battery_temperature_charge_curve,
                             current_charge_rate=current_charge_rate / MINUTE_WATT,
+                            pv_window_kwh=pv_window_kwh,
                         )
                         new_charge_rate = int(new_charge_rate * MINUTE_WATT)
 

apps/predbat/prediction.py

@@ -588,6 +588,15 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
             pv_forecast_minute_step_flat = pv_forecast_minute_step
             load_minutes_step_flat = load_minutes_step
 
+        # PV forecast remaining from each step to the end of the forecast, used to work out how much PV a charge
+        # window still overlaps with as low power charging must be abandoned when the sun is contributing
+        pv_remaining_kwh = {}
+        if set_charge_low_power:
+            pv_remaining = 0.0
+            for minute_step in range(((self.forecast_minutes - 1) // step) * step, -1, -step):
+                pv_remaining += pv_forecast_minute_step_flat.get(minute_step, 0.0)
+                pv_remaining_kwh[minute_step] = pv_remaining
+
         # Simulate each forward minute
         minute = 0
         while minute < self.forecast_minutes:
@@ -932,6 +941,13 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
                     battery_rate_max_charge_combined = battery_rate_max_charge + min(battery_rate_max_charge_dc - battery_rate_max_charge, pv_above)
                 else:
                     battery_rate_max_charge_combined = battery_rate_max_charge
+
+                # How much PV is still to come before this charge window closes?
+                pv_window_kwh = 0.0
+                if set_charge_low_power:
+                    window_end_step = min(max(((charge_window[charge_window_n]["end"] - self.minutes_now) // step) * step, minute), self.forecast_minutes)
+                    pv_window_kwh = pv_remaining_kwh.get(minute, 0.0) - pv_remaining_kwh.get(window_end_step, 0.0)
+
                 charge_rate_now, charge_rate_now_curve = find_charge_rate(
                     minute_absolute,
                     soc,
@@ -948,6 +964,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
                     None,
                     battery_temperature,
                     self.battery_temperature_charge_curve,
+                    pv_window_kwh=pv_window_kwh,
                 )
                 charge_rate_now_curve_step = charge_rate_now_curve * step
 

apps/predbat/tests/test_execute.py

@@ -217,6 +217,7 @@ def run_execute_test(
     car_soc=0,
     battery_temperature=20,
     assert_immediate_charge_soc_freeze_array=[],
+    pv_forecast=0.0,
 ):
     print("> Run scenario {}".format(name))
     my_predbat.log("> Run scenario {}".format(name))
@@ -231,6 +232,8 @@ def run_execute_test(
     my_predbat.charge_low_power_margin = charge_low_power_margin
     my_predbat.minutes_now = minutes_now
     my_predbat.battery_temperature_charge_curve = {20: 1.0, 10: 0.5, 9: 0.5, 8: 0.5, 7: 0.5, 6: 0.3, 5: 0.3, 4: 0.3, 3: 0.262, 2: 0.1, 1: 0.1, 0: 0}
+    # Flat PV forecast of pv_forecast kW, reset every scenario so a solar run does not leak into the next one
+    my_predbat.pv_forecast_minute = {minute: pv_forecast / 60.0 for minute in range(minutes_now, minutes_now + my_predbat.forecast_minutes)}
 
     charge_window_best = charge_window_best.copy()
     charge_limit_best = charge_limit_best.copy()
@@ -852,6 +855,49 @@ def run_execute_tests(my_predbat):
     if failed:
         return failed
 
+    # Same window, but 1kW of PV forecast across the 60 minutes (1kWh, over the 0.1kWh threshold).
+    # Throttling would cap how much of that PV reaches the battery, so charge at the max rate instead
+    failed |= run_execute_test(
+        my_predbat,
+        "charge_low_power_pv",
+        charge_window_best=charge_window_best,
+        charge_limit_best=charge_limit_best,
+        assert_charge_time_enable=True,
+        soc_kw=9,
+        set_charge_window=True,
+        set_export_window=True,
+        set_charge_low_power=True,
+        assert_status="Charging",
+        assert_charge_start_time_minutes=-1,
+        assert_charge_end_time_minutes=my_predbat.minutes_now + 60,
+        assert_charge_rate=2000,
+        battery_max_rate=2000,
+        pv_forecast=1.0,
+    )
+    if failed:
+        return failed
+
+    # A trace of PV (0.05kWh over the window) is under the threshold, low power charging still applies
+    failed |= run_execute_test(
+        my_predbat,
+        "charge_low_power_pv_trace",
+        charge_window_best=charge_window_best,
+        charge_limit_best=charge_limit_best,
+        assert_charge_time_enable=True,
+        soc_kw=9,
+        set_charge_window=True,
+        set_export_window=True,
+        set_charge_low_power=True,
+        assert_status="Charging",
+        assert_charge_start_time_minutes=-1,
+        assert_charge_end_time_minutes=my_predbat.minutes_now + 60,
+        assert_charge_rate=600,
+        battery_max_rate=2000,
+        pv_forecast=0.05,
+    )
+    if failed:
+        return failed
+
     # 60 minutes - 10 minute margin = 50 minutes to add 0.4kWh to each battery (x2 inverters)
     # (60 / 50) * 400 = 480
     failed |= run_execute_test(

apps/predbat/tests/test_find_charge_rate.py

@@ -68,6 +68,68 @@ def test_find_charge_rate(my_predbat):
     return failed
 
 
+def test_find_charge_rate_pv_overlap(my_predbat):
+    """
+    Test that low power charging is bypassed when the charge window overlaps with PV production
+
+    Throttling the charge rate while the sun is shining stops PV going into the battery, the surplus
+    is exported cheaply and the target is then made up with grid import, which increases the plan cost.
+    """
+    failed = 0
+
+    log_to = print
+    minutes_now = my_predbat.minutes_now
+    soc = 1.0
+    soc_max = 10.0
+    target_soc = 5.0
+    # Plenty of time left in the window so low power mode has room to slow the charge right down
+    window = {"start": minutes_now - 30, "end": minutes_now + 300}
+    # Flat curves so the rate is not capped by the battery or its temperature
+    battery_charge_power_curve = my_predbat.validate_curve({100: 1.0, 99: 1.0, 98: 1.0, 97: 1.0, 96: 1.0, 95: 1.0, 94: 1.0, 93: 1.0, 92: 1.0, 91: 1.0, 90: 1.0}, "test_flat_charge_curve")
+    battery_temperature_curve = my_predbat.validate_curve({20: 1.0, 19: 1.0, 18: 1.0, 17: 1.0, 16: 1.0, 15: 1.0}, "test_flat_temperature_curve")
+    max_rate = 3000
+
+    args = [
+        minutes_now,
+        soc,
+        window,
+        target_soc,
+        max_rate / MINUTE_WATT,
+        soc_max,
+        battery_charge_power_curve,
+        True,
+        my_predbat.charge_low_power_margin,
+        0,
+        1,
+        0.96,
+        log_to,
+    ]
+    kwargs = {"battery_temperature": 17.0, "battery_temperature_curve": battery_temperature_curve, "current_charge_rate": max_rate / MINUTE_WATT}
+
+    # Without PV low power mode should slow the charge down below the max rate
+    dark_rate, dark_rate_real = find_charge_rate(*args, **kwargs)
+    print("No PV - Best_rate {} Best_rate_real {}".format(dark_rate * MINUTE_WATT, dark_rate_real * MINUTE_WATT))
+    if dark_rate * MINUTE_WATT >= max_rate:
+        print("**** ERROR: Low power mode should reduce the rate below {}W when there is no PV, got {}W ****".format(max_rate, dark_rate * MINUTE_WATT))
+        failed = 1
+
+    # A trace of PV in the window is not enough to matter, low power mode should still apply
+    trace_rate, trace_rate_real = find_charge_rate(*args, pv_window_kwh=0.05, **kwargs)
+    print("Trace PV - Best_rate {} Best_rate_real {}".format(trace_rate * MINUTE_WATT, trace_rate_real * MINUTE_WATT))
+    if trace_rate != dark_rate:
+        print("**** ERROR: 0.05kWh of PV in the window should not change the rate, expected {}W got {}W ****".format(dark_rate * MINUTE_WATT, trace_rate * MINUTE_WATT))
+        failed = 1
+
+    # Real PV production in the window means we must charge at full rate so the PV is not throttled away
+    sun_rate, sun_rate_real = find_charge_rate(*args, pv_window_kwh=2.0, **kwargs)
+    print("With PV - Best_rate {} Best_rate_real {}".format(sun_rate * MINUTE_WATT, sun_rate_real * MINUTE_WATT))
+    if sun_rate * MINUTE_WATT != max_rate:
+        print("**** ERROR: PV production in the window should force the max rate {}W, got {}W ****".format(max_rate, sun_rate * MINUTE_WATT))
+        failed = 1
+
+    return failed
+
+
 def test_find_charge_rate_string_temperature(my_predbat):
     """
     Test find_charge_rate with string temperature indices in the curve

apps/predbat/tests/test_infra.py

@@ -624,6 +624,7 @@ def simple_scenario(
     calculate_export_on_pv=True,
     assert_clipped=0,
     pv_ac_limit=0,
+    pv_hours=None,
 ):
     """
     No PV, No Load
@@ -742,11 +743,11 @@ def simple_scenario(
     load10_step = {}
 
     for minute in range(0, my_predbat.forecast_minutes, 5):
-        pv_step[minute] = pv_amount / (60 / 5) if not pv10 else 0
+        # pv_hours limits PV to the first N hours of the forecast, otherwise it runs at pv_amount all day
+        pv_now = 0 if (pv_hours is not None and minute >= pv_hours * 60) else pv_amount
+        pv_step[minute] = pv_now / (60 / 5) if not pv10 else 0
         load_step[minute] = load_amount / (60 / 5) if not pv10 else 0
-
-    for minute in range(0, my_predbat.forecast_minutes, 5):
-        pv10_step[minute] = pv_amount / (60 / 5) if pv10 else 0
+        pv10_step[minute] = pv_now / (60 / 5) if pv10 else 0
         load10_step[minute] = load_amount / (60 / 5) if pv10 else 0
 
     if charge_car:

apps/predbat/tests/test_model.py

@@ -2009,6 +2009,40 @@ def run_model_tests(my_predbat, prediction_kernel=False):
     # pv_ac_limit must NOT apply to hybrid inverters (PV is DC-coupled, clipping handled by inverter_limit)
     failed |= simple_scenario("pv_ac_limit_hybrid_ignored", my_predbat, 0, 2.0, assert_final_metric=-export_rate * 24, assert_final_soc=24, with_battery=True, hybrid=True, pv_ac_limit=1.5, assert_clipped=0)
 
+    # Low power charging must not make the plan more expensive when the charge window overlaps PV production.
+    # The planner costs every charge window at the full charge rate as low power is only applied to the final
+    # plan, so a throttled rate that caps how much PV reaches the battery pushes the cost above the plan.
+    reset_rates(my_predbat, import_rate, export_rate)
+    reset_inverter(my_predbat)
+
+    # 6kW of PV for the first 2 hours only, with an 8 hour charge window to 12kWh and a 6kW max charge rate.
+    # At full rate the PV alone fills the battery inside those 2 hours, costing nothing. Throttled to fit the
+    # 8 hour window the battery would take only 1.5kW, exporting the other 4.5kW of PV at 5p and then
+    # importing the missing 9kWh at 10p once the sun has gone - 45p worse than the planner costed it at.
+    low_power_pv = {
+        "load_amount": 0,
+        "pv_amount": 6.0,
+        "pv_hours": 2,
+        "charge": 12,
+        "charge_window_best": [{"start": my_predbat.minutes_now, "end": my_predbat.minutes_now + 480, "average": import_rate}],
+        "battery_size": 20,
+        "battery_soc": 0,
+        "battery_rate_max_charge": 6.0,
+        "inverter_limit": 10.0,
+        "export_limit": 10.0,
+        "assert_final_soc": 12,
+        "assert_final_metric": 0,
+    }
+    failed |= simple_scenario("low_power_pv_full_rate", my_predbat, set_charge_low_power=False, **low_power_pv)
+    failed |= simple_scenario("low_power_pv_low_power", my_predbat, set_charge_low_power=True, **low_power_pv)
+
+    # With no PV in the window low power charging still applies, the whole 12kWh comes from the grid either way
+    low_power_dark = dict(low_power_pv)
+    low_power_dark["pv_amount"] = 0
+    low_power_dark["assert_final_metric"] = import_rate * 12
+    failed |= simple_scenario("low_power_dark_full_rate", my_predbat, set_charge_low_power=False, **low_power_dark)
+    failed |= simple_scenario("low_power_dark_low_power", my_predbat, set_charge_low_power=True, **low_power_dark)
+
     my_predbat.prediction_kernel_enable = False
     if failed:
         print("**** ERROR: Some Model tests failed ****")

apps/predbat/unit_test.py

@@ -74,7 +74,7 @@
 from tests.test_web_chart_grouping import run_web_chart_grouping_tests
 from tests.test_web_entity_unit_resolution import run_web_entity_unit_resolution_tests
 from tests.test_window import run_window_sort_tests, run_intersect_window_tests
-from tests.test_find_charge_rate import test_find_charge_rate, test_find_charge_rate_string_temperature, test_find_charge_rate_string_charge_curve
+from tests.test_find_charge_rate import test_find_charge_rate, test_find_charge_rate_pv_overlap, test_find_charge_rate_string_temperature, test_find_charge_rate_string_charge_curve
 from tests.test_manual_api import run_test_manual_api
 from tests.test_manual_soc import run_test_manual_soc
 from tests.test_manual_times import run_test_manual_times
@@ -301,6 +301,7 @@ def main():
         ("rate_replicate", test_rate_replicate, "Rate replicate comprehensive tests (missing slots, IO, offsets, gas)", False),
         ("find_charge_window", test_find_charge_window, "Find charge window gap handling tests", False),
         ("find_charge_rate", test_find_charge_rate, "Find charge rate tests", False),
+        ("find_charge_rate_pv", test_find_charge_rate_pv_overlap, "Find charge rate with PV overlap", False),
         ("find_charge_rate_string_temp", test_find_charge_rate_string_temperature, "Find charge rate string temperature", False),
         ("find_charge_rate_string_curve", test_find_charge_rate_string_charge_curve, "Find charge rate string charge curve", False),
         ("find_charge_curve", run_find_charge_curve_tests, "Find charge curve tests", False),

apps/predbat/utils.py

@@ -19,7 +19,7 @@
 import array
 from datetime import datetime, timedelta, timezone, time
 from functools import lru_cache
-from const import MINUTE_WATT, PREDICT_STEP, TIME_FORMAT, TIME_FORMAT_SECONDS, TIME_FORMAT_OCTOPUS, MAX_INCREMENT, TIME_FORMAT_DAILY
+from const import LOW_POWER_PV_THRESHOLD, MINUTE_WATT, PREDICT_STEP, TIME_FORMAT, TIME_FORMAT_SECONDS, TIME_FORMAT_OCTOPUS, MAX_INCREMENT, TIME_FORMAT_DAILY
 import copy
 
 DAY_OF_WEEK_MAP = {"mon": 0, "tue": 1, "wed": 2, "thu": 3, "fri": 4, "sat": 5, "sun": 6}
@@ -1150,9 +1150,14 @@ def find_charge_rate(
     battery_temperature=20,
     battery_temperature_curve={},
     current_charge_rate=None,
+    pv_window_kwh=0.0,
 ):
     """
     Find the lowest charge rate that fits the charge slow
+
+    pv_window_kwh is the PV forecast in kWh over the remainder of the charge window, when the window
+    overlaps PV production low power charging is abandoned as the throttled rate applies for the whole
+    window and would push the PV out of the battery, raising the cost above the planned full rate charge
     """
     margin = charge_low_power_margin
     target_soc = round(target_soc, 2)
@@ -1169,6 +1174,13 @@ def find_charge_rate(
 
     min_battery_rate = max(400, int(round(battery_rate_min * MINUTE_WATT)))
     if set_charge_low_power:
+        # If the charge window overlaps with PV production then charge at max rate, a throttled rate would
+        # cap the PV going into the battery, exporting the surplus and importing to make the target up later
+        if pv_window_kwh > LOW_POWER_PV_THRESHOLD:
+            if log_to:
+                log_to("Low power mode: PV forecast in window {}kWh > {}kWh, default to max rate".format(dp2(pv_window_kwh), LOW_POWER_PV_THRESHOLD))
+            return max_rate, max_rate_real
+
         minutes_left = window["end"] - minutes_now - margin
         abs_minutes_left = window["end"] - minutes_now
 

docs/customisation.md

@@ -382,6 +382,10 @@ as otherwise the low power charge may not reach the charge target in time.
 The minimum requested charge rate used in this mode is 400 watts (subject to inverter/battery minimum rate limits).
 This setting is off by default.
 
+Low-power charging is skipped for any charge window that overlaps with forecast solar production, the full charge rate is used instead.
+Throttling the charge rate while the sun is shining would cap how much solar reaches the battery, the surplus would be exported at the
+export rate and the charge target then made up from grid import later, which costs more than the full rate charge Predbat planned for.
+
 The YouTube video [low power charging and charging curve](https://youtu.be/L2vY_Vj6pQg?si=0ZiIVrDLHkeDCx7h)
 explains how the low-power charging works and shows how Predbat automatically creates it.