feat(plan): add a mode that forbids charging the battery from the grid

apps/predbat/config.py

@@ -933,6 +933,13 @@
         "enable": "num_cars",
         "enable_condition": "num_cars > 0",
     },
+    {
+        "name": "battery_charging_from_grid",
+        "friendly_name": "Allow battery to charge from grid",
+        "type": "switch",
+        "default": True,
+        "reset_inverter": True,
+    },
     {
         "name": "calculate_export_oncharge",
         "oldname": "calculate_discharge_oncharge",
@@ -2527,6 +2534,7 @@
     "charge_end_time": {"type": "sensor_list", "sensor_type": "string", "modify": True, "entries": "num_inverters"},
     "charge_limit": {"type": "sensor_list", "sensor_type": "float", "modify": True, "entries": "num_inverters"},
     "scheduled_charge_enable": {"type": "sensor_list", "sensor_type": "boolean", "modify": True, "entries": "num_inverters"},
+    "grid_charge_enable": {"type": "sensor_list", "sensor_type": "boolean", "modify": True, "entries": "num_inverters"},
     "scheduled_discharge_enable": {"type": "sensor_list", "sensor_type": "boolean", "modify": True, "entries": "num_inverters"},
     "discharge_start_time": {"type": "sensor_list", "sensor_type": "string", "modify": True, "entries": "num_inverters"},
     "discharge_end_time": {"type": "sensor_list", "sensor_type": "string", "modify": True, "entries": "num_inverters"},
@@ -2710,6 +2718,8 @@
     "discharge_start_service": {"type": "dict_list|string"},
     "discharge_stop_service": {"type": "dict_list|string"},
     "charge_freeze_service": {"type": "dict_list|string"},
+    "grid_charge_enable_service": {"type": "dict_list|string"},
+    "grid_charge_disable_service": {"type": "dict_list|string"},
     "discharge_freeze_service": {"type": "dict_list|string"},
     "device_id": {"type": "string", "empty": False},
     "predheat": {"type": "dict"},

apps/predbat/execute.py

@@ -186,6 +186,10 @@ def execute_plan(self):
                 self.clear_control_ledger("inverter {} is calibrating, so its own firmware is driving the settings".format(inverter.id))
                 break
 
+            # Assert the device's own grid-charge control, where it has one. Negative import rates are
+            # the single exemption: being paid to import is worth taking even in no-grid-charge mode.
+            inverter.adjust_grid_charge(self.battery_charging_from_grid or (self.rate_import.get(self.minutes_now, 0) < 0))
+
             resetDischarge = self.set_charge_window or self.set_export_window
             resetCharge = self.set_charge_window or self.set_export_window
             resetPause = self.set_charge_window or self.set_export_window
@@ -803,8 +807,18 @@ def is_freeze_charge(self, charge_limit_kwh):
         """
         Check if a charge limit (in kWh) represents a freeze charge (i.e., equals reserve)
         Uses percentage comparison to avoid floating point rounding issues
+
+        A limit that could only be reached by importing is also treated as a freeze when grid charging
+        is disabled. The planner will not produce one, but a plan computed before the switch was turned
+        off - or restored across a restart - otherwise executes an import the planner would now reject.
         """
-        return calc_percent_limit(charge_limit_kwh, self.soc_max) == self.reserve_percent
+        target_percent = calc_percent_limit(charge_limit_kwh, self.soc_max)
+        if target_percent == self.reserve_percent:
+            return True
+        if not self.battery_charging_from_grid and (target_percent > self.soc_percent) and (self.rate_import.get(self.minutes_now, 0) >= 0):
+            self.log("Charge target {}% above SoC {}% downgraded to a hold as battery_charging_from_grid is off".format(target_percent, self.soc_percent))
+            return True
+        return False
 
     def reset_inverter(self):
         """

apps/predbat/fetch.py

@@ -2709,6 +2709,7 @@ def fetch_config_options(self):
         self.set_reserve_hold = True
         self.set_export_freeze = self.get_arg("set_export_freeze")
         self.set_charge_freeze = self.get_arg("set_charge_freeze")
+        self.battery_charging_from_grid = self.get_arg("battery_charging_from_grid")
         self.set_charge_low_power = self.get_arg("set_charge_low_power")
         self.set_export_low_power = self.get_arg("set_export_low_power")
         self.charge_low_power_margin = self.get_arg("charge_low_power_margin")

apps/predbat/inverter.py

@@ -1773,6 +1773,47 @@ def mimic_target_soc(self, current_charge_limit, discharge=False):
                     self.base.log(f"Current SoC {self.soc_percent}% is less than Target SoC {current_charge_limit}. Grid Charge enabled, amp rate written to inverter.")
                 self.base.log(f"Current SoC {self.soc_percent}% is less than Target SoC {current_charge_limit}. Grid charging enabled with charge current set to {self.base.get_arg('timed_charge_current', index=self.id, default=65):0.2f}")
 
+    def adjust_grid_charge(self, allow):
+        """
+        Allow or forbid charging the battery from the grid at the device itself
+
+        A no-op unless one of the optional controls below is configured, so nothing changes for
+        inverters that have no such control. Where the device does have one, asserting it means a
+        grid charge cannot happen even if Predbat's own logic is wrong or a schedule is stale.
+
+        Two forms are supported. An entity is simplest where the integration exposes a switch;
+        the service pair suits cloud inverters driven through REST commands, which is how a
+        Powerwall reaches the Fleet API disallow_charge_from_grid flag.
+
+        Output Entities:
+        ================
+
+            Config arg                         Type          Units
+            ----------                         ----          -----
+            grid_charge_enable                 switch        on/off
+
+        Services:
+        =========
+
+            grid_charge_enable_service         called when grid charging becomes allowed
+            grid_charge_disable_service        called when grid charging becomes forbidden
+
+        Parameters:
+        - allow: True to permit charging from the grid, False to forbid it
+        """
+        entity_id = self.base.get_arg("grid_charge_enable", indirect=False, index=self.id, default=None)
+        if entity_id:
+            self.write_and_poll_switch("grid_charge_enable", entity_id, bool(allow))
+            return
+
+        # Its own dedupe domain, so an unchanged assertion is skipped without interfering with the
+        # charge/discharge service hashes
+        service_data = {"device_id": self.base.get_arg("device_id", index=self.id, default=""), "allow": bool(allow)}
+        if allow:
+            self.call_service_template("grid_charge_enable_service", service_data, domain="grid_charge")
+        else:
+            self.call_service_template("grid_charge_disable_service", service_data, domain="grid_charge")
+
     def adjust_reserve(self, reserve):
         """
         Adjust the output reserve target %

apps/predbat/plan.py

@@ -1984,6 +1984,16 @@ def optimise_charge_limit(self, window_n, record_charge_windows, charge_limit, c
             if not allow_freeze and (self.reserve in try_socs):
                 try_socs.remove(self.reserve)
 
+        # With grid charging disabled the only legal window states are off and hold. A charge limit above
+        # the SoC is met from the grid in the model (see run_prediction), whereas a hold charges from PV
+        # only and an off window runs in ECO mode, which soaks surplus anyway - so no solar is given up.
+        if not self.allow_grid_charge_window(charge_window, window_n, all_n):
+            try_socs = [try_soc for try_soc in try_socs if try_soc <= self.reserve]
+            if allow_freeze and (self.reserve not in try_socs):
+                try_socs.append(self.reserve)
+            if 0 not in try_socs:
+                try_socs.append(0)
+
         # Run the simulations in parallel
         results = []
         results10 = []
@@ -2834,8 +2844,13 @@ def clip_charge_slots(self, minutes_now, predict_soc, charge_window_best, charge
         be merged, which is only sound when the limit had no influence on the simulated charge. The achieved SoC
         can land just under the limit even when the limit never clamped it (e.g. charge loss/rounding), and can dip
         a hair below its own peak for the same reason, so both tests need a margin of one charge step to tell a real effect from rounding.
+
+        The clip-up is skipped entirely while grid charging is disabled. It runs after the window optimiser, so
+        raising a hold to a full limit there quietly reintroduces exactly what allow_grid_charge_window excluded -
+        the plan then reads as a charge, and execute has to recognise and downgrade it on every cycle.
         """
         charge_step = self.battery_rate_max_charge * self.battery_rate_max_scaling * step
+        allow_clip_up = self.battery_charging_from_grid
         for window_n in range(min(record_charge_windows, len(charge_window_best))):
             window = charge_window_best[window_n]
             limit = charge_limit_best[window_n]
@@ -2870,18 +2885,18 @@ def clip_charge_slots(self, minutes_now, predict_soc, charge_window_best, charge
                     # model whether the nominal plan changes without the slot, which subsumes the old
                     # never-reaches-limit and freeze-at-100% removal branches. What is left here narrows the
                     # limit to what the window can actually achieve, so adjacent windows share a limit and merge.
-                    if soc_max < (limit - charge_step):
+                    if allow_clip_up and soc_max < (limit - charge_step):
                         # Work out what can be achieved in the window and set the target to match that
                         window["target"] = soc_max
                         charge_limit_best[window_n] = self.soc_max
                         if self.debug_enable:
                             self.log("Clip up charge window {} from {} - {} from limit {} to new limit {} target set to {}".format(window_n, window_start, window_end, limit, charge_limit_best[window_n], window["target"]))
-                    elif (soc_max > (soc_m1 + charge_step)) and soc_max == limit:
+                    elif allow_clip_up and (soc_max > (soc_m1 + charge_step)) and soc_max == limit:
                         window["target"] = soc_max
                         charge_limit_best[window_n] = self.soc_max
                         if self.debug_enable:
                             self.log("Clip up charge window {} from {} - {} from limit {} to new limit {} target set to {}".format(window_n, window_start, window_end, limit, charge_limit_best[window_n], window["target"]))
-                    elif limit == self.reserve and (dp1(soc_min) == dp1(self.soc_max)) and (dp1(soc_max) == dp1(self.soc_max)):
+                    elif allow_clip_up and limit == self.reserve and (dp1(soc_min) == dp1(self.soc_max)) and (dp1(soc_max) == dp1(self.soc_max)):
                         # Reserve slot, so set to 100% if we are already at 100%
                         window["target"] = soc_max
                         charge_limit_best[window_n] = self.soc_max
@@ -3603,6 +3618,30 @@ def optimise_swap_charge(self, record_charge_windows, debug_mode=False):
 
             self.log("Swap charge optimisation finished metric {}{}, cost {}{}".format(dp2(selected_metric), curr, dp2(selected_cost), curr))
 
+    def allow_grid_charge_window(self, charge_window, window_n, all_n=None):
+        """
+        Is the battery allowed to charge from the grid in this window?
+
+        True unless the user has turned battery_charging_from_grid off, in which case the only
+        exemption is a window whose import rate is negative - being paid to import is the one case
+        where filling the battery from the grid is unambiguously worth doing. When several windows
+        are optimised together every one of them must be negative to qualify.
+
+        Parameters:
+        - charge_window: list of charge windows
+        - window_n: index of the window being optimised
+        - all_n: indices when a group of windows is optimised together, else None
+
+        Returns:
+        - bool: True if a charge limit above the current SoC may be considered
+        """
+        if self.battery_charging_from_grid:
+            return True
+        for window_id in all_n if all_n else [window_n]:
+            if charge_window[window_id].get("average", 0) >= 0:
+                return False
+        return True
+
     def allow_this_charge_window(self, charge_window_n):
         """
         Allowed to optimise this charge window?
@@ -4234,7 +4273,13 @@ def optimise_charge_windows_manual(self):
                 elif self.charge_window_best[window_n]["start"] in self.manual_freeze_export_times:
                     self.charge_limit_best[window_n] = 0
                 elif self.charge_window_best[window_n]["start"] in self.manual_charge_times:
-                    self.charge_limit_best[window_n] = self.soc_max
+                    if self.allow_grid_charge_window(self.charge_window_best, window_n):
+                        self.charge_limit_best[window_n] = self.soc_max
+                    else:
+                        # Grid charging is off, so downgrade the manual charge to a hold rather than
+                        # letting a manual slot quietly do the one thing the mode exists to prevent
+                        self.log("Manual charge at {} downgraded to hold as battery_charging_from_grid is off".format(self.time_abs_str(self.charge_window_best[window_n]["start"])))
+                        self.charge_limit_best[window_n] = self.reserve
                 elif self.charge_window_best[window_n]["start"] in self.manual_freeze_charge_times:
                     self.charge_limit_best[window_n] = self.reserve
 

apps/predbat/teslemetry.py

@@ -521,6 +521,16 @@ def in_window(minutes_now, window):
             return start <= minutes_now < end
         return minutes_now >= start or minutes_now < end
 
+    def grid_charging_permitted(self):
+        """Return False when Predbat's battery_charging_from_grid switch forbids charging from the grid.
+
+        The Powerwall's disallow_charge_from_grid_with_solar_installed flag is the device-level
+        expression of that switch, so it is asserted here rather than only in the plan - a stale
+        schedule or a Predbat restart must not be able to reintroduce a grid charge. Defaults to
+        permitted when the base has not populated the flag yet, matching Predbat's own default.
+        """
+        return bool(getattr(getattr(self, "base", None), "battery_charging_from_grid", True))
+
     def evaluate_schedule(self, minutes_now, soc):
         """Map the committed schedule + wall clock + live SOC to the desired device tuple.
 
@@ -533,16 +543,22 @@ def evaluate_schedule(self, minutes_now, soc):
         charge = self.schedule.get("charge", {})
         discharge = self.schedule.get("discharge", {})
         reserve = self.schedule.get("reserve", 20)
+        grid_allowed = self.grid_charging_permitted()
         if self.in_window(minutes_now, charge):
             target = int(charge.get("soc", 100))
-            grid = soc < target
+            grid = grid_allowed and soc < target
+            if not grid_allowed:
+                # Solar can still fill the battery towards the target; the grid cannot. Hold the reserve at
+                # the SOC rather than the target, or backup mode would sit waiting for an import that is
+                # forbidden and stop the house drawing on the battery in the meantime.
+                target = min(target, int(soc))
             return {"export_rule": "pv_only", "grid_charging": grid, "reserve": target, "mode": "backup"}
         if self.in_window(minutes_now, discharge):
             target = int(discharge.get("soc", 10))
             if soc > target:
                 return {"export_rule": "battery_ok", "grid_charging": False, "reserve": target, "mode": "autonomous"}
             return {"export_rule": "pv_only", "grid_charging": False, "reserve": target, "mode": "self_consumption"}
-        return {"export_rule": "pv_only", "grid_charging": True, "reserve": int(reserve), "mode": "self_consumption"}
+        return {"export_rule": "pv_only", "grid_charging": grid_allowed, "reserve": int(reserve), "mode": "self_consumption"}
 
     def publish_schedule_entities(self):
         """Publish the schedule entities from the pending schedule (pending == committed after boot/apply).

apps/predbat/tests/test_clip_charge_slots.py

@@ -27,6 +27,7 @@ def run_clip_charge_slots_tests(my_predbat):
     failed |= test_clip_margin_scales_with_charge_rate(my_predbat)
     failed |= test_clip_margin_scales_with_step(my_predbat)
     failed |= test_freeze_charge_to_charge_at_100_soc(my_predbat)
+    failed |= test_clip_up_skipped_without_grid_charging(my_predbat)
     failed |= test_freeze_charge_kept_below_100_soc(my_predbat)
     failed |= test_normal_window_unchanged(my_predbat)
     failed |= test_multiple_windows_mixed(my_predbat)
@@ -364,6 +365,51 @@ def test_clip_margin_scales_with_step(my_predbat):
     return failed
 
 
+def test_clip_up_skipped_without_grid_charging(my_predbat):
+    """With grid charging disabled a hold must stay a hold rather than being clipped up to a full charge.
+
+    clip_charge_slots runs after the window optimiser, so a clip-up here reintroduces exactly the charge
+    that allow_grid_charge_window excluded. The plan then reads as a charge and execute has to recognise
+    and downgrade it on every cycle.
+    """
+    print("**** test_clip_up_skipped_without_grid_charging ****")
+    failed = False
+    setup(my_predbat)
+    saved = my_predbat.battery_charging_from_grid
+
+    minutes_now = 720
+    predict_soc = make_predict_soc(minutes_now, my_predbat.soc_max, 60)
+
+    # Baseline: with grid charging allowed the freeze is clipped up, which is the long-standing behaviour
+    my_predbat.battery_charging_from_grid = True
+    _, limits_on = my_predbat.clip_charge_slots(minutes_now, predict_soc, [make_window(720, 750)], [my_predbat.reserve], 1, 5)
+    if limits_on[0] != my_predbat.soc_max:
+        print("ERROR: with grid charging on the freeze should clip up to {} but got {}".format(my_predbat.soc_max, limits_on[0]))
+        failed = True
+
+    # With it off the hold survives untouched
+    my_predbat.battery_charging_from_grid = False
+    windows_off, limits_off = my_predbat.clip_charge_slots(minutes_now, predict_soc, [make_window(720, 750)], [my_predbat.reserve], 1, 5)
+    if limits_off[0] != my_predbat.reserve:
+        print("ERROR: with grid charging off the freeze should stay at the reserve {} but got {}".format(my_predbat.reserve, limits_off[0]))
+        failed = True
+    if windows_off[0]["target"] > my_predbat.reserve:
+        print("ERROR: with grid charging off the window target should not exceed the reserve, got {}".format(windows_off[0]["target"]))
+        failed = True
+
+    # The other clip-up branch: a limit the charge never reached must also be left alone
+    low_soc = make_predict_soc(minutes_now, 2.0, 60)
+    _, limits_low = my_predbat.clip_charge_slots(minutes_now, low_soc, [make_window(720, 750)], [8.0], 1, 5)
+    if limits_low[0] == my_predbat.soc_max:
+        print("ERROR: with grid charging off an unreached limit should not be clipped up to soc_max")
+        failed = True
+
+    my_predbat.battery_charging_from_grid = saved
+    if not failed:
+        print("PASS")
+    return failed
+
+
 def test_freeze_charge_to_charge_at_100_soc(my_predbat):
     """Freeze charge (limit==reserve) at 100% SoC should be changed to a full charge"""
     print("**** test_freeze_charge_to_charge_at_100_soc ****")