diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 0bbe49f28..e3a74ae27 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -359,6 +359,9 @@ def __init__(self, base, id=0, quiet=False): self.track_discharge_end = "00:00:00" self.idle_start_minutes = 0 self.idle_end_minutes = 0 + # Last export schedule (start, end, enabled) Predbat actually committed to this inverter. + # None means nothing has been committed yet in this run, so the next call commits once. + self.last_export_schedule_committed = None self.inverter_type = self.base.get_arg("inverter_type", "GE", indirect=False, index=self.id) # An assumed type is not a chosen one. The user-facing "source returned no data" warnings @@ -2073,7 +2076,7 @@ def adjust_battery_target(self, soc, isCharging=False, isExporting=False): # For inverters that need a button press to apply changes (e.g., Fox), press the button now if self.inv_time_button_press: - self.press_and_poll_button() + self.press_and_poll_button(side="charge") if self.base.set_inverter_notify: self.base.call_notify(f"{self.base.prefix.capitalize()}: Inverter {self.id} Target SoC has been changed to {soc}% at {self.base.time_now_str()}") @@ -2447,7 +2450,7 @@ def adjust_pause_mode(self, pause_charge=False, pause_discharge=False): self.base.log("Inverter {} set pause mode to {}".format(self.id, new_pause_mode)) - def adjust_inverter_mode(self, force_export): + def adjust_inverter_mode(self, force_export, changed_start_end=False): """ Adjust inverter mode between force export and Demand mode (ECO) @@ -2465,6 +2468,10 @@ def adjust_inverter_mode(self, force_export): """ inverter_mode_configured = "inverter_mode" in self.base.args + if changed_start_end: + # XXX: Workaround for GivTCP window state update time to take effort + self.base.log("Sleeping (workaround) as start/end of discharge window was just adjusted") + self.sleep(30) old_inverter_mode = self.base.get_arg("inverter_mode", index=self.id) if not self.inv_has_fox_inverter_mode and not self.inv_has_ge_eco_toggle: @@ -2628,6 +2635,12 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No self.log("Warn: Inverter {} unable read discharge window as neither REST, discharge_start_time or discharge_start_hour are set".format(self.id)) return False + # Whether the caller asked us to manage the export times at all this cycle. execute.py calls + # adjust_force_export(False) with no times whenever nothing is being exported, which is a + # different thing from both an explicit caller-supplied window and the plain-GS midnight + # disable window synthesised below. + times_supplied = (new_start_time is not None) or (new_end_time is not None) + # Start time to correct format if new_start_time: new_start_time += timedelta(seconds=self.base.inverter_clock_skew_discharge_start * 60) @@ -2662,9 +2675,11 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No if self.inv_charge_control_immediate: self.enable_charge_discharge_with_time_current("discharge", False) + schedule_write_ok = True + # Turn off scheduled discharge if not force_export and old_discharge_enable: - self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), False) + schedule_write_ok = self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), False) and schedule_write_ok self.log("Inverter {} Turning off scheduled export".format(self.id)) self.base.log("Inverter {} Adjust force export to {}, change times from {} - {} to {} - {}".format(self.id, force_export, old_start, old_end, new_start, new_end)) @@ -2684,25 +2699,27 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No # Always write to this as it is the GE default changed_start_end = True entity_discharge_start_time_id = self.base.get_arg("discharge_start_time", indirect=False, index=self.id) - self.write_and_poll_option("discharge_start_time", entity_discharge_start_time_id, new_start) + schedule_write_ok = self.write_and_poll_option("discharge_start_time", entity_discharge_start_time_id, new_start) and schedule_write_ok if self.inv_charge_time_format == "H M": # If the inverter uses hours and minutes then write to these entities too # If the entity is a time entity (e.g. for FB00 firmware), write the full time string instead of just the integer component start_hour_id = self.base.get_arg("discharge_start_hour", indirect=False, index=self.id) if start_hour_id and isinstance(start_hour_id, str) and start_hour_id.startswith("time."): - self.write_and_poll_option("discharge_start_hour", start_hour_id, new_start) - else: - self.write_and_poll_option("discharge_start_hour", start_hour_id, int(new_start[:2])) + schedule_write_ok = self.write_and_poll_option("discharge_start_hour", start_hour_id, new_start) and schedule_write_ok + elif start_hour_id: + schedule_write_ok = self.write_and_poll_option("discharge_start_hour", start_hour_id, int(new_start[:2])) and schedule_write_ok start_minute_id = self.base.get_arg("discharge_start_minute", indirect=False, index=self.id) if start_minute_id and isinstance(start_minute_id, str) and start_minute_id.startswith("time."): - self.write_and_poll_option("discharge_start_minute", start_minute_id, new_start) - else: - self.write_and_poll_option("discharge_start_minute", start_minute_id, int(new_start[3:5])) + schedule_write_ok = self.write_and_poll_option("discharge_start_minute", start_minute_id, new_start) and schedule_write_ok + elif start_minute_id: + schedule_write_ok = self.write_and_poll_option("discharge_start_minute", start_minute_id, int(new_start[3:5])) and schedule_write_ok elif self.inv_charge_time_format == "H:M-H:M": # If the inverter uses hours and minutes then write to these entities too discharge_time = new_start + "-" + new_end - self.write_and_poll_option("discharge_time", self.base.get_arg("discharge_time", indirect=False, index=self.id), discharge_time) + discharge_time_id = self.base.get_arg("discharge_time", indirect=False, index=self.id) + if discharge_time_id: + schedule_write_ok = self.write_and_poll_option("discharge_time", discharge_time_id, discharge_time) and schedule_write_ok else: self.log("Warn: Inverter {} unable write export start time as neither REST or discharge_start_time are set".format(self.id)) @@ -2713,21 +2730,21 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No # Always write to this as it is the GE default changed_start_end = True entity_discharge_end_time_id = self.base.get_arg("discharge_end_time", indirect=False, index=self.id) - self.write_and_poll_option("discharge_end_time", entity_discharge_end_time_id, new_end) + schedule_write_ok = self.write_and_poll_option("discharge_end_time", entity_discharge_end_time_id, new_end) and schedule_write_ok # If the inverter uses hours and minutes then write to these entities too if self.inv_charge_time_format == "H M": # If the entity is a time entity (e.g. for FB00 firmware), write the full time string instead of just the integer component end_hour_id = self.base.get_arg("discharge_end_hour", indirect=False, index=self.id) if end_hour_id and isinstance(end_hour_id, str) and end_hour_id.startswith("time."): - self.write_and_poll_option("discharge_end_hour", end_hour_id, new_end) - else: - self.write_and_poll_option("discharge_end_hour", end_hour_id, int(new_end[:2])) + schedule_write_ok = self.write_and_poll_option("discharge_end_hour", end_hour_id, new_end) and schedule_write_ok + elif end_hour_id: + schedule_write_ok = self.write_and_poll_option("discharge_end_hour", end_hour_id, int(new_end[:2])) and schedule_write_ok end_minute_id = self.base.get_arg("discharge_end_minute", indirect=False, index=self.id) if end_minute_id and isinstance(end_minute_id, str) and end_minute_id.startswith("time."): - self.write_and_poll_option("discharge_end_minute", end_minute_id, new_end) - else: - self.write_and_poll_option("discharge_end_minute", end_minute_id, int(new_end[3:5])) + schedule_write_ok = self.write_and_poll_option("discharge_end_minute", end_minute_id, new_end) and schedule_write_ok + elif end_minute_id: + schedule_write_ok = self.write_and_poll_option("discharge_end_minute", end_minute_id, int(new_end[3:5])) and schedule_write_ok elif self.inv_charge_time_format == "H:M-H:M": pass else: @@ -2758,23 +2775,60 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No # Change scheduled discharge enable if force_export: - self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), True) + schedule_write_ok = self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), True) and schedule_write_ok if not old_discharge_enable: self.log("Inverter {} Turning on scheduled export".format(self.id)) - if (new_end != old_end) or (new_start != old_start) or (force_export != old_discharge_enable) or changed_start_end: + # Whether the export schedule itself actually needs (re)committing to the inverter, as opposed + # to us merely having rewritten the time registers. For H M format those registers are rewritten + # every cycle as a write-reliability workaround (#1529), so changed_start_end is True on every + # cycle of a stable export window and must not be used to gate the commit - on Solis each button + # press zeroes the timed current registers (#4709), and it also triggers the 30s GivTCP sleep in + # adjust_inverter_mode. Tracking what we last committed keeps a stable window quiet while still + # committing once after a restart, when nothing has been committed yet (#4000). + # When neither side is managing times this cycle (e.g. idle cloud/GS_fb00 types), comparing None + # against the time the inverter still reports is never equal and would press the update button on + # every idle cycle for the rest of the day (#2328). Plain GS-style inverters still synthesise and + # manage a midnight disable window above, so their internally generated times must continue to count + # as a real schedule change that needs committing; GE's immediate-control disable path remains + # excluded from that synthetic-window rule, but still counts a caller-supplied window as managed. + times_managed = times_supplied or (not self.inv_has_discharge_enable_time and not self.inv_has_ge_inverter_mode and not force_export) + start_changed = times_managed and new_start != old_start + end_changed = times_managed and new_end != old_end + export_schedule = (new_start, new_end, force_export) + # Separately, whether the start/end times themselves actually moved - used below to gate the + # GivTCP settle sleep, which exists for the window write specifically ("start/end of discharge + # window was just adjusted"). schedule_changed alone is too broad for that: it also goes True on + # a bare scheduled_discharge_enable flip with the window untouched, which doesn't need settling. + times_changed = start_changed or end_changed + schedule_changed = times_changed or (force_export != old_discharge_enable) + if is_hm_format and export_schedule != self.last_export_schedule_committed: + # Only the H M path rewrites unconditionally, so only it needs the extra commit-once-per-run + # safety net; every other format already commits on a real change alone. Treat it as a real + # window write too, so the settle sleep still runs on this first post-restart commit. + schedule_changed = True + times_changed = True + + if schedule_changed: + button_committed = True if self.inv_time_button_press: - self.press_and_poll_button() + button_committed = self.press_and_poll_button(side="discharge") + committed = schedule_write_ok and button_committed + if committed: + # Only remember a commit that actually succeeded, otherwise a failed button press would + # be recorded as done and never retried until the schedule next changes on its own; the + # same applies if one of the time/switch writes failed before the button press. + self.last_export_schedule_committed = export_schedule # Force export, turn it on after we change the window if force_export: - self.adjust_inverter_mode(force_export) + self.adjust_inverter_mode(force_export, changed_start_end=times_changed) if not self.inv_has_charge_enable_time and (self.inv_output_charge_control == "current"): if self.inv_charge_control_immediate: self.enable_charge_discharge_with_time_current("discharge", True) # Notify - if changed_start_end: + if schedule_changed and changed_start_end: if self.base.set_inverter_notify: self.base.call_notify(f"{self.base.prefix.capitalize()}: Inverter {self.id} Export time slot set to {new_start} - {new_end} at time {self.base.time_now_str()}") @@ -2828,7 +2882,7 @@ def disable_charge_window(self, notify=True): else: # Press button if needed if self.inv_time_button_press: - self.press_and_poll_button() + self.press_and_poll_button(side="charge") # Updated cached status to disabled # Don't do it if notify is not set as this is just a temporary call when setting the charge window @@ -3246,14 +3300,22 @@ def adjust_charge_window(self, charge_start_time, charge_end_time, minutes_now): if (new_start != old_start) or (new_end != old_end) or (old_charge_schedule_enable == "off"): # For Solis inverters and fox we also have to press the update_charge_discharge button to send the times to the inverter if self.inv_time_button_press: - self.press_and_poll_button() + self.press_and_poll_button(side="charge") - def press_and_poll_button(self): + def press_and_poll_button(self, side="both"): """ Press charge/discharge update button(s) for the inverter. Priority: 1. charge_discharge_update_button 2. charge_update_button and discharge_update_button + + `side` is one of "charge", "discharge" or "both" - the side that actually changed and + needs its update pressed. A single combined button (schedule_write_button or + charge_discharge_update_button) always covers both sides regardless, so `side` only + matters when separate charge_update_button/discharge_update_button entities are + configured (e.g. some Solis setups) - pressing the unrelated side's button there writes + an unnecessary command to the inverter (and, on hardware that counts these towards flash/ + EEPROM wear, batpred#2328). """ success = True @@ -3266,17 +3328,18 @@ def press_and_poll_button(self): elif entity_id_charge_discharge_updated_button: success = self._press_single_button_and_poll(entity_id_charge_discharge_updated_button) else: - # Try separate charge and discharge buttons - charge_button = self.base.get_arg("charge_update_button", indirect=False, index=self.id) - discharge_button = self.base.get_arg("discharge_update_button", indirect=False, index=self.id) - - if charge_button: - if not self._press_single_button_and_poll(charge_button): - success = False - - if discharge_button: - if not self._press_single_button_and_poll(discharge_button): - success = False + # Try separate charge and discharge buttons - only press the one(s) relevant to `side` + if side in ("charge", "both"): + charge_button = self.base.get_arg("charge_update_button", indirect=False, index=self.id) + if charge_button: + if not self._press_single_button_and_poll(charge_button): + success = False + + if side in ("discharge", "both"): + discharge_button = self.base.get_arg("discharge_update_button", indirect=False, index=self.id) + if discharge_button: + if not self._press_single_button_and_poll(discharge_button): + success = False return success @@ -3284,9 +3347,14 @@ def _press_single_toggle_button(self, entity_id): """ This is just a switch we can toggle to on, it will turn off again automatically. """ - self.base.call_service_wrapper("switch/turn_on", entity_id=entity_id) - self.log(f"Inverter {self.id} pressed toggle button {entity_id}") - return True + result = self.base.call_service_wrapper("switch/turn_on", entity_id=entity_id) + state = self.base.get_state_wrapper(entity_id=entity_id) + if result or (isinstance(state, str) and state.lower() in ["on", "enable", "true"]) or (state is True): + self.log(f"Inverter {self.id} pressed toggle button {entity_id}") + return True + self.base.log(f"Warn: Inverter {self.id} Trying to press toggle button {entity_id} failed") + self.base.record_status(f"Warn: Inverter {self.id} Trying to press toggle button {entity_id} failed", had_errors=True) + return False def _press_single_button_and_poll(self, entity_id): """ diff --git a/apps/predbat/tests/test_inverter.py b/apps/predbat/tests/test_inverter.py index 23b72da37..61e44c846 100644 --- a/apps/predbat/tests/test_inverter.py +++ b/apps/predbat/tests/test_inverter.py @@ -1461,6 +1461,81 @@ def adjust_export_immediate(self, target_soc, freeze=False) return failed +def test_press_and_poll_button_side_scoping(my_predbat, inv): + """ + Tests; + def press_and_poll_button(self, side="both") + + Covers GitHub issue #2328: with separate charge_update_button/discharge_update_button + entities configured (rather than a single combined button), pressing one side must not also + press the other - the reporter's log showed both buttons pressed together even when only one + side had anything to update, and on hardware that counts button presses towards flash/EEPROM + wear this doubles the unnecessary writes. Bypasses the real button-press-and-poll mechanism + (its retry/timestamp-polling behaviour is a separate concern from the side-scoping this test + checks) by recording which entity_id(s) press_and_poll_button() actually tries to press. + """ + failed = False + print("**** Running Test: press_and_poll_button_side_scoping ****") + + pressed = [] + + def fake_press(entity_id): + pressed.append(entity_id) + return True + + # my_predbat/inv are shared across the whole test run - save everything this test touches so + # it can be restored exactly, rather than leaking a changed/missing arg or method into later + # tests (e.g. test_force_export_unchanged_times_HM_format relies on schedule_write_button + # already being set from run_inverter_tests' own setup). + unset = object() + real_press_single_button_and_poll = inv._press_single_button_and_poll + saved_args = {key: my_predbat.args.get(key, unset) for key in ("charge_update_button", "discharge_update_button", "charge_discharge_update_button", "schedule_write_button")} + + try: + inv._press_single_button_and_poll = fake_press + + my_predbat.args["charge_update_button"] = "button.charge_update" + my_predbat.args["discharge_update_button"] = "button.discharge_update" + for key in ("charge_discharge_update_button", "schedule_write_button"): + if key in my_predbat.args: + del my_predbat.args[key] + + cases = [ + ("charge", ["button.charge_update"]), + ("discharge", ["button.discharge_update"]), + ("both", ["button.charge_update", "button.discharge_update"]), + ] + for side, expected in cases: + pressed.clear() + inv.press_and_poll_button(side=side) + if sorted(pressed) != sorted(expected): + print(f"ERROR: press_and_poll_button(side='{side}') pressed {pressed}, expected {expected}") + failed = True + else: + print(f" PASS: side='{side}' pressed only {pressed}") + + # Default (no side given) must still press both, for any caller that hasn't been updated + pressed.clear() + inv.press_and_poll_button() + if sorted(pressed) != sorted(["button.charge_update", "button.discharge_update"]): + print(f"ERROR: press_and_poll_button() with no side should press both, got {pressed}") + failed = True + else: + print(" PASS: default side='both' pressed both") + finally: + inv._press_single_button_and_poll = real_press_single_button_and_poll + for key, value in saved_args.items(): + if value is unset: + my_predbat.args.pop(key, None) + else: + my_predbat.args[key] = value + + if not failed: + print("PASS: All press_and_poll_button side scoping tests passed") + + return failed + + def test_call_service_template(test_name, my_predbat, inv, service_name="test", domain="charge", data=None, extra_data=None, clear=True, repeat=False, service_template=None, expected_result=None, twice=True): """ tests @@ -2746,6 +2821,355 @@ def test_force_export_unchanged_times_HM_format(test_name, ha, inv): return failed +def test_force_export_stable_window_presses_button_once(test_name, ha, inv): + """ + Regression test for issue #4709: a stable export window must commit to the inverter once, not on + every Predbat cycle. + + H M format rewrites the time registers unconditionally as a write-reliability workaround (#1529), + so changed_start_end is True on every cycle of an unchanged window. Gating the button press on it + (as #4000 did) pressed the button every 5 minutes for the whole window. On Solis each press zeroes + the timed charge/discharge current registers, stopping export until something restores them, and it + also triggers the 30 second GivTCP settle sleep in adjust_inverter_mode. + + The writes themselves must be preserved - this asserts only that the *commit* stops repeating. + """ + failed = False + print("Test: {}".format(test_name)) + + inv.rest_data = None + inv.inv_charge_time_format = "H M" + inv.inv_time_button_press = True + + export_time = "07:58:00" + export_end = "09:02:00" + + ha.dummy_items["select.discharge_start_time"] = export_time + ha.dummy_items["select.discharge_end_time"] = export_end + ha.dummy_items["time.discharge_start_hour"] = export_time + ha.dummy_items["time.discharge_end_hour"] = export_end + ha.dummy_items["switch.scheduled_discharge_enable"] = "on" + ha.dummy_items["number.discharge_target_soc"] = inv.reserve_percent + ha.dummy_items["select.inverter_mode"] = "Timed Export" + + inv.base.args["discharge_start_time"] = "select.discharge_start_time" + inv.base.args["discharge_end_time"] = "select.discharge_end_time" + inv.base.args["discharge_start_hour"] = "time.discharge_start_hour" + inv.base.args["discharge_end_hour"] = "time.discharge_end_hour" + inv.base.args["scheduled_discharge_enable"] = "switch.scheduled_discharge_enable" + + ts = datetime.strptime(export_time, "%H:%M:%S") + te = datetime.strptime(export_end, "%H:%M:%S") + + # Nothing committed yet, as after a Predbat restart - this cycle must commit the schedule (#4000) + inv.last_export_schedule_committed = None + ha.dummy_items["switch.inverter_button"] = "off" + inv.adjust_force_export(True, ts, te) + + if ha.dummy_items.get("switch.inverter_button") != "on": + print(f"ERROR: {test_name}: first cycle should commit the schedule (button pressed), got {ha.dummy_items.get('switch.inverter_button')}") + failed = True + + # Subsequent cycles of the same, unchanged window must not press the button again (#4709) + for cycle in range(2, 4): + ha.dummy_items["switch.inverter_button"] = "off" + before_writes = inv.count_register_writes + inv.adjust_force_export(True, ts, te) + + if ha.dummy_items.get("switch.inverter_button") != "off": + print(f"ERROR: {test_name}: cycle {cycle} of an unchanged window should not press the button again") + failed = True + + # The time registers are still rewritten - only the commit is suppressed + if inv.count_register_writes < before_writes + 2: + print(f"ERROR: {test_name}: cycle {cycle} should still rewrite the time registers, writes went {before_writes} -> {inv.count_register_writes}") + failed = True + + # A genuine change to the window must commit again + ha.dummy_items["switch.inverter_button"] = "off" + inv.adjust_force_export(True, ts, datetime.strptime("09:32:00", "%H:%M:%S")) + + if ha.dummy_items.get("switch.inverter_button") != "on": + print(f"ERROR: {test_name}: a changed window must be committed, got {ha.dummy_items.get('switch.inverter_button')}") + failed = True + + # A failed button press must not be recorded as committed - it has to be retried next cycle, rather + # than the schedule being treated as applied until it happens to change again on its own. + saved_press = inv.press_and_poll_button + try: + inv.last_export_schedule_committed = None + inv.press_and_poll_button = lambda side="both": False + inv.adjust_force_export(True, ts, te) + + attempts = [] + inv.press_and_poll_button = lambda side="both", _a=attempts: (_a.append(side), True)[1] + inv.adjust_force_export(True, ts, te) + + if not attempts: + print(f"ERROR: {test_name}: a failed commit must be retried on the next cycle, not recorded as done") + failed = True + finally: + inv.press_and_poll_button = saved_press + + # The same retry rule applies when the combined schedule_write_button service itself is rejected. + saved_call_service_wrapper = inv.base.call_service_wrapper + try: + inv.last_export_schedule_committed = None + ha.dummy_items["switch.inverter_button"] = "off" + + def fail_schedule_write_button(service, **kwargs): + if (service == "switch/turn_on") and (kwargs.get("entity_id") == "switch.inverter_button"): + return None + return saved_call_service_wrapper(service, **kwargs) + + inv.base.call_service_wrapper = fail_schedule_write_button + inv.adjust_force_export(True, ts, te) + if inv.last_export_schedule_committed is not None: + print(f"ERROR: {test_name}: a rejected schedule_write_button call must not be recorded as committed") + failed = True + finally: + inv.base.call_service_wrapper = saved_call_service_wrapper + + ha.dummy_items["switch.inverter_button"] = "off" + inv.adjust_force_export(True, ts, te) + if ha.dummy_items.get("switch.inverter_button") != "on": + print(f"ERROR: {test_name}: the same stable window should retry after a rejected schedule_write_button call") + failed = True + + # A failed schedule-register write followed by a successful button press must also retry next cycle, + # rather than caching the tuple and skipping the button while part of the slot is still unapplied. + saved_write_and_poll_option = inv.write_and_poll_option + changed_end = datetime.strptime("09:32:00", "%H:%M:%S") + try: + inv.last_export_schedule_committed = None + ha.dummy_items["switch.inverter_button"] = "off" + + def fail_discharge_end_hour(name, entity_id, new_value, ignore_fail=False): + if name == "discharge_end_hour": + return False + return saved_write_and_poll_option(name, entity_id, new_value, ignore_fail=ignore_fail) + + inv.write_and_poll_option = fail_discharge_end_hour + inv.adjust_force_export(True, ts, changed_end) + if inv.last_export_schedule_committed is not None: + print(f"ERROR: {test_name}: a failed schedule-register write must not be recorded as committed") + failed = True + finally: + inv.write_and_poll_option = saved_write_and_poll_option + + ha.dummy_items["switch.inverter_button"] = "off" + inv.adjust_force_export(True, ts, changed_end) + if ha.dummy_items.get("switch.inverter_button") != "on": + print(f"ERROR: {test_name}: the same stable window should retry after a failed schedule-register write") + failed = True + + inv.last_export_schedule_committed = None + return failed + + +def test_force_export_enable_only_flip_skips_settle_sleep(test_name, ha, inv): + """ + Regression test: turning scheduled export on/off with the window's start/end times unchanged + must not trigger the 30 second GivTCP settle sleep in adjust_inverter_mode. + + That sleep exists specifically for "start/end of discharge window was just adjusted" (its own log + message) - a HH:MM:SS-format inverter (e.g. Fox, GE) whose scheduled_discharge_enable flips while + new_start/new_end already match what's on the inverter has not written any time register at all, + so there is nothing for GivTCP to settle. schedule_changed alone is too broad a gate for this - it + is also True on a bare enable flip - so adjust_force_export must gate the sleep on times_changed + (start/end actually differing) instead, while still pressing the update button and recording the + commit via the broader schedule_changed (the enable state genuinely did change and does need + committing). + """ + failed = False + print("Test: {}".format(test_name)) + + unset = object() + saved_args = {key: inv.base.args.get(key, unset) for key in ("discharge_start_time", "discharge_end_time", "discharge_start_hour", "discharge_end_hour", "scheduled_discharge_enable")} + saved_items = {key: ha.dummy_items.get(key, unset) for key in ("select.discharge_start_time", "select.discharge_end_time", "switch.scheduled_discharge_enable", "number.discharge_target_soc", "select.inverter_mode", "switch.inverter_button")} + saved_rest_data = inv.rest_data + saved_inv_charge_time_format = inv.inv_charge_time_format + saved_inv_time_button_press = inv.inv_time_button_press + saved_last_export_schedule_committed = inv.last_export_schedule_committed + saved_sleep = inv.sleep + + inv.rest_data = None + inv.inv_charge_time_format = "HH:MM:SS" + inv.inv_time_button_press = True + inv.last_export_schedule_committed = None + + export_time = "18:00:00" + export_end = "19:00:00" + + # Window already on the inverter matches what we're about to request - only the enable flag differs. + ha.dummy_items["select.discharge_start_time"] = export_time + ha.dummy_items["select.discharge_end_time"] = export_end + ha.dummy_items["switch.scheduled_discharge_enable"] = "off" + ha.dummy_items["number.discharge_target_soc"] = inv.reserve_percent + ha.dummy_items["select.inverter_mode"] = "Timed Export" + ha.dummy_items["switch.inverter_button"] = "off" + + inv.base.args["discharge_start_time"] = "select.discharge_start_time" + inv.base.args["discharge_end_time"] = "select.discharge_end_time" + inv.base.args.pop("discharge_start_hour", None) + inv.base.args.pop("discharge_end_hour", None) + inv.base.args["scheduled_discharge_enable"] = "switch.scheduled_discharge_enable" + + ts = datetime.strptime(export_time, "%H:%M:%S") + te = datetime.strptime(export_end, "%H:%M:%S") + + sleep_calls = [] + inv.sleep = lambda seconds, _c=sleep_calls: _c.append(seconds) + + try: + inv.adjust_force_export(True, ts, te) + + if ha.dummy_items.get("switch.inverter_button") != "on": + print(f"ERROR: {test_name}: enable flip must still commit (button pressed), got {ha.dummy_items.get('switch.inverter_button')}") + failed = True + + # Other writes (the enable switch itself, the target SoC) legitimately sleep briefly to poll for + # confirmation - inv_write_and_poll_sleep, not the 30s GivTCP settle sleep. Only that specific + # 30-second value is what this test cares about. + if 30 in sleep_calls: + print(f"ERROR: {test_name}: enable-only flip with unchanged times should not trigger the 30s GivTCP settle sleep, got sleep({sleep_calls})") + failed = True + + # Sanity check the sleep gate still fires when the times genuinely do change alongside the flip + ha.dummy_items["switch.scheduled_discharge_enable"] = "off" + ha.dummy_items["switch.inverter_button"] = "off" + sleep_calls.clear() + inv.adjust_force_export(True, ts, datetime.strptime("20:00:00", "%H:%M:%S")) + + if 30 not in sleep_calls: + print(f"ERROR: {test_name}: a genuine window time change must still trigger the 30s settle sleep, got sleep({sleep_calls})") + failed = True + finally: + inv.sleep = saved_sleep + inv.rest_data = saved_rest_data + inv.inv_charge_time_format = saved_inv_charge_time_format + inv.inv_time_button_press = saved_inv_time_button_press + inv.last_export_schedule_committed = saved_last_export_schedule_committed + for key, value in saved_args.items(): + if value is unset: + inv.base.args.pop(key, None) + else: + inv.base.args[key] = value + for key, value in saved_items.items(): + if value is unset: + ha.dummy_items.pop(key, None) + else: + ha.dummy_items[key] = value + + return failed + + +def test_force_export_off_does_not_press_every_cycle(test_name, ha, my_predbat): + """ + Regression test for issue #2328: with no export scheduled the update button must not be pressed on + every cycle. + + execute.py calls adjust_force_export(False) with no times whenever nothing is being exported. On an + inverter with has_discharge_enable_time set (GS_fb00, and the FoxCloud/TESLA/Enphase/Deye/Sunsynk/ + AlphaESS cloud types) the midnight override is skipped, so new_start/new_end stay None while the + inverter still reports a real time - and None never compares equal, so every cycle looked like a + change and pressed the button. That is most of the day, not just export windows. + + FoxCloud stands in for the cloud types here. They reach the bug by the same route as GS_fb00 but + settle somewhere different: being HH:MM:SS format they never take the H M unconditional register + rewrite, so the commit-once safety net gated on that format is skipped too and an idle cloud + inverter should press zero times - not the one GS_fb00 spends on its first post-restart commit. + Pinning that separately keeps the cloud behaviour from resting on GS_fb00 happening to share + has_discharge_enable_time. + + Plain GS takes the midnight-override path and is not affected by this None-comparison route, so it + is checked here too to pin the difference down. That same internally generated midnight window still + has to count as a managed schedule: if the inverter drifts away from 00:00-00:00 while export is + already off, Predbat must press the commit button once to re-apply the disable window. + """ + failed = False + print("Test: {}".format(test_name)) + + # my_predbat/ha are shared across the whole test run - save everything this test touches so it + # can be restored exactly, rather than leaking a changed/missing arg or dummy entity value into + # later tests and making results order-dependent. + unset = object() + saved_args = {key: my_predbat.args.get(key, unset) for key in ("inverter_type", "discharge_start_time", "discharge_end_time", "scheduled_discharge_enable")} + saved_items = { + key: ha.dummy_items.get(key, unset) + for key in ("select.discharge_start_time", "select.discharge_end_time", "switch.scheduled_discharge_enable", "select.inverter_mode", "number.discharge_target_soc", "select.idle_start_time", "select.idle_end_time") + } + + # Times for the positive control below - any window will do, it just has to differ from idle. + export_start = datetime.strptime("18:00:00", "%H:%M:%S") + export_end = datetime.strptime("19:00:00", "%H:%M:%S") + + try: + for inverter_type, expected_presses in (("GS_fb00", 1), ("GS", 1), ("FoxCloud", 0)): + my_predbat.args["inverter_type"] = [inverter_type] + inv = Inverter(my_predbat, 0, quiet=True) + inv.rest_data = None + # Only the press count matters here. TestHAInterface applies a service call synchronously, + # so the write-and-poll read-back already sees the new value and the sleep between them is + # pure wall clock - several seconds per register write, across three inverter types. + inv.sleep = lambda seconds: None + + ha.dummy_items["select.discharge_start_time"] = "00:00:00" + ha.dummy_items["select.discharge_end_time"] = "00:00:00" + ha.dummy_items["switch.scheduled_discharge_enable"] = "off" + ha.dummy_items["select.inverter_mode"] = "Eco" + my_predbat.args["discharge_start_time"] = "select.discharge_start_time" + my_predbat.args["discharge_end_time"] = "select.discharge_end_time" + my_predbat.args["scheduled_discharge_enable"] = "switch.scheduled_discharge_enable" + + presses = [] + # Must report success, as a real press does - a falsy return means "not committed, retry" + inv.press_and_poll_button = lambda side="both", _p=presses: (_p.append(side), True)[1] + + # Several cycles with nothing to export - the first may commit, the rest must be silent + for _ in range(4): + inv.adjust_force_export(False) + + if len(presses) > expected_presses: + print(f"ERROR: {test_name}: {inverter_type} pressed the button {len(presses)} times over 4 idle cycles, expected at most {expected_presses}") + failed = True + + # Positive control: the suppression has to be specific to idle cycles. An inverter that had + # stopped committing altogether would satisfy the check above just as well, so entering an + # export window straight afterwards must still commit, exactly once. + idle_presses = len(presses) + inv.adjust_force_export(True, export_start, export_end) + if len(presses) != idle_presses + 1: + print(f"ERROR: {test_name}: {inverter_type} should press once on entering an export window, pressed {len(presses) - idle_presses} times") + failed = True + + if inverter_type == "GS": + # Once the idle disable window has been committed, drifting away from midnight while + # export is still off must trigger one more commit to restore it. + inv.last_export_schedule_committed = ("00:00:00", "00:00:00", False) + ha.dummy_items["select.discharge_start_time"] = "03:33:00" + ha.dummy_items["select.discharge_end_time"] = "04:44:00" + ha.dummy_items["switch.scheduled_discharge_enable"] = "off" + drift_presses = len(presses) + inv.adjust_force_export(False) + if len(presses) != drift_presses + 1: + print(f"ERROR: {test_name}: {inverter_type} should re-commit a drifted midnight disable window once") + failed = True + finally: + for key, value in saved_args.items(): + if value is unset: + my_predbat.args.pop(key, None) + else: + my_predbat.args[key] = value + for key, value in saved_items.items(): + if value is unset: + ha.dummy_items.pop(key, None) + else: + ha.dummy_items[key] = value + + return failed + + def test_time_entity_hour_write(test_name, ha, inv, dummy_rest, direction, new_start, new_end): """ Test that when *_start_hour / *_end_hour args resolve to time.* entities the full @@ -3875,6 +4299,10 @@ def run_inverter_tests(my_predbat_dummy): if failed: return failed + failed |= test_press_and_poll_button_side_scoping(my_predbat, inv) + if failed: + return failed + failed |= test_auto_restart( "auto_restart0", my_predbat, @@ -4061,6 +4489,19 @@ def run_inverter_tests(my_predbat_dummy): if failed: return failed + # Regression test for issue #4709: a stable export window must be committed once, not every cycle + failed |= test_force_export_stable_window_presses_button_once("force_export_stable_window_button_once", ha, inv) + if failed: + return failed + + # Regression test: an enable-only flip (times unchanged) must not trigger the GivTCP settle sleep + failed |= test_force_export_enable_only_flip_skips_settle_sleep("force_export_enable_only_flip_skips_settle_sleep", ha, inv) + + # Regression test for issue #2328: idle (non-export) cycles must not press the button every time + failed |= test_force_export_off_does_not_press_every_cycle("force_export_off_no_repeat_press", ha, my_predbat) + if failed: + return failed + # Regression test: export target SoC must track the minimum reserve SoC in both directions failed |= test_discharge_target_tracks_reserve("discharge_target_tracks_reserve", ha, inv, dummy_rest) if failed: