diff --git a/apps/predbat/config.py b/apps/predbat/config.py index ef47fd863..f8f803721 100644 --- a/apps/predbat/config.py +++ b/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"}, diff --git a/apps/predbat/execute.py b/apps/predbat/execute.py index b338d62ee..47f9377f3 100644 --- a/apps/predbat/execute.py +++ b/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): """ diff --git a/apps/predbat/fetch.py b/apps/predbat/fetch.py index 2abfa7979..2d80ac323 100644 --- a/apps/predbat/fetch.py +++ b/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") diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index ac785da77..669c2273d 100644 --- a/apps/predbat/inverter.py +++ b/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 % diff --git a/apps/predbat/plan.py b/apps/predbat/plan.py index fcd28ae73..dac15ac50 100644 --- a/apps/predbat/plan.py +++ b/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 diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index dd6029d77..300ba90ff 100644 --- a/apps/predbat/teslemetry.py +++ b/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). diff --git a/apps/predbat/tests/test_clip_charge_slots.py b/apps/predbat/tests/test_clip_charge_slots.py index 8cb9bf131..1c97fd941 100644 --- a/apps/predbat/tests/test_clip_charge_slots.py +++ b/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 ****") diff --git a/apps/predbat/tests/test_execute.py b/apps/predbat/tests/test_execute.py index 68cc32c0f..ce400531d 100644 --- a/apps/predbat/tests/test_execute.py +++ b/apps/predbat/tests/test_execute.py @@ -42,6 +42,7 @@ def __init__(self, id, soc_kw, soc_max, now_utc): self.inv_has_target_soc = True self.inv_has_charge_enable_time = True self.inv_has_timed_pause = True + self.grid_charge_allowed = None self.inv_has_discharge_enable_time = True self.inv_has_ge_eco_toggle = False self.inv_has_ge_inverter_mode = False @@ -131,6 +132,9 @@ def adjust_inverter_mode(self, force_export, changed_start_end=False): self.force_export = force_export self.changed_start_end = changed_start_end + def adjust_grid_charge(self, allow): + self.grid_charge_allowed = allow + def adjust_reserve(self, reserve): self.reserve_last = reserve self.reserve_current = max(reserve, self.reserve) diff --git a/apps/predbat/tests/test_infra.py b/apps/predbat/tests/test_infra.py index 8d947542c..f1b18e622 100644 --- a/apps/predbat/tests/test_infra.py +++ b/apps/predbat/tests/test_infra.py @@ -366,6 +366,7 @@ def get_default_config(self): "set_reserve_enable": True, "set_export_freeze": True, "set_charge_freeze": True, + "battery_charging_from_grid": True, "set_charge_low_power": False, "set_export_low_power": False, "charge_low_power_margin": 10, @@ -560,6 +561,7 @@ def reset_inverter(my_predbat): my_predbat.set_export_window = True my_predbat.set_charge_freeze = True my_predbat.set_export_freeze = True + my_predbat.battery_charging_from_grid = True def plot(name, prediction): diff --git a/apps/predbat/tests/test_inverter.py b/apps/predbat/tests/test_inverter.py index 4739a68c6..cf20b09a9 100644 --- a/apps/predbat/tests/test_inverter.py +++ b/apps/predbat/tests/test_inverter.py @@ -1330,6 +1330,78 @@ def adjust_charge_immediate(self, target_soc, freeze=False) return failed +def test_call_adjust_grid_charge(test_name, my_predbat, ha, inv, dummy_items): + """ + Tests; + def adjust_grid_charge(self, allow) + """ + failed = False + print("**** Running Test: {} ****".format(test_name)) + + saved_args = {key: my_predbat.args.get(key) for key in ("grid_charge_enable", "grid_charge_enable_service", "grid_charge_disable_service")} + # The service dedupe cache is shared across the inverter tests, so leave it exactly as found - + # clearing it makes the next test's "this was already called" expectation fail + saved_service_hash = dict(my_predbat.last_service_hash) + + def check(label, expected): + """Compare the captured service calls against what this case should produce.""" + result = ha.get_service_store() + if json.dumps(expected) != json.dumps(result): + print("ERROR: adjust_grid_charge {} should call {} got {}".format(label, expected, result)) + return True + return False + + # Nothing configured: a no-op rather than an error, so inverters without the control are unaffected + my_predbat.args["grid_charge_enable"] = None + my_predbat.args["grid_charge_enable_service"] = None + my_predbat.args["grid_charge_disable_service"] = None + ha.service_store_enable = True + ha.service_store = [] + inv.adjust_grid_charge(False) + failed |= check("with nothing configured", []) + + # Service form: enable and disable call their own hooks + my_predbat.args["grid_charge_enable_service"] = "grid_charge_on" + my_predbat.args["grid_charge_disable_service"] = "grid_charge_off" + my_predbat.args["device_id"] = "DID0" + + ha.service_store = [] + my_predbat.last_service_hash = {} + inv.adjust_grid_charge(False) + failed |= check("disable", [["grid_charge_off", {"device_id": "DID0", "allow": False}]]) + + ha.service_store = [] + my_predbat.last_service_hash = {} + inv.adjust_grid_charge(True) + failed |= check("enable", [["grid_charge_on", {"device_id": "DID0", "allow": True}]]) + + # Repeating the same assertion is deduped, so an unchanged mode costs no API calls + ha.service_store = [] + inv.adjust_grid_charge(True) + failed |= check("repeated enable", []) + + # The entity form takes precedence when both are configured: the switch is written and the + # service hook is not called. The switch write itself is covered by the write_and_poll_switch tests. + my_predbat.args["grid_charge_enable"] = "switch.grid_charge" + dummy_items["switch.grid_charge"] = "on" + ha.service_store = [] + my_predbat.last_service_hash = {} + inv.adjust_grid_charge(False) + result = ha.get_service_store() + if any(call[0] in ("grid_charge_off", "grid_charge_on") for call in result): + print("ERROR: adjust_grid_charge should not call the service hook when an entity is configured, got {}".format(result)) + failed = True + if not any(call[0] == "switch/turn_off" for call in result): + print("ERROR: adjust_grid_charge entity form should write the switch off, got {}".format(result)) + failed = True + + for key, value in saved_args.items(): + my_predbat.args[key] = value + my_predbat.last_service_hash = saved_service_hash + ha.service_store_enable = False + return failed + + def test_call_adjust_export_immediate(test_name, my_predbat, ha, inv, dummy_items, soc, repeat=False, freeze=False, clear=False, charge_stop=False, discharge_start_time="00:00:00", discharge_end_time="23:55:00", no_freeze=False): """ Tests; @@ -3283,6 +3355,7 @@ def run_inverter_tests(my_predbat_dummy): failed |= test_call_adjust_charge_immediate("charge_immediate7", my_predbat, ha, inv, dummy_items, 50, freeze=True) failed |= test_call_adjust_charge_immediate("charge_immediate8", my_predbat, ha, inv, dummy_items, 50, freeze=False, no_freeze=True) failed |= test_call_adjust_charge_immediate("charge_immediate9", my_predbat, ha, inv, dummy_items, 51.0) + failed |= test_call_adjust_grid_charge("grid_charge_control", my_predbat, ha, inv, dummy_items) if failed: return failed diff --git a/apps/predbat/tests/test_no_grid_charge.py b/apps/predbat/tests/test_no_grid_charge.py new file mode 100644 index 000000000..27414f985 --- /dev/null +++ b/apps/predbat/tests/test_no_grid_charge.py @@ -0,0 +1,294 @@ +# ----------------------------------------------------------------------------- +# 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 battery_charging_from_grid mode. + +With the switch off the planner may only leave a charge window off or hold it at the reserve, because +any limit above the SoC is met from the grid in the prediction model. Solar charging is unaffected: +a hold takes PV, and an off window runs in ECO mode which soaks surplus anyway. +""" + +from tests.test_infra import reset_inverter, reset_rates, update_rates_import, update_rates_export +from prediction import Prediction + + +def build_windows(my_predbat, price_cheap=5.0, price_peak=30.0): + """Build 48 half-hour windows alternating a cheap overnight block with an expensive daytime one.""" + charge_window_best = [] + for n in range(0, 48): + off_peak = (n % 24) > 12 + price = price_cheap if off_peak else price_peak + charge_window_best.append({"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": price}) + return charge_window_best + + +def run_plan(my_predbat, charge_window_best, battery_charging_from_grid, load_amount=0.5, pv_amount=0.0, battery_size=10.0, battery_soc=5.0): + """Drive a full window optimisation and return the resulting charge limits. + + Deliberately does not assert an exact plan the way run_optimise_all_windows does: these tests are + about an invariant that must hold for every window, not about one particular optimiser outcome, + so pinning exact limits would make them fail on unrelated tuning changes. + """ + end_record = my_predbat.forecast_minutes + my_predbat.calculate_best_charge = True + my_predbat.calculate_best_export = True + my_predbat.soc_max = battery_size + my_predbat.soc_kw = battery_soc + my_predbat.reserve = 0.5 + my_predbat.set_charge_freeze = True + my_predbat.best_soc_keep = 0.0 + my_predbat.debug_enable = False + my_predbat.battery_charging_from_grid = battery_charging_from_grid + + export_window_best = [] + reset_rates(my_predbat, 10.0, 5.5) + update_rates_import(my_predbat, charge_window_best) + update_rates_export(my_predbat, export_window_best) + + pv_step = {} + load_step = {} + for minute in range(0, my_predbat.forecast_minutes, 5): + pv_step[minute] = pv_amount / (60 / 5) + load_step[minute] = load_amount / (60 / 5) + my_predbat.load_minutes_step = load_step + my_predbat.load_minutes_step10 = load_step + my_predbat.pv_forecast_minute_step = pv_step + my_predbat.pv_forecast_minute10_step = pv_step + my_predbat.prediction = Prediction(my_predbat, pv_step, pv_step, load_step, load_step) + + charge_limit_best = [0 for _ in range(len(charge_window_best))] + export_limits_best = [] + metric, _, _, _, _, _, _, _, metric_keep, _, _ = my_predbat.run_prediction(charge_limit_best, charge_window_best, export_window_best, export_limits_best, False, end_record=end_record) + + my_predbat.charge_limit_best = charge_limit_best + my_predbat.export_limits_best = export_limits_best + my_predbat.charge_window_best = charge_window_best + my_predbat.export_window_best = export_window_best + + my_predbat.optimise_all_windows(metric, metric_keep) + + # Re-simulate the chosen plan so the returned SoC belongs to it, rather than to whichever internal + # candidate simulation happened to run last + final = my_predbat.run_prediction(my_predbat.charge_limit_best, my_predbat.charge_window_best, my_predbat.export_window_best, my_predbat.export_limits_best, False, end_record=end_record, save="best") + final_soc = final[5] + return my_predbat.charge_limit_best, my_predbat.charge_window_best, final_soc + + +def test_grid_charge_allowed_window(my_predbat): + """allow_grid_charge_window gates on the switch and exempts only negative-rate windows.""" + print(" - test_grid_charge_allowed_window") + failed = False + windows = [{"start": 0, "end": 30, "average": 10.0}, {"start": 30, "end": 60, "average": -2.0}, {"start": 60, "end": 90, "average": 0.0}] + saved = my_predbat.battery_charging_from_grid + + my_predbat.battery_charging_from_grid = True + for window_n in range(len(windows)): + if not my_predbat.allow_grid_charge_window(windows, window_n): + print("ERROR: window {} should be allowed when the switch is on".format(window_n)) + failed = True + + my_predbat.battery_charging_from_grid = False + expected = [False, True, False] # only the negative-rate window is exempt; zero is not negative + for window_n in range(len(windows)): + got = my_predbat.allow_grid_charge_window(windows, window_n) + if got != expected[window_n]: + print("ERROR: window {} allowed should be {} got {}".format(window_n, expected[window_n], got)) + failed = True + + # A group of windows only qualifies if every member is negative + if my_predbat.allow_grid_charge_window(windows, 0, all_n=[0, 1]): + print("ERROR: a mixed group should not be exempt") + failed = True + if not my_predbat.allow_grid_charge_window(windows, 1, all_n=[1]): + print("ERROR: an all-negative group should be exempt") + failed = True + + my_predbat.battery_charging_from_grid = saved + return failed + + +def test_no_grid_charge_plan(my_predbat): + """With the switch off the planner never targets above the reserve; with it on, it does.""" + print(" - test_no_grid_charge_plan") + failed = False + reset_inverter(my_predbat) + + # Baseline: grid charging allowed, so the cheap overnight windows get used to fill the battery + charge_limit_best, _, _ = run_plan(my_predbat, build_windows(my_predbat), battery_charging_from_grid=True) + if not [limit for limit in charge_limit_best if limit > my_predbat.reserve]: + print("ERROR: with grid charging allowed the planner should charge in at least one window") + failed = True + + # Switch off: every window must now be off (0) or a hold (reserve), never a grid charge + charge_limit_best, _, _ = run_plan(my_predbat, build_windows(my_predbat), battery_charging_from_grid=False) + for window_n, limit in enumerate(charge_limit_best): + if limit > my_predbat.reserve: + print("ERROR: window {} has charge limit {} above reserve {} with grid charging off".format(window_n, limit, my_predbat.reserve)) + failed = True + + return failed + + +def test_no_grid_charge_keeps_solar(my_predbat): + """Turning the mode off must not stop the battery filling from surplus solar.""" + print(" - test_no_grid_charge_keeps_solar") + failed = False + reset_inverter(my_predbat) + + # Plenty of PV, small load: the battery should still end the run well above where it started even + # though no window may charge from the grid + _, _, final_soc = run_plan(my_predbat, build_windows(my_predbat), battery_charging_from_grid=False, load_amount=0.2, pv_amount=3.0, battery_soc=1.0) + if final_soc <= 1.0: + print("ERROR: solar should still charge the battery with grid charging off, final SoC {}".format(final_soc)) + failed = True + + return failed + + +def test_no_grid_charge_negative_rate_exemption(my_predbat): + """A negative import rate is the one case where the mode still permits a grid charge.""" + print(" - test_no_grid_charge_negative_rate_exemption") + failed = False + reset_inverter(my_predbat) + + # Half the windows are paid-to-import, the rest are expensive + charge_window_best = [] + for n in range(0, 48): + price = -10.0 if (n % 24) > 12 else 30.0 + charge_window_best.append({"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": price}) + + charge_limit_best, charge_window_out, _ = run_plan(my_predbat, charge_window_best, battery_charging_from_grid=False) + + charged = False + for window_n, limit in enumerate(charge_limit_best): + rate = charge_window_out[window_n]["average"] + if limit > my_predbat.reserve: + charged = True + if rate >= 0: + print("ERROR: window {} at rate {} charged above reserve despite grid charging being off".format(window_n, rate)) + failed = True + if not charged: + print("ERROR: negative-rate windows should still be allowed to charge from the grid") + failed = True + + return failed + + +def test_manual_charge_downgraded(my_predbat): + """A manual charge slot is downgraded to a hold rather than defeating the mode.""" + print(" - test_manual_charge_downgraded") + failed = False + saved_flag = my_predbat.battery_charging_from_grid + saved_manual = my_predbat.manual_charge_times + saved_limits = my_predbat.charge_limit_best + saved_windows = my_predbat.charge_window_best + saved_export_limits = my_predbat.export_limits_best + saved_export_windows = my_predbat.export_window_best + + start = my_predbat.minutes_now + my_predbat.charge_window_best = [{"start": start, "end": start + 30, "average": 10.0}] + my_predbat.charge_limit_best = [0] + my_predbat.export_window_best = [] + my_predbat.export_limits_best = [] + my_predbat.manual_charge_times = [start] + my_predbat.calculate_best_charge = True + my_predbat.calculate_best_export = True + + my_predbat.battery_charging_from_grid = True + my_predbat.optimise_charge_windows_manual() + if my_predbat.charge_limit_best[0] != my_predbat.soc_max: + print("ERROR: manual charge should target soc_max {} got {}".format(my_predbat.soc_max, my_predbat.charge_limit_best[0])) + failed = True + + my_predbat.charge_limit_best = [0] + my_predbat.battery_charging_from_grid = False + my_predbat.optimise_charge_windows_manual() + if my_predbat.charge_limit_best[0] != my_predbat.reserve: + print("ERROR: manual charge should be downgraded to reserve {} got {}".format(my_predbat.reserve, my_predbat.charge_limit_best[0])) + failed = True + + # A negative-rate manual slot is still honoured in full + my_predbat.charge_window_best = [{"start": start, "end": start + 30, "average": -5.0}] + my_predbat.charge_limit_best = [0] + my_predbat.optimise_charge_windows_manual() + if my_predbat.charge_limit_best[0] != my_predbat.soc_max: + print("ERROR: manual charge at a negative rate should still target soc_max got {}".format(my_predbat.charge_limit_best[0])) + failed = True + + my_predbat.battery_charging_from_grid = saved_flag + my_predbat.manual_charge_times = saved_manual + my_predbat.charge_limit_best = saved_limits + my_predbat.charge_window_best = saved_windows + my_predbat.export_limits_best = saved_export_limits + my_predbat.export_window_best = saved_export_windows + return failed + + +def test_is_freeze_charge_backstop(my_predbat): + """execute's is_freeze_charge downgrades an import-requiring target when grid charging is off.""" + print(" - test_is_freeze_charge_backstop") + failed = False + saved_flag = my_predbat.battery_charging_from_grid + saved_rates = my_predbat.rate_import + saved_soc_max = my_predbat.soc_max + saved_soc_percent = my_predbat.soc_percent + saved_reserve_percent = my_predbat.reserve_percent + + my_predbat.soc_max = 10.0 + my_predbat.soc_percent = 50 + my_predbat.reserve_percent = 10 + my_predbat.rate_import = {my_predbat.minutes_now: 20.0} + + # Switch on: a target above SoC is a real charge, and only the reserve counts as a freeze + my_predbat.battery_charging_from_grid = True + if my_predbat.is_freeze_charge(8.0): + print("ERROR: 80% target should be a real charge when grid charging is allowed") + failed = True + if not my_predbat.is_freeze_charge(1.0): + print("ERROR: a limit at the reserve should always be a freeze") + failed = True + + # Switch off: the same target is downgraded to a hold + my_predbat.battery_charging_from_grid = False + if not my_predbat.is_freeze_charge(8.0): + print("ERROR: 80% target should be downgraded to a hold when grid charging is off") + failed = True + # A target at or below SoC needs no import, so it is left alone + if my_predbat.is_freeze_charge(4.0): + print("ERROR: a 40% target below the 50% SoC needs no import and should not be downgraded") + failed = True + # Negative rates are exempt even with the switch off + my_predbat.rate_import = {my_predbat.minutes_now: -3.0} + if my_predbat.is_freeze_charge(8.0): + print("ERROR: a negative import rate should still permit a real charge") + failed = True + + my_predbat.battery_charging_from_grid = saved_flag + my_predbat.rate_import = saved_rates + my_predbat.soc_max = saved_soc_max + my_predbat.soc_percent = saved_soc_percent + my_predbat.reserve_percent = saved_reserve_percent + return failed + + +def run_no_grid_charge_tests(my_predbat): + """Run every battery_charging_from_grid test.""" + print("**** Running no grid charge tests ****\n") + failed = test_grid_charge_allowed_window(my_predbat) + failed |= test_is_freeze_charge_backstop(my_predbat) + failed |= test_manual_charge_downgraded(my_predbat) + if failed: + return failed + failed |= test_no_grid_charge_plan(my_predbat) + if failed: + return failed + failed |= test_no_grid_charge_keeps_solar(my_predbat) + failed |= test_no_grid_charge_negative_rate_exemption(my_predbat) + return failed diff --git a/apps/predbat/tests/test_optimise_all_windows.py b/apps/predbat/tests/test_optimise_all_windows.py index 925312664..c3294e496 100644 --- a/apps/predbat/tests/test_optimise_all_windows.py +++ b/apps/predbat/tests/test_optimise_all_windows.py @@ -35,6 +35,7 @@ def run_optimise_all_windows( best_soc_keep=0.0, best_soc_keep_weight=0.5, second_pass=False, + battery_charging_from_grid=True, ): print("Starting optimise all windows test {}".format(name)) end_record = my_predbat.forecast_minutes @@ -50,6 +51,7 @@ def run_optimise_all_windows( my_predbat.reserve = 0.5 my_predbat.set_charge_freeze = True my_predbat.calculate_second_pass = second_pass + my_predbat.battery_charging_from_grid = battery_charging_from_grid reset_rates(my_predbat, rate_import, rate_export) update_rates_import(my_predbat, charge_window_best) diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index 173d7ca02..be9f839be 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -39,6 +39,7 @@ from tests.test_predheat import test_predheat from tests.test_debug_enable_auto_scope import test_debug_enable_auto_scope from tests.test_charge_hold import run_charge_hold_tests +from tests.test_no_grid_charge import run_no_grid_charge_tests from tests.test_octopus_slots import run_load_octopus_slots_tests from tests.test_multi_car_iog import run_multi_car_iog_tests from tests.test_fetch_config_options import test_fetch_config_options @@ -382,6 +383,7 @@ def main(): ("predheat", test_predheat, "Predheat scheduler and predheat_enable gate tests (#4670)", False), ("debug_enable_auto_scope", test_debug_enable_auto_scope, "debug_enable auto-disable-after-N-hours tests (#4438 review)", False), ("charge_hold", run_charge_hold_tests, "Charge freeze hold modelling tests", False), + ("no_grid_charge", run_no_grid_charge_tests, "battery_charging_from_grid mode tests", False), ("basic_rates", test_basic_rates, "Basic rates tests", False), ("rate_min_forward_calc", test_rate_min_forward_calc, "Rate min forward calc tests", False), ("rate_export_max_forward_calc", test_rate_export_max_forward_calc, "Rate export max forward calc tests", False), diff --git a/docs/apps-yaml.md b/docs/apps-yaml.md index 3c98f1db0..c5642d862 100644 --- a/docs/apps-yaml.md +++ b/docs/apps-yaml.md @@ -1466,6 +1466,7 @@ or - **charge_limit_enable** - Optional switch entity that enables the AC charge upper percent limit. When set, Predbat will turn this switch on whenever it writes a new charge limit value. Used by inverters (such as GivEnergy via GE Cloud) that have a separate enable/disable control for the charge limit register. - **scheduled_charge_enable** - Switch to enable/disable battery charge according to the charge start/end times defined above. - **scheduled_discharge_enable** - Switch to enable/disable battery discharge according to the discharge start/end times defined above. +- **grid_charge_enable** - Optional switch entity for an inverter that can forbid charging the battery from the grid at the device itself. When set, Predbat asserts it every cycle to match **switch.predbat_battery_charging_from_grid**, so the device enforces the rule even if Predbat is restarted or a stale plan is executed. Leave it out for inverters with no such control - Predbat will simply not write it. - **discharge_target_soc** - Set the battery target percent for timed exports, will be written to minimum by Predbat. - **pause_mode** - GivEnergy pause mode register (if present) - **pause_start_time** - scheduled pause start time (only if supported by your inverter) diff --git a/docs/customisation.md b/docs/customisation.md index d7a7a3014..4f83dc2d2 100644 --- a/docs/customisation.md +++ b/docs/customisation.md @@ -461,6 +461,20 @@ once it has been reached or to protect against discharging beyond the set limit. **switch.predbat_set_charge_freeze** (_expert mode_) When turned On will allow Predbat to hold the current battery level while drawing from the grid/solar as an alternative to charging. On by default. +**switch.predbat_battery_charging_from_grid** When turned Off, Predbat will never charge the battery from the grid. On by default, which is the existing behaviour. + +Use this when your tariff or utility programme forbids grid charging outright, rather than setting an artificially high import rate - a fake rate distorts every other +number the planner produces, including the cost of holding and the value of exporting. + +With this turned Off a charge window can only be left off or held at the reserve, so the battery is charged from solar alone. No solar is given up: a hold still takes PV, +and a window that is off runs in demand mode which stores surplus anyway. House load that the battery cannot cover is simply imported, and because the plan prices that +import honestly, Predbat will arrange to run the battery down during the cheapest hours. + +The single exception is a negative import rate. If you are being paid to import, Predbat will still charge from the grid in those windows. + +Where the inverter has its own grid-charging control, Predbat asserts it too (see **grid_charge_enable** in [apps.yaml](apps-yaml.md)), so a stale plan or a restart cannot +reintroduce a grid charge. Note that turning this Off does not stop the car or an iBoost from drawing from the grid - it applies to the battery only. + **switch.predbat_set_export_freeze** When turned On (the default) will allow Predbat to export Solar to the grid rather than charging the battery. **switch.predbat_set_export_freeze_only** (_expert mode_) When turned On forced export is prevented, but export freeze can be used (if enabled) to export excess solar rather than charging the battery.