diff --git a/apps/predbat/deye.py b/apps/predbat/deye.py index a26d81cdc..d769b8766 100644 --- a/apps/predbat/deye.py +++ b/apps/predbat/deye.py @@ -26,6 +26,7 @@ from component_base import ComponentBase from mock_base import MockBase from oauth_mixin import OAuthMixin +from tou_schedule import TouScheduleMixin from deye_const import ( DEYE_BASE_URLS, DEYE_ENDPOINTS, @@ -37,7 +38,6 @@ FREEZE_EXPORT_SOC, TOU_FIELD, TOU_SLOT_COUNT, - TOU_FILLER_TIMES, DEYE_TOU_DAYS, DEYE_ORDER_MAX_POLLS, DEYE_BUSY_CODES, @@ -66,7 +66,7 @@ ) -class DeyeAPI(ComponentBase, OAuthMixin): +class DeyeAPI(ComponentBase, OAuthMixin, TouScheduleMixin): """DEYE Cloud API component.""" # Trace every API request/response while the DEYE integration beds in; flip to @@ -74,6 +74,9 @@ class DeyeAPI(ComponentBase, OAuthMixin): # caller (tests, the CLI tool) builds the object without going through initialize(). api_debug = True + # How many slots the TOU programme holds, read by TouScheduleMixin.build_tou_slots. + TOU_SLOTS = TOU_SLOT_COUNT + def initialize( self, app_id="", @@ -709,57 +712,6 @@ def _slot_for(self, start_time, state, reserve, self_use_power): return self._action_slot(start_time, state) return self._self_use_slot(start_time, reserve, self_use_power) - def build_tou_slots(self, schedule, current_soc, self_use_power): - """Build exactly TOU_SLOT_COUNT ordered slots covering 24h from the schedule windows.""" - reserve = int(schedule.get("reserve", 0)) - # Collect (start_time, state) segment boundaries, from a DERIVED baseline at 00:00. - # "No window is active" is not always demand - a freeze export is exactly that state - # plus a zero charge rate (see derive_control_state) - so every slot the schedule - # does not otherwise claim carries the baseline, and a freeze covers the whole - # programme instead of being defeated by the first filler that covers the current - # time. Predbat never says when a freeze ends, so there is no boundary to write. - baseline = self.derive_control_state({"reserve": reserve, "charge": {"enable": False, "power": int(schedule.get("charge", {}).get("power", 0))}, "export": {"enable": False, "power": int(schedule.get("export", {}).get("power", 0))}}, current_soc) - segments = {"00:00": dict(baseline)} - for direction in ("charge", "export"): - window = schedule.get(direction, {}) - if window.get("enable") and window.get("start") and window.get("end"): - # Normalised to HH:MM here: these strings become DEYE slot times, and the - # entities they came from carry seconds. - start_time = self._to_slot_time(window["start"]) - end_time = self._to_slot_time(window["end"]) - if start_time == end_time: - # Mirrors the guard in _window_active: a zero-length window has no - # interval to act over. Compared on the NORMALISED times, so "02:00:00" - # against "02:00" is caught too. Without this, an enable event arriving - # before the time fields (both still the "00:00:00" default) would add an - # action segment whose matching return-to-self-use segment cannot be - # added at the same key — an unterminated, multi-hour full-power - # grid-charge/export slot, even though _active_state correctly reports - # the window inactive. - continue - intent = {"reserve": reserve, "charge": {"enable": False}, "export": {"enable": False}} - intent[direction] = {"enable": True, "soc": window.get("soc", 0), "power": window.get("power", 0)} - segments[start_time] = self.derive_control_state(intent, current_soc) - # After the window, return to the baseline. - segments.setdefault(end_time, dict(baseline)) - ordered = sorted(segments.items(), key=lambda kv: kv[0]) - slots = [] - for start_time, state in ordered: - slots.append(self._slot_for(start_time, state, reserve, self_use_power)) - # Normalise to exactly TOU_SLOT_COUNT slots, each with a DISTINCT ascending - # start time (DEYE rejects/mis-applies duplicate slot times). Pad with - # self-use slots at filler times not already used by a window boundary, - # then sort and trim keeping the earliest (imminent) slots. - used = {slot[TOU_FIELD["time"]] for slot in slots} - for filler_time in TOU_FILLER_TIMES: - if len(slots) >= TOU_SLOT_COUNT: - break - if filler_time not in used: - slots.append(self._slot_for(filler_time, baseline, reserve, self_use_power)) - used.add(filler_time) - slots = sorted(slots, key=lambda slot: slot[TOU_FIELD["time"]])[:TOU_SLOT_COUNT] - return slots - def _now_minutes(self): """Return minutes since local midnight, for time-aware window selection.""" try: @@ -767,65 +719,6 @@ def _now_minutes(self): except (TypeError, ValueError): return 0 - @staticmethod - def _to_slot_time(value): - """Normalise a schedule time to the HH:MM DEYE's TOU slots require. - - The control entities carry HH:MM:SS because that is the format Predbat writes (see - INVERTER_DEF charge_time_format), but DEYE's timeUseSettingItems take HH:MM, so the - seconds are dropped here — at the one point a schedule time becomes a slot time. - """ - text = str(value or "00:00") - parts = text.split(":") - if len(parts) < 2: - return "00:00" - try: - return f"{int(parts[0]):02d}:{int(parts[1]):02d}" - except ValueError: - return "00:00" - - @staticmethod - def _hm_to_minutes(hm): - """Convert a HH:MM string to minutes since midnight (0 on bad input).""" - try: - parts = str(hm).split(":") - return int(parts[0]) * 60 + int(parts[1]) - except (ValueError, IndexError): - return 0 - - def _window_active(self, window, now_minutes): - """Return True if an enabled window covers now_minutes (handles a midnight wrap).""" - if not window.get("enable") or not window.get("start") or not window.get("end"): - return False - start = self._hm_to_minutes(window["start"]) - end = self._hm_to_minutes(window["end"]) - if start == end: - return False - if start < end: - return start <= now_minutes < end - return now_minutes >= start or now_minutes < end # window wraps past midnight - - def _active_state(self, schedule, current_soc, now_minutes): - """Derive the control state for the window active at now_minutes, else idle. - - DEYE has a single global work mode per schedule, so the top-level mode must - follow the window active RIGHT NOW rather than a static export-first - precedence: otherwise an export window enabled elsewhere in the day would - pin the mode to SELLING_FIRST and block the charge window's grid charging. - """ - reserve = int(schedule.get("reserve", 0)) - charge = schedule.get("charge", {}) - export = schedule.get("export", {}) - # The rates are carried even when no window is active: with both windows shut a zero - # charge rate is Predbat's Freeze Export, and the mode has to follow it rather than - # sit in demand (see derive_control_state). - intent = {"reserve": reserve, "charge": {"enable": False, "power": int(charge.get("power", 0))}, "export": {"enable": False, "power": int(export.get("power", 0))}} - if self._window_active(export, now_minutes): - intent["export"] = {"enable": True, "soc": export.get("soc", 0), "power": export.get("power", 0)} - elif self._window_active(charge, now_minutes): - intent["charge"] = {"enable": True, "soc": charge.get("soc", 0), "power": charge.get("power", 0)} - return self.derive_control_state(intent, current_soc) - def build_dynamic_payload(self, sn, schedule, current_soc, now_minutes=None): """Build the strategy_dynamic_control body for one inverter. diff --git a/apps/predbat/deye_const.py b/apps/predbat/deye_const.py index f173ffcd7..e7dd9df5d 100644 --- a/apps/predbat/deye_const.py +++ b/apps/predbat/deye_const.py @@ -25,14 +25,6 @@ DEYE_RETRIES = 3 TOU_SLOT_COUNT = 6 FREEZE_EXPORT_SOC = 99 - -# Distinct, ascending self-use start times used to pad a schedule out to -# TOU_SLOT_COUNT slots. DEYE's 6 TOU slots are sequential intervals ("from this -# start until the next slot's start"), so every slot must have a UNIQUE start -# time — duplicate times create zero-length/ambiguous intervals the API may -# reject. Seven options guarantee at least TOU_SLOT_COUNT distinct times remain -# after removing any that collide with the schedule's own window boundaries. -TOU_FILLER_TIMES = ["00:00", "04:00", "08:00", "12:00", "16:00", "20:00", "23:00"] # Number of run() cycles a control order may stay unconfirmed before the applied-payload # cache is invalidated and the next apply is forced to re-write. DEYE_ORDER_MAX_POLLS = 3 diff --git a/apps/predbat/sunsynk.py b/apps/predbat/sunsynk.py index a3058639c..6d9cd465f 100644 --- a/apps/predbat/sunsynk.py +++ b/apps/predbat/sunsynk.py @@ -21,13 +21,16 @@ import argparse import asyncio +import os import hashlib import json import time import aiohttp +import yaml from component_base import ComponentBase from mock_base import MockBase from oauth_mixin import OAuthMixin +from tou_schedule import TouScheduleMixin, MINUTES_PER_DAY from sunsynk_const import ( SUNSYNK_REGIONS, SUNSYNK_ENDPOINTS, @@ -55,6 +58,7 @@ LIFEPO4_NOMINAL_VOLTS_PER_CELL, SUNSYNK_WORKMODE, SUNSYNK_WORKMODE_FIELD, + SUNSYNK_TOU_TEST_SETTLE, SUNSYNK_SOLAR_SELL_FIELD, SUNSYNK_TOU_ENABLE_FIELD, SUNSYNK_SERIAL_FIELD, @@ -63,7 +67,6 @@ SUNSYNK_DAY_FIELDS, TOU_FIELD, TOU_SLOT_COUNT, - TOU_FILLER_TIMES, FREEZE_EXPORT_SOC, SUNSYNK_SETTLE_POLLS, SUNSYNK_TTL_STATIC, @@ -80,7 +83,7 @@ ) -class SunsynkAPI(ComponentBase, OAuthMixin): +class SunsynkAPI(ComponentBase, OAuthMixin, TouScheduleMixin): """Sunsynk Connect cloud API component.""" # Trace every API request/response while the Sunsynk integration beds in. Nobody on @@ -88,6 +91,9 @@ class SunsynkAPI(ComponentBase, OAuthMixin): # for the inferred wire format; flip to False once the format is confirmed. api_debug = True + # How many slots the TOU programme holds, read by TouScheduleMixin.build_tou_slots. + TOU_SLOTS = TOU_SLOT_COUNT + def initialize( self, username="", @@ -735,43 +741,6 @@ def _freeze_export_state(reserve): """ return {"behaviour": "freeze_export", "work_mode": SUNSYNK_WORKMODE["selling_first"], "grid_charge": False, "solar_sell": True, "slot_soc": int(reserve), "power": 0} - @staticmethod - def _to_slot_time(value): - """Normalise a schedule time to the HH:MM Sunsynk's slots require. - - The control entities carry HH:MM:SS because that is what Predbat writes (see - INVERTER_DEF charge_time_format), so the seconds are dropped here — at the one - point a schedule time becomes a slot time. - """ - parts = str(value or "00:00").split(":") - if len(parts) < 2: - return "00:00" - try: - return f"{int(parts[0]):02d}:{int(parts[1]):02d}" - except ValueError: - return "00:00" - - @staticmethod - def _hm_to_minutes(hm): - """Convert an HH:MM string to minutes since midnight (0 on bad input).""" - try: - parts = str(hm).split(":") - return int(parts[0]) * 60 + int(parts[1]) - except (ValueError, IndexError): - return 0 - - def _window_active(self, window, now_minutes): - """Return True if an enabled window covers now_minutes, handling a midnight wrap.""" - if not window.get("enable") or not window.get("start") or not window.get("end"): - return False - start = self._hm_to_minutes(self._to_slot_time(window["start"])) - end = self._hm_to_minutes(self._to_slot_time(window["end"])) - if start == end: - return False - if start < end: - return start <= now_minutes < end - return now_minutes >= start or now_minutes < end - def _self_use_slot(self, start_time, reserve, self_use_power): """Build a self-use slot holding at the reserve SOC. @@ -805,69 +774,6 @@ def _slot_for(self, start_time, state, reserve, self_use_power): return self._action_slot(start_time, state) return self._self_use_slot(start_time, reserve, self_use_power) - def build_tou_slots(self, schedule, current_soc, self_use_power): - """Build exactly TOU_SLOT_COUNT ordered slots covering 24h from the schedule windows. - - Slots are sequential intervals ("from this start until the next slot's start") and - Sunsynk documents that they MUST be set chronologically, so every start is written - distinct and ascending. A slot is an interval whatever its grid-charge flag says, - which is what lets a filler slot terminate the charge window before it. - - Segment boundaries are collected from a 00:00 baseline plus each enabled window's - start (its action) and end (back to the baseline), then padded with fillers and - trimmed to the earliest, most imminent TOU_SLOT_COUNT. - - The baseline is DERIVED rather than assumed to be self-use, because "no window is - active" is not always demand: a freeze export is exactly that state plus a zero - charge rate (see derive_control_state). Every slot the schedule does not otherwise - claim - the 00:00 start, each window's end, and the fillers - therefore carries the - baseline, so a freeze covers the whole programme instead of being defeated by the - first filler that happens to cover the current time. - - That coarseness is deliberate. Predbat never tells this component when a freeze - ends - it disables the export window rather than describing it - so there is no - boundary to write. Pinning the freeze to "now" instead would move every slot time - on every tick and turn the applied-payload change detection into a write per cycle. - The programme is rebuilt whenever Predbat's plan changes (run() applies each tick - for control_active inverters), so it reverts to self-use as soon as the charge rate - comes back. - """ - reserve = int(schedule.get("reserve", 0)) - baseline = self.derive_control_state({"reserve": reserve, "charge": {"enable": False, "power": int(schedule.get("charge", {}).get("power", 0))}, "export": {"enable": False, "power": int(schedule.get("export", {}).get("power", 0))}}, current_soc) - segments = {"00:00": dict(baseline)} - for direction in ("charge", "export"): - window = schedule.get(direction, {}) - if not (window.get("enable") and window.get("start") and window.get("end")): - continue - start_time = self._to_slot_time(window["start"]) - end_time = self._to_slot_time(window["end"]) - if start_time == end_time: - # Mirrors the guard in _window_active: a zero-length window has no interval - # to act over. Compared on the NORMALISED times so "02:00:00" vs "02:00" - # is caught too. Without this, an enable event that arrives before the time - # fields (both still the "00:00:00" default) would add an action segment at - # 00:00 with no matching return-to-self-use segment - an unterminated, - # multi-hour full-power grid-charge/export slot even though _active_state - # correctly reports the window inactive. - continue - intent = {"reserve": reserve, "charge": {"enable": False}, "export": {"enable": False}} - intent[direction] = {"enable": True, "soc": window.get("soc", 0), "power": window.get("power", 0)} - segments[start_time] = self.derive_control_state(intent, current_soc) - segments.setdefault(end_time, dict(baseline)) - - slots = [] - for start_time, state in sorted(segments.items(), key=lambda item: item[0]): - slots.append(self._slot_for(start_time, state, reserve, self_use_power)) - - used = {slot["time"] for slot in slots} - for filler in TOU_FILLER_TIMES: - if len(slots) >= TOU_SLOT_COUNT: - break - if filler not in used: - slots.append(self._slot_for(filler, baseline, reserve, self_use_power)) - used.add(filler) - return sorted(slots, key=lambda slot: slot["time"])[:TOU_SLOT_COUNT] - def _now_minutes(self): """Return minutes since local midnight, for time-aware window selection.""" try: @@ -875,27 +781,6 @@ def _now_minutes(self): except (TypeError, ValueError): return 0 - def _active_state(self, schedule, current_soc, now_minutes): - """Derive the control state for the window active at now_minutes, else idle. - - Sunsynk has a single global work mode, so the top-level mode must follow the - window active RIGHT NOW rather than a static export-first precedence: otherwise - an export window enabled elsewhere in the day would pin the mode to selling-first - and block the charge window's grid charging. - """ - reserve = int(schedule.get("reserve", 0)) - charge = schedule.get("charge", {}) - export = schedule.get("export", {}) - # The charge rate is carried even when no window is active: with both windows shut - # a zero rate is Predbat's Freeze Export, and the mode has to follow it rather than - # sit in demand (see derive_control_state). - intent = {"reserve": reserve, "charge": {"enable": False, "power": int(charge.get("power", 0))}, "export": {"enable": False, "power": int(export.get("power", 0))}} - if self._window_active(export, now_minutes): - intent["export"] = {"enable": True, "soc": export.get("soc", 0), "power": export.get("power", 0)} - elif self._window_active(charge, now_minutes): - intent["charge"] = {"enable": True, "soc": charge.get("soc", 0), "power": charge.get("power", 0)} - return self.derive_control_state(intent, current_soc) - def _owned_payload(self, sn, schedule, current_soc, now_minutes): """Build only the fields Predbat owns, with ZERO network I/O. @@ -1777,6 +1662,59 @@ async def final(self): await self.save_control() +# apps.yaml key -> the SunsynkAPI keyword it supplies. Only what is needed to authenticate +# and address one inverter; nothing else in the file is read. +APPS_YAML_CREDENTIAL_KEYS = { + "sunsynk_username": "username", + "sunsynk_password": "password", + "sunsynk_key": "key", + "sunsynk_region": "region", + "sunsynk_auth_method": "auth_method", + "sunsynk_token_expires_at": "token_expires_at", + "sunsynk_token_hash": "token_hash", + "sunsynk_inverter_sn": "inverter_sn", +} + + +def load_apps_yaml_credentials(path): + """Read Sunsynk credentials out of a real apps.yaml for the standalone CLI. + + Lets the CLI drive a real account without retyping credentials, and in particular + without a password at all: an install using the oauth flow has only a bearer token, and + --username/--password cannot express that. Returns a kwargs dict for SunsynkAPI, or + raises ValueError with the reason when the file cannot supply a usable account - the + caller is a CLI, and a precise refusal beats a half-built client failing at login. + + Only the keys in APPS_YAML_CREDENTIAL_KEYS are read, and none of their values are ever + printed: the CLI reports the serial and the region, never the token. + """ + try: + with open(path, "r") as handle: + config = yaml.safe_load(handle) or {} + except (OSError, yaml.YAMLError) as error: + raise ValueError(f"could not read {path}: {error}") + if not isinstance(config, dict): + raise ValueError(f"{path} is not a YAML mapping") + # A real apps.yaml nests everything under pred_bat; accept a bare mapping too, so a + # hand-trimmed file holding just the sunsynk keys works. + config = config.get("pred_bat", config) or {} + credentials = {name: config[key] for key, name in APPS_YAML_CREDENTIAL_KEYS.items() if config.get(key) not in (None, "")} + + serials = credentials.get("inverter_sn") + if not serials: + raise ValueError(f"{path} has no sunsynk_inverter_sn, so there is no inverter to address") + credentials["inverter_sn"] = [str(serial) for serial in (serials if isinstance(serials, list) else [serials])] + if credentials.get("token_expires_at") is not None: + # yaml parses an ISO timestamp into a datetime; _parse_expiry wants the string back. + credentials["token_expires_at"] = str(credentials["token_expires_at"]) + if credentials.get("auth_method") == "oauth": + if not credentials.get("key"): + raise ValueError(f"{path} selects the oauth flow but carries no sunsynk_key token") + elif not (credentials.get("username") and credentials.get("password")): + raise ValueError(f"{path} has neither an oauth token nor a sunsynk_username/sunsynk_password pair") + return credentials + + def _build_sunsynk(mock_base, args): # pragma: no cover """Construct a SunsynkAPI around a MockBase for standalone command-line use. @@ -1790,13 +1728,21 @@ def _build_sunsynk(mock_base, args): # pragma: no cover _build_deye does, so a serial restricts discovery itself (run() -> refresh_static() -> get_device_list()) rather than being trusted unverified after the fact. """ + credentials = dict(getattr(args, "credentials", None) or {}) + # --serial still wins over the file, so one inverter of a multi-inverter account can be + # singled out without editing apps.yaml. + if args.serial: + credentials["inverter_sn"] = args.serial return SunsynkAPI( mock_base, - username=args.username, - password=args.password, - region=args.region, - auth_method=args.auth_method, - inverter_sn=args.serial, + username=credentials.pop("username", args.username), + password=credentials.pop("password", args.password), + region=credentials.pop("region", args.region), + auth_method=credentials.pop("auth_method", args.auth_method), + inverter_sn=credentials.pop("inverter_sn", None), + key=credentials.pop("key", ""), + token_expires_at=credentials.pop("token_expires_at", None), + token_hash=credentials.pop("token_hash", ""), # The CLI is the verification tool, so control is on - but test_sunsynk_api still # asks before it sends anything to a real inverter, and run() only ever calls # apply_settings/apply_schedule for a serial already in control_active, which @@ -1806,6 +1752,123 @@ def _build_sunsynk(mock_base, args): # pragma: no cover ) +def _tou_test_window(now_minutes, lead_minutes=240, length_minutes=180): + """Return an HH:MM charge window placed clear of the current time. + + The programme --tou-test writes exists only to be read back and removed, so it must + never be one the inverter acts on. Starting a few hours ahead and running for a few + more keeps the current minute outside it however the window wraps past midnight, so + the battery cannot enter the window during the seconds it is live. + """ + start = (now_minutes + lead_minutes) % MINUTES_PER_DAY + end = (start + length_minutes) % MINUTES_PER_DAY + return f"{start // 60:02d}:{start % 60:02d}", f"{end // 60:02d}:{end % 60:02d}" + + +def _tou_snapshot(settings): + """Take the System Mode subset of a settings read - everything a restore needs.""" + return {key: value for key, value in (settings or {}).items() if key in SUNSYNK_SYSTEM_MODE_FIELDS} + + +def _tou_slot_times(settings): + """Return the six slot start times from a settings object, for printing.""" + return [(settings or {}).get(TOU_FIELD["time"].format(n=n)) for n in range(1, TOU_SLOT_COUNT + 1)] + + +async def _tou_round_trip(client, sn): # pragma: no cover + """Write a padded TOU programme to a real inverter, verify it, then restore the original. + + This is the only thing in the tree that proves the slot builder against real firmware + rather than a double. The mocked tests can say what build_tou_slots produces; only this + says the inverter stored it. + + The original programme is written to a JSON file BEFORE anything is sent, the restore + runs from a finally, and the restore is itself verified by reading back - so a failure + anywhere still leaves the inverter as it was found, or names the file that will put it + back by hand. + """ + original = await client.fetch_settings(sn) + snapshot = _tou_snapshot(original) + if len(snapshot) < TOU_SLOT_COUNT: + print(f"\nNo System Mode programme read back for {sn} ({len(snapshot)} owned keys), refusing to write.") + return False + snapshot_path = f"sunsynk_tou_restore_{sn}_{time.strftime('%Y%m%d_%H%M%S')}.json" + with open(snapshot_path, "w") as handle: + json.dump(snapshot, handle, indent=2, sort_keys=True, default=str) + print(f"\n--- {sn} TOU round trip ---") + print(f"Current programme: {_tou_slot_times(original)}") + print(f"Restore snapshot saved to {os.path.abspath(snapshot_path)}") + + now_minutes = client._now_minutes() + start, end = _tou_test_window(now_minutes) + schedule = { + "reserve": max(client.battery_reserve_min(sn), 10), + "charge": {"enable": True, "soc": 95, "power": int(client.battery_rate_max(sn)) or 3000, "start": start, "end": end}, + "export": {"enable": False, "soc": 0, "power": 0, "start": "00:00:00", "end": "00:00:00"}, + } + self_use_power = int(client.inverter_limit(sn)) or client._existing_slot_power(sn) + expected = client.build_tou_slots(schedule, current_soc=50, self_use_power=self_use_power) + print(f"Writing charge {start}-{end} (now {now_minutes // 60:02d}:{now_minutes % 60:02d}, deliberately outside the window)") + print(f"Expected programme: {[(slot['time'], 'charge' if slot['grid_charge'] else 'self-use') for slot in expected]}") + print(" (the slots after the window are padding - they must carry the state they follow, not end it)") + try: + confirm = input("Send this to the inverter and then restore the original? [y/N] ") + except (EOFError, KeyboardInterrupt): + print("\nNo input available, nothing sent.") + os.remove(snapshot_path) + return False + if confirm.strip().lower() != "y": + print("Nothing sent.") + os.remove(snapshot_path) + return False + + failed = False + try: + if not await client.apply_settings(sn, schedule, current_soc=50, force=True): + print("FAILED: apply_settings reported no write - the programme did not reach the inverter.") + return True + # Read back rather than trusting the write's own success: this API accepts an + # oversized object and silently discards it, and has been seen to drop individual + # fields while persisting the rest (see SUNSYNK_BOOL_FIELDS), so only a read proves + # the programme landed. + print(f"Written, re-reading in {SUNSYNK_TOU_TEST_SETTLE}s to let the dongle collect it...") + await asyncio.sleep(SUNSYNK_TOU_TEST_SETTLE) + readback = await client.fetch_settings(sn) + if not readback: + print("FAILED: could not read the settings back after writing.") + return True + sent = client.applied_payload.get(sn, {}) + asserted = [TOU_FIELD[field].format(n=index) for index in range(1, TOU_SLOT_COUNT + 1) for field in ("time", "grid_charge", "sell")] + for key in asserted: + if str(readback.get(key)) != str(sent.get(key)): + print(f"FAILED: {key} read back as {readback.get(key)!r}, wrote {sent.get(key)!r}") + failed = True + for key, value in sent.items(): + # Reported, not failed: the inverter may normalise a power or a voltage, and + # that is not what this test is about. + if key not in asserted and key != SUNSYNK_SERIAL_FIELD and str(readback.get(key)) != str(value): + print(f" note: {key} read back as {readback.get(key)!r}, wrote {value!r}") + if not failed: + print(f"VERIFIED: the inverter stored {_tou_slot_times(readback)} - the window runs unbroken through its padding.") + finally: + print("Restoring the original programme...") + if await client._post("settings_set", sn=sn, body=snapshot) is None: + print(f"RESTORE FAILED - the inverter is still on the test programme. Restore by hand from {os.path.abspath(snapshot_path)}") + failed = True + else: + await asyncio.sleep(SUNSYNK_TOU_TEST_SETTLE) + after = await client.fetch_settings(sn) + differing = [key for key, value in snapshot.items() if str((after or {}).get(key)) != str(value)] + if differing: + print(f"RESTORE INCOMPLETE - {len(differing)} field(s) still differ: {differing[:10]}") + print(f" the original is saved at {os.path.abspath(snapshot_path)}") + failed = True + else: + print(f"Restored: {_tou_slot_times(after)}") + os.remove(snapshot_path) + return failed + + async def test_sunsynk_api(args): # pragma: no cover """Run one read-only Sunsynk cycle via run(), then dump what it discovered. @@ -1830,13 +1893,17 @@ async def test_sunsynk_api(args): # pragma: no cover - a token but no device_list -> login worked, discovery found nothing (or failed - see self.discovery_ok, Fix 3). - a device_list but run() still failed -> the first telemetry poll came back empty. - These three are exhaustive for the auth methods this CLI exposes (password, - password_legacy): check_and_refresh_oauth_token() only ever returns False for the - oauth flow, which --auth-method cannot select. + These three are exhaustive for the password flows (--auth-method password, + password_legacy). --apps-yaml can additionally supply an oauth account, where + check_and_refresh_oauth_token() can itself return False - an expired token with no + usable refresh reports as the no-access-token case above, which is the right diagnosis: + the fix is a fresh sunsynk_key in apps.yaml. """ mock_base = MockBase() client = _build_sunsynk(mock_base, args) - print(f"Region {args.region} -> {client.base_url} (source={client.source}), auth={args.auth_method}") + # client.auth_method, not args.auth_method: --apps-yaml can override it, and reporting + # the CLI default while running an oauth account makes a login failure undiagnosable. + print(f"Region {client.region} -> {client.base_url} (source={client.source}), auth={client.auth_method}") print("Calling run() once (read-only: login, discover, poll config/telemetry, publish)...") ok = await client.run(seconds=0, first=True) @@ -1862,6 +1929,7 @@ async def test_sunsynk_api(args): # pragma: no cover return serials = client.device_list + failures = [] print(f"Inverters: {serials}") for sn in serials: print(f"\n--- {sn} detail ---") @@ -1899,7 +1967,12 @@ async def test_sunsynk_api(args): # pragma: no cover if confirm.strip().lower() == "y": await client.apply_settings(sn, schedule, current_soc=50, force=True) print("Written. Re-reading in 60 seconds is the only way to confirm the dongle collected it.") + if args.tou_test: + if await _tou_round_trip(client, sn): + failures.append(sn) + if failures: + print(f"\nTOU round trip FAILED for {failures}") await client.final() print("Done") @@ -1907,14 +1980,26 @@ async def test_sunsynk_api(args): # pragma: no cover def main(): # pragma: no cover """Command-line entry point for Sunsynk diagnostics.""" parser = argparse.ArgumentParser(description="Sunsynk Cloud API diagnostics") - parser.add_argument("--username", required=True, help="Sunsynk Connect account e-mail") - parser.add_argument("--password", required=True, help="Sunsynk Connect account password") + parser.add_argument("--apps-yaml", default=None, help="Read credentials from a real apps.yaml instead of --username/--password (the only way to use an oauth-token account)") + parser.add_argument("--username", default=None, help="Sunsynk Connect account e-mail (not needed with --apps-yaml)") + parser.add_argument("--password", default=None, help="Sunsynk Connect account password (not needed with --apps-yaml)") parser.add_argument("--region", default="sunsynk", choices=sorted(SUNSYNK_REGIONS), help="API region") parser.add_argument("--auth-method", default="password", choices=["password", "password_legacy"], help="Login flow: RSA-encrypted (default) or the pre-2025 plaintext one") parser.add_argument("--serial", default=None, help="Restrict to one inverter serial") parser.add_argument("--dump-settings", action="store_true", help="Print the full settings object") parser.add_argument("--write-test", action="store_true", help="Build a harmless self-use payload and offer to send it") - asyncio.run(test_sunsynk_api(parser.parse_args())) + parser.add_argument("--tou-test", action="store_true", help="Write a padded TOU programme to the inverter, verify the read-back, then restore the original (asks first)") + args = parser.parse_args() + args.credentials = None + if args.apps_yaml: + try: + args.credentials = load_apps_yaml_credentials(args.apps_yaml) + except ValueError as error: + parser.error(str(error)) + print(f"Credentials loaded from {args.apps_yaml} for inverter(s) {args.credentials['inverter_sn']}") + elif not (args.username and args.password): + parser.error("--username and --password are required unless --apps-yaml is given") + asyncio.run(test_sunsynk_api(args)) if __name__ == "__main__": diff --git a/apps/predbat/sunsynk_const.py b/apps/predbat/sunsynk_const.py index b427744e0..50a3861b7 100644 --- a/apps/predbat/sunsynk_const.py +++ b/apps/predbat/sunsynk_const.py @@ -57,27 +57,6 @@ TOU_SLOT_COUNT = 6 FREEZE_EXPORT_SOC = 99 -# Distinct ascending start times used to pad a schedule out to TOU_SLOT_COUNT. -# -# CONFIRMED by Sunsynk's own documentation ("Avoiding conflicts in the System Mode timer"): -# the six slots are sequential chronological intervals, each running until the next slot -# begins, and "Timers MUST be set chronologically from Timer 1 to Timer 6". Timer 6 is the -# only one permitted to roll over midnight and continue until Timer 1 restarts. So distinct -# ascending starts are a real requirement, not a Predbat preference. -# -# A slot is an interval regardless of its grid-charge flag: the documented factory default -# runs all six timers as ranges with Grid Charge ticked on only two of them. That is what -# makes Predbat's padding work - a filler slot with grid charge off still TERMINATES the -# charge window before it. -# -# Note an inverter can nonetheless be found sitting with all six slots at 00:00 (an -# unconfigured default). The API stores that happily; it is simply not a chronological -# programme, so no conclusion about valid schedules should be drawn from it. -# -# Seven options guarantee TOU_SLOT_COUNT distinct times survive even if all four of a -# schedule's own window boundaries collide with fillers. -TOU_FILLER_TIMES = ["00:00", "04:00", "08:00", "12:00", "16:00", "20:00", "23:00"] - # CONFIRMED live (inverter 2405116013, 2026-08-18). The Sunsynk app's System Mode screen # lists Work Mode as Selling First / Zero-Export + Limit To Load Only / Limited to Home, and # with "Limited to Home" selected the settings object reported sysWorkMode '2' - so the @@ -146,6 +125,12 @@ ) # Top-level settings keys Predbat owns. +# Seconds the standalone CLI's --tou-test waits between writing the settings object and +# reading it back. A settings write is collected by the dongle rather than applied +# synchronously, so an immediate read shows the PREVIOUS programme and would report a +# perfectly good write as a failure. +SUNSYNK_TOU_TEST_SETTLE = 20 + SUNSYNK_WORKMODE_FIELD = "sysWorkMode" SUNSYNK_SOLAR_SELL_FIELD = "solarSell" SUNSYNK_TOU_ENABLE_FIELD = "peakAndVallery" diff --git a/apps/predbat/tests/test_deye_control.py b/apps/predbat/tests/test_deye_control.py index 1b0d0558b..4478bb893 100644 --- a/apps/predbat/tests/test_deye_control.py +++ b/apps/predbat/tests/test_deye_control.py @@ -9,7 +9,7 @@ """Tests for the DEYE behaviour to work-mode derivation (``derive_control_state``).""" from unittest.mock import patch -from deye_const import DEYE_WORKMODE, FREEZE_EXPORT_SOC, TOU_FIELD, TOU_SLOT_COUNT, TOU_FILLER_TIMES, DEYE_ORDER_MAX_POLLS +from deye_const import DEYE_WORKMODE, FREEZE_EXPORT_SOC, TOU_FIELD, TOU_SLOT_COUNT, DEYE_ORDER_MAX_POLLS from deye_const import CONFIG_BATTERY_KEYS, DEYE_TOU_DAYS from tests.test_deye_api import MockDeye, MOCK_RATED_POWER from tests.test_infra import run_async as run_async_local @@ -401,10 +401,10 @@ def test_run_clears_pending_order_and_count_on_success(): def test_window_times_reach_the_slots_in_deye_format(): """A HH:MM:SS window becomes HH:MM slot times and actually produces a window. - Live regression: every control payload for two hours carried only the filler times - (00:00/04:00/08:00/12:00/16:00/20:00) with grid charge off everywhere — the signature of - build_tou_slots seeing no window at all — because Predbat was writing the window times - to dummy entities this component never reads. + Live regression: every control payload for two hours carried only padding slots with + grid charge off everywhere — the signature of build_tou_slots seeing no window at all — + because Predbat was writing the window times to dummy entities this component never + reads. """ failed = False d = MockDeye() @@ -425,9 +425,10 @@ def test_window_times_reach_the_slots_in_deye_format(): if not any(s[TOU_FIELD["grid_charge"]] and s[TOU_FIELD["soc"]] == 95 for s in slots): print(f"ERROR: no grid-charge slot was produced for the window: {slots}") failed = True - # The all-filler payload is exactly what the bug produced, so assert we are not back there - if times == TOU_FILLER_TIMES[:TOU_SLOT_COUNT]: - print(f"ERROR: slots are pure filler, the window was lost: {times}") + # The window's end must come back too: a payload carrying the start but no return to + # self-use is an unterminated window, the other way this can go wrong. + if "05:30" not in times: + print(f"ERROR: the window's end never reached the slots: {times}") failed = True # A HH:MM window still works, so a stale entity value cannot break the payload @@ -1147,6 +1148,114 @@ async def fake_fail(endpoint_key, body): assert not failed, "test_fetch_tou_config_caches_and_survives_an_unsupported_model" +def _hm(text): + """Convert an HH:MM slot time to minutes past midnight. + + Deliberately not DeyeAPI._hm_to_minutes: this is the test's own reading of the payload, + and sharing the parser would make the check agree with a broken one. + """ + hours, minutes = str(text).split(":") + return int(hours) * 60 + int(minutes) + + +def _slot_in_effect(slots, minute): + """Return the slot DEYE would be running at a given minute past midnight. + + A slot runs from its own start until the next slot's start, so the one in effect is + the latest slot starting at or before the minute; before the first slot the programme + has wrapped round from the last slot of the previous day. + """ + effective = slots[-1] + for slot in slots: + if _hm(slot[TOU_FIELD["time"]]) <= minute: + effective = slot + else: + break + return effective + + +def _acting_at(slots, minute, field, target_soc): + """Return True if the slot running at this minute both sets the flag and holds the target SoC.""" + slot = _slot_in_effect(slots, minute) + return bool(slot[field]) and slot[TOU_FIELD["soc"]] == target_soc + + +def test_padding_slots_do_not_truncate_a_window(): + """Every minute of a window must stay in that window's state, padding included. + + GH#5156: the 6-slot programme used to be padded out at fixed clock hours, and those + slots carried the idle baseline whatever time they landed on. A padding slot inside a + window is a boundary the inverter obeys, so grid charging stopped dead at + 04:00/08:00/12:00 while Predbat's plan still showed the window running to its end - + confirmed against the reporter's captured payload and their inverter's own + Time-of-Use table. + + Checked minute by minute rather than by counting slots, because the bug did not change + how many slots there were or what times they carried: the 12:00 slot was correctly + present, it simply said the wrong thing. Placement is covered in test_tou_schedule; + this is the DEYE encoding's view of the same property. + """ + failed = False + d = MockDeye() + cases = [ + # (label, direction, start, end, target soc) - the two reported reproductions, an + # overnight window (the commonest shape, and the one the old fixed-hour padding + # broke most often), and the export side of the same flaw. + ("reported day 2", "charge", "09:45", "15:30", 100), + ("reported day 1", "charge", "11:15", "15:30", 100), + ("overnight charge", "charge", "00:30", "07:30", 100), + # 11:00-13:00 specifically: under the old fixed-hour padding only a window crossing + # 04:00, 08:00 or 12:00 got a padding slot landing inside it, so an export window + # over 16:00 passed whether or not the bug was present and proved nothing. + ("export over 12:00", "export", "11:00", "13:00", 20), + ] + for label, direction, start, end, target_soc in cases: + sched = _state(reserve=8, charge_power=12288, export_power=12000) + sched[direction] = {"enable": True, "soc": target_soc, "power": 12288, "start": start, "end": end} + slots = d.build_tou_slots(sched, current_soc=40, self_use_power=MOCK_RATED_POWER) + if len(slots) != TOU_SLOT_COUNT: + print(f"ERROR: {label} expected {TOU_SLOT_COUNT} slots got {len(slots)}") + failed = True + continue + # The action flag AND the target SoC: a slot that kept the flag but fell back to the + # reserve SoC would not charge or export either. + field = TOU_FIELD["grid_charge"] if direction == "charge" else TOU_FIELD["sell"] + broke_at = [minute for minute in range(_hm(start), _hm(end)) if not _acting_at(slots, minute, field, target_soc)] + if broke_at: + times = [s[TOU_FIELD["time"]] for s in slots] + first = broke_at[0] + print(f"ERROR: {label} {direction} window {start}-{end} stops at {first // 60:02d}:{first % 60:02d} ({len(broke_at)} minutes lost); slots {times}") + failed = True + assert not failed, "test_padding_slots_do_not_truncate_a_window" + + +def test_padding_slots_outside_a_window_stay_idle(): + """Every slot outside the window is idle in the DEYE encoding, padding included. + + The other half of the GH#5156 fix. Placement is covered in test_tou_schedule; this is + the vendor's view of it - the padding must reach the wire as genuinely idle slots, with + grid charge and the sell flag both off, rather than inheriting a window's action. + """ + failed = False + d = MockDeye() + sched = {"reserve": 8, "charge": {"enable": True, "soc": 100, "power": 12288, "start": "09:45", "end": "15:30"}, "export": {"enable": False, "soc": 0, "power": 12000}} + slots = d.build_tou_slots(sched, current_soc=40, self_use_power=MOCK_RATED_POWER) + for slot in slots: + inside = "09:45" <= slot[TOU_FIELD["time"]] < "15:30" + if inside and not slot[TOU_FIELD["grid_charge"]]: + print(f"ERROR: slot inside the window is not charging: {slot}") + failed = True + if not inside and (slot[TOU_FIELD["grid_charge"]] or slot[TOU_FIELD["sell"]]): + print(f"ERROR: slot outside the window is not idle: {slot}") + failed = True + # The window's own end must return to idle rather than running on. + end_slot = next((slot for slot in slots if slot[TOU_FIELD["time"]] == "15:30"), None) + if end_slot is None or end_slot[TOU_FIELD["grid_charge"]]: + print(f"ERROR: the window end slot must return to idle, got {end_slot}") + failed = True + assert not failed, "test_padding_slots_outside_a_window_stay_idle" + + def run_deye_control_tests(my_predbat): """Run all DEYE control-logic tests.""" failed = False @@ -1154,6 +1263,8 @@ def run_deye_control_tests(my_predbat): ("derive_table", test_derive_control_state_table), ("tou_slots", test_build_tou_slots_charge_window), ("tou_slots_distinct", test_build_tou_slots_times_are_distinct), + ("tou_padding_no_truncation", test_padding_slots_do_not_truncate_a_window), + ("tou_padding_idle_outside", test_padding_slots_outside_a_window_stay_idle), ("payload", test_build_dynamic_payload_and_equality), ("apply_suppress", test_apply_dynamic_control_suppresses_when_unchanged), ("apply_write", test_apply_dynamic_control_writes_and_caches_on_change), diff --git a/apps/predbat/tests/test_sunsynk_config.py b/apps/predbat/tests/test_sunsynk_config.py index 7e5ae14fe..9dee6ef48 100644 --- a/apps/predbat/tests/test_sunsynk_config.py +++ b/apps/predbat/tests/test_sunsynk_config.py @@ -13,6 +13,7 @@ from config import INVERTER_DEF, APPS_SCHEMA from components import COMPONENT_LIST from sunsynk_const import SUNSYNK_TTL_STATIC +from sunsynk import load_apps_yaml_credentials, _tou_test_window from tests.test_sunsynk_api import MockSunsynk from tests.test_sunsynk_publish import PublishingSunsynk from tests.test_infra import run_async as run_async_local @@ -806,11 +807,110 @@ def test_sign_flags_are_claimed_not_inherited(): assert not failed, "test_sign_flags_are_claimed_not_inherited" +def test_apps_yaml_credentials_loader(): + """The standalone CLI's --apps-yaml loader builds a usable account or refuses precisely. + + This is what lets the CLI drive a real oauth account for the live TOU round trip, where + there is a bearer token and no password at all. A bug here either half-builds a client + that fails at login with a confusing message, or silently drops the serial. + """ + failed = False + import os + import tempfile + + import yaml as yaml_module + + def write(folder, name, payload): + """Write a YAML fixture and return its path.""" + path = os.path.join(folder, name) + with open(path, "w") as handle: + yaml_module.safe_dump(payload, handle) + return path + + with tempfile.TemporaryDirectory() as folder: + oauth = write(folder, "oauth.yaml", {"pred_bat": {"sunsynk_auth_method": "oauth", "sunsynk_key": "token-abc", "sunsynk_token_hash": "hash-abc", "sunsynk_inverter_sn": ["2211093089"], "sunsynk_token_expires_at": "2026-09-19T22:11:54.287+00:00"}}) + credentials = load_apps_yaml_credentials(oauth) + expected = {"auth_method": "oauth", "key": "token-abc", "token_hash": "hash-abc", "inverter_sn": ["2211093089"], "token_expires_at": "2026-09-19T22:11:54.287+00:00"} + if credentials != expected: + print(f"ERROR: oauth config loaded as {credentials}, expected {expected}") + failed = True + + # A YAML-parsed timestamp must come back as the string _parse_expiry wants, not a + # datetime - safe_load turns an unquoted ISO date into one. + from datetime import datetime as real_datetime + + typed = write(folder, "typed.yaml", {"pred_bat": {"sunsynk_auth_method": "oauth", "sunsynk_key": "token-abc", "sunsynk_inverter_sn": ["2211093089"], "sunsynk_token_expires_at": real_datetime(2026, 9, 19, 22, 11, 54)}}) + if not isinstance(load_apps_yaml_credentials(typed).get("token_expires_at"), str): + print("ERROR: a YAML-typed expiry must be handed on as a string") + failed = True + + # A bare mapping (a hand-trimmed file with no pred_bat wrapper) is accepted too. + bare = write(folder, "bare.yaml", {"sunsynk_auth_method": "oauth", "sunsynk_key": "token-abc", "sunsynk_inverter_sn": "2211093089"}) + if load_apps_yaml_credentials(bare).get("inverter_sn") != ["2211093089"]: + print("ERROR: a bare mapping with a scalar serial should load as a one-element list") + failed = True + + # Each way of being unusable must refuse with its own reason, never half-build. + refusals = { + "no_serial.yaml": {"pred_bat": {"sunsynk_auth_method": "oauth", "sunsynk_key": "token-abc"}}, + "oauth_no_token.yaml": {"pred_bat": {"sunsynk_auth_method": "oauth", "sunsynk_inverter_sn": ["2211093089"]}}, + "password_incomplete.yaml": {"pred_bat": {"sunsynk_username": "someone@example.com", "sunsynk_inverter_sn": ["2211093089"]}}, + } + for name, payload in refusals.items(): + try: + load_apps_yaml_credentials(write(folder, name, payload)) + print(f"ERROR: {name} should have been refused") + failed = True + except ValueError: + pass + for name, path in (("missing", os.path.join(folder, "missing.yaml")), ("not a mapping", write(folder, "list.yaml", ["a", "b"]))): + try: + load_apps_yaml_credentials(path) + print(f"ERROR: {name} should have been refused") + failed = True + except ValueError: + pass + + # A password account is complete without any token. + password = write(folder, "password.yaml", {"pred_bat": {"sunsynk_username": "someone@example.com", "sunsynk_password": "hunter2", "sunsynk_inverter_sn": ["2211093089"]}}) + if load_apps_yaml_credentials(password).get("username") != "someone@example.com": + print("ERROR: a username/password account should load without a token") + failed = True + assert not failed, "test_apps_yaml_credentials_loader" + + +def test_tou_test_window_avoids_now(): + """The CLI's live TOU window never contains the current time. + + The programme --tou-test writes is meant to be stored, read back and removed without + the inverter ever acting on it. A window covering the current minute would put the + battery into a real grid-charge for the seconds before the restore. + """ + failed = False + for now_minutes in range(0, 24 * 60, 7): + start, end = _tou_test_window(now_minutes) + start_minutes = int(start[:2]) * 60 + int(start[3:]) + end_minutes = int(end[:2]) * 60 + int(end[3:]) + if start_minutes < end_minutes: + inside = start_minutes <= now_minutes < end_minutes + else: + inside = now_minutes >= start_minutes or now_minutes < end_minutes + if inside: + print(f"ERROR: window {start}-{end} contains the current time {now_minutes // 60:02d}:{now_minutes % 60:02d}") + failed = True + if start == end: + print(f"ERROR: window {start}-{end} is zero length and would be dropped") + failed = True + assert not failed, "test_tou_test_window_avoids_now" + + def run_sunsynk_config_tests(my_predbat): """Run all Sunsynk configuration tests.""" failed = False for name, fn in [ ("inverter_def", test_inverter_def_registered), + ("apps_yaml_credentials", test_apps_yaml_credentials_loader), + ("tou_test_window", test_tou_test_window_avoids_now), ("component_registered", test_component_registered), ("apps_schema", test_apps_schema_keys), ("automatic_config", test_automatic_config_maps_control_entities), diff --git a/apps/predbat/tests/test_sunsynk_const.py b/apps/predbat/tests/test_sunsynk_const.py index 71383330b..64d8d0d50 100644 --- a/apps/predbat/tests/test_sunsynk_const.py +++ b/apps/predbat/tests/test_sunsynk_const.py @@ -18,7 +18,6 @@ SUNSYNK_DAY_FIELDS, TOU_FIELD, TOU_SLOT_COUNT, - TOU_FILLER_TIMES, FREEZE_EXPORT_SOC, encode_setting, ) @@ -93,26 +92,6 @@ def test_tou_field_templates(): assert not failed, "test_tou_field_templates" -def test_tou_filler_times_sufficient(): - """There are enough distinct filler times to pad any schedule out to 6 slots.""" - failed = False - if len(set(TOU_FILLER_TIMES)) != len(TOU_FILLER_TIMES): - print(f"ERROR: filler times not distinct: {TOU_FILLER_TIMES}") - failed = True - # A schedule contributes at most 4 boundary times (charge start/end, export start/end), - # so padding needs TOU_SLOT_COUNT distinct fillers to survive every one colliding. - if len(TOU_FILLER_TIMES) <= TOU_SLOT_COUNT: - print(f"ERROR: need more than {TOU_SLOT_COUNT} filler times, got {len(TOU_FILLER_TIMES)}") - failed = True - if TOU_FILLER_TIMES != sorted(TOU_FILLER_TIMES): - print(f"ERROR: filler times not ascending: {TOU_FILLER_TIMES}") - failed = True - if TOU_FILLER_TIMES[0] != "00:00": - print(f"ERROR: first filler time must be 00:00, got {TOU_FILLER_TIMES[0]}") - failed = True - assert not failed, "test_tou_filler_times_sufficient" - - def test_telemetry_maps_cover_predbat_args(): """Every sensor Predbat needs is sourced from a declared endpoint and field.""" failed = False @@ -192,7 +171,6 @@ def run_sunsynk_const_tests(my_predbat): ("endpoints", test_sunsynk_endpoints), ("workmode_semantics", test_sunsynk_workmode_semantics), ("tou_field_templates", test_tou_field_templates), - ("tou_filler_times", test_tou_filler_times_sufficient), ("telemetry_maps", test_telemetry_maps_cover_predbat_args), ("encode_setting", test_encode_setting_types), ("freeze_export_soc", test_freeze_export_soc), diff --git a/apps/predbat/tests/test_tou_schedule.py b/apps/predbat/tests/test_tou_schedule.py new file mode 100644 index 000000000..571850b7f --- /dev/null +++ b/apps/predbat/tests/test_tou_schedule.py @@ -0,0 +1,284 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# Test the shared Time-of-Use slot programme builder +# ----------------------------------------------------------------------------- + +"""Tests for TouScheduleMixin, the TOU programme builder shared by DEYE and Sunsynk. + +These exercise the mixin directly against a minimal fake component, so they hold whatever +either vendor does with the slots afterwards. The vendor-level equivalents live in +test_deye_control.py and test_sunsynk_control.py. +""" + +from tou_schedule import TouScheduleMixin, TOU_PADDING_STEP, MINUTES_PER_DAY + +SLOT_COUNT = 6 + + +class FakeTouComponent(TouScheduleMixin): + """Smallest component the mixin can drive: states are plain labels, slots plain dicts.""" + + TOU_SLOTS = SLOT_COUNT + + def derive_control_state(self, intent, current_soc): + """Label the intent as charge, export or idle, carrying the target SoC.""" + if intent.get("charge", {}).get("enable"): + return {"action": "charge", "slot_soc": intent["charge"].get("soc", 0), "grid_charge": True, "power": 3000} + if intent.get("export", {}).get("enable"): + return {"action": "export", "slot_soc": intent["export"].get("soc", 0), "solar_sell": True, "power": 3000} + return {"action": "idle", "slot_soc": intent.get("reserve", 0), "grid_charge": False, "power": 0} + + def _slot_for(self, start_time, state, reserve, self_use_power): + """Encode a slot as the time plus the state's label and SoC.""" + return {"time": start_time, "action": state["action"], "soc": state["slot_soc"]} + + +def _schedule(charge=None, export=None, reserve=10): + """Build a schedule dict in the shape build_tou_slots expects.""" + sched = {"reserve": reserve, "charge": {"enable": False, "soc": 0, "power": 3000}, "export": {"enable": False, "soc": 0, "power": 3000}} + if charge: + sched["charge"] = {"enable": True, "soc": 95, "power": 3000, "start": charge[0], "end": charge[1]} + if export: + sched["export"] = {"enable": True, "soc": 20, "power": 3000, "start": export[0], "end": export[1]} + return sched + + +def _build(charge=None, export=None, reserve=10): + """Run the builder over a schedule and return its slots.""" + return FakeTouComponent().build_tou_slots(_schedule(charge, export, reserve), current_soc=50, self_use_power=5000) + + +def _hm(text): + """Convert an HH:MM slot time to minutes past midnight.""" + hours, minutes = str(text).split(":") + return int(hours) * 60 + int(minutes) + + +def _action_at(slots, minute): + """Return the action in force at a minute past midnight. + + A slot runs from its own start until the next slot starts, and the programme loops at + midnight, so before the first slot the last one is still running. + """ + action = slots[-1]["action"] + for slot in slots: + if _hm(slot["time"]) <= minute: + action = slot["action"] + else: + break + return action + + +def _covered(slots, start, end, action): + """Return the minutes between start and end NOT showing the expected action.""" + first, last = _hm(start), _hm(end) + minutes = range(first, last) if first < last else list(range(first, MINUTES_PER_DAY)) + list(range(0, last)) + return [minute for minute in minutes if _action_at(slots, minute) != action] + + +def test_programme_shape(): + """Every programme is exactly TOU_SLOTS slots, distinct, ascending, starting at 00:00.""" + failed = False + cases = { + "idle": _build(), + "one charge window": _build(charge=("02:00", "05:00")), + "charge and export": _build(charge=("02:00", "05:00"), export=("16:00", "19:00")), + "charge wrapping midnight": _build(charge=("22:00", "02:00")), + "wrapping charge and export": _build(charge=("23:30", "05:30"), export=("16:00", "19:00")), + "window ending at midnight": _build(charge=("22:00", "00:00")), + "window starting at midnight": _build(charge=("00:00", "06:00")), + } + for label, slots in cases.items(): + times = [slot["time"] for slot in slots] + if len(slots) != SLOT_COUNT: + print(f"ERROR: {label} produced {len(slots)} slots, expected {SLOT_COUNT}: {times}") + failed = True + if len(set(times)) != len(times): + print(f"ERROR: {label} has duplicate slot times: {times}") + failed = True + if times != sorted(times): + print(f"ERROR: {label} slot times are not ascending: {times}") + failed = True + if times and times[0] != "00:00": + print(f"ERROR: {label} must start at 00:00, got {times[0]}") + failed = True + assert not failed, "test_programme_shape" + + +def test_padding_never_interrupts_a_window(): + """A window runs unbroken from its start to its end, whatever padding is added. + + GH#5156: padding used to be placed at fixed clock hours, so any window crossing one of + them was silently ended there while Predbat's plan still showed it running. + """ + failed = False + cases = [ + ("reported day 2", ("09:45", "15:30"), None, "charge"), + ("reported day 1", ("11:15", "15:30"), None, "charge"), + ("overnight crossing 04:00", ("00:30", "07:30"), None, "charge"), + ("export crossing 12:00", None, ("11:00", "13:00"), "export"), + ("charge crossing 08:00 with an export", ("06:00", "10:00"), ("16:00", "19:00"), "charge"), + ] + for label, charge, export, action in cases: + window = charge or export + slots = _build(charge=charge, export=export) + missed = _covered(slots, window[0], window[1], action) + if missed: + first = missed[0] + print(f"ERROR: {label} {window[0]}-{window[1]} loses {len(missed)} minutes from {first // 60:02d}:{first % 60:02d}: {[s['time'] for s in slots]}") + failed = True + assert not failed, "test_padding_never_interrupts_a_window" + + +def test_window_spanning_midnight_is_split(): + """A wrapping window becomes a pre-midnight and a post-midnight slot, charging throughout. + + No slot may span midnight, so a 22:00-02:00 charge has to be written as 22:00-00:00 plus + 00:00-02:00. Without the split the post-midnight half was silently dropped, because the + 00:00 slot was always seeded idle. + """ + failed = False + for label, charge in (("22:00-02:00", ("22:00", "02:00")), ("23:30-05:30", ("23:30", "05:30"))): + slots = _build(charge=charge) + missed = _covered(slots, charge[0], charge[1], "charge") + if missed: + first = missed[0] + print(f"ERROR: wrapping charge {label} loses {len(missed)} minutes from {first // 60:02d}:{first % 60:02d}: {[(s['time'], s['action']) for s in slots]}") + failed = True + if slots[0]["action"] != "charge": + print(f"ERROR: wrapping charge {label} must leave slot 1 at 00:00 charging, got {slots[0]}") + failed = True + # And it must stop at the window's end rather than running all day. + if _action_at(slots, _hm(charge[1]) + 1) != "idle": + print(f"ERROR: wrapping charge {label} did not return to idle after {charge[1]}") + failed = True + assert not failed, "test_window_spanning_midnight_is_split" + + +def test_split_at_midnight_halves(): + """_split_at_midnight returns halves that never span midnight.""" + failed = False + expected = { + ("02:00", "05:00"): [("02:00", "05:00")], + ("22:00", "02:00"): [("00:00", "02:00"), ("22:00", "00:00")], + ("22:00", "00:00"): [("22:00", "00:00")], + ("00:00", "06:00"): [("00:00", "06:00")], + } + for (start, end), want in expected.items(): + got = TouScheduleMixin._split_at_midnight(start, end) + if got != want: + print(f"ERROR: split {start}-{end} expected {want} got {got}") + failed = True + for half_start, half_end in got: + if half_end != "00:00" and half_start >= half_end: + print(f"ERROR: split {start}-{end} produced a non-forward half {half_start}-{half_end}") + failed = True + assert not failed, "test_split_at_midnight_halves" + + +def test_padding_sits_after_the_last_slot(): + """Padding follows the last real slot, carrying its state, at TOU_PADDING_STEP intervals.""" + failed = False + slots = _build(charge=("02:00", "05:00")) + times = [slot["time"] for slot in slots] + # 00:00 idle, 02:00 charge, 05:00 idle, then three padding slots after 05:00. + if times[:3] != ["00:00", "02:00", "05:00"]: + print(f"ERROR: expected the schedule's own boundaries first, got {times}") + failed = True + expected_padding = [f"05:{TOU_PADDING_STEP * step:02d}" for step in range(1, 4)] + if times[3:] != expected_padding: + print(f"ERROR: expected padding at {expected_padding}, got {times[3:]}") + failed = True + if any(slot["action"] != "idle" for slot in slots[3:]): + print(f"ERROR: padding must carry the state it follows (idle), got {slots[3:]}") + failed = True + assert not failed, "test_padding_sits_after_the_last_slot" + + +def test_padding_falls_back_when_the_day_runs_out(): + """Padding that will not fit before midnight moves into the widest gap instead. + + A window ending at 23:50 leaves only 10 minutes before the programme loops, which is not + enough for the slots still to be placed. They go into the largest gap instead, where they + carry that gap's own state and so still change nothing. + """ + failed = False + slots = _build(charge=("23:40", "23:50")) + times = [slot["time"] for slot in slots] + if len(times) != SLOT_COUNT or len(set(times)) != SLOT_COUNT: + print(f"ERROR: expected {SLOT_COUNT} distinct times, got {times}") + failed = True + if times != sorted(times): + print(f"ERROR: padding fallback broke the ascending order: {times}") + failed = True + if any(_hm(time) >= MINUTES_PER_DAY for time in times): + print(f"ERROR: a padding slot ran past midnight: {times}") + failed = True + # The window itself must survive the fallback. + missed = _covered(slots, "23:40", "23:50", "charge") + if missed: + print(f"ERROR: the fallback padding truncated the window: {[(s['time'], s['action']) for s in slots]}") + failed = True + assert not failed, "test_padding_falls_back_when_the_day_runs_out" + + +def test_idle_schedule_holds_one_state_all_day(): + """With no windows every slot is idle, so the padding cannot introduce a change. + + The degenerate case: one state for 24 hours still has to fill TOU_SLOTS slots, and they + all say the same thing rather than needing the day split artificially. + """ + failed = False + slots = _build() + if any(slot["action"] != "idle" for slot in slots): + print(f"ERROR: an idle schedule must be idle in every slot, got {slots}") + failed = True + if _covered(slots, "00:00", "23:59", "idle"): + print(f"ERROR: an idle schedule must read idle at every minute, got {slots}") + failed = True + assert not failed, "test_idle_schedule_holds_one_state_all_day" + + +def test_zero_length_window_is_ignored(): + """A window whose start equals its end has no interval and must not reach the slots.""" + failed = False + slots = _build(charge=("03:00", "03:00")) + if any(slot["action"] != "idle" for slot in slots): + print(f"ERROR: a zero-length window must produce no action slot, got {slots}") + failed = True + # The same via the HH:MM:SS form the control entities actually carry. + sched = _schedule() + sched["charge"] = {"enable": True, "soc": 95, "power": 3000, "start": "03:00:00", "end": "03:00"} + if any(slot["action"] != "idle" for slot in FakeTouComponent().build_tou_slots(sched, current_soc=50, self_use_power=5000)): + print("ERROR: a zero-length window must be caught on the normalised times too") + failed = True + assert not failed, "test_zero_length_window_is_ignored" + + +def run_tou_schedule_tests(my_predbat): + """Run all shared TOU programme builder tests.""" + failed = False + for name, fn in [ + ("programme_shape", test_programme_shape), + ("padding_no_interruption", test_padding_never_interrupts_a_window), + ("midnight_split", test_window_spanning_midnight_is_split), + ("split_halves", test_split_at_midnight_halves), + ("padding_after_last", test_padding_sits_after_the_last_slot), + ("padding_fallback", test_padding_falls_back_when_the_day_runs_out), + ("idle_all_day", test_idle_schedule_holds_one_state_all_day), + ("zero_length_window", test_zero_length_window_is_ignored), + ]: + try: + if fn(): + print(f" FAILED: tou_schedule.{name}") + failed = True + except Exception as e: + print(f" EXCEPTION in tou_schedule.{name}: {e}") + import traceback + + traceback.print_exc() + failed = True + return failed diff --git a/apps/predbat/tou_schedule.py b/apps/predbat/tou_schedule.py new file mode 100644 index 000000000..b053d3962 --- /dev/null +++ b/apps/predbat/tou_schedule.py @@ -0,0 +1,215 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# Time-of-Use slot programme builder, shared by the DEYE-family components +# ----------------------------------------------------------------------------- + +"""Build a fixed-length Time-of-Use slot programme from Predbat's schedule. + +DEYE and Sunsynk are the same inverter firmware behind two different clouds, so both +expose the same TOU model: a fixed number of slots (TOU_SLOTS, 6 on both) that +between them tile the whole 24 hours. Only the wire encoding differs - DEYE sends a +JSON list of slot objects, Sunsynk sends positional numbered fields (sellTime1..6) - +so the ENCODING stays in each component while the programme logic lives here. + +The rules this builder implements, which are properties of the firmware rather than +Predbat preferences. The first two are CONFIRMED by Sunsynk's own documentation +("Avoiding conflicts in the System Mode timer"), which describes the same registers: + +* A slot carries only a START time and runs until the next slot starts, so the slots + must be distinct and chronological - "Timers MUST be set chronologically from Timer + 1 to Timer 6" - and slot 1 must start at 00:00. There is no "off" slot: whatever a + slot says is in force for its whole interval, whatever its grid-charge flag says. + The documented factory default runs all six timers as ranges with Grid Charge ticked + on only two of them. +* No slot may span midnight: Timer 6 is the only one permitted to roll over, and it + does so by running until Timer 1 restarts. A window that wraps therefore has to be + SPLIT into a pre-midnight slot and a post-midnight one - see _split_at_midnight. +* Predbat describes at most one charge window and one export window, so the schedule + needs at most 5 slots (the 00:00 seed, plus two boundaries per window; a wrapping + window still costs 2 because its post-midnight half reuses the 00:00 seed). There + is therefore always at least one slot left over to pad with. + +Note an inverter can nonetheless be found sitting with all six slots at 00:00 (an +unconfigured default) and the API stores that happily. It is simply not a chronological +programme, so no conclusion about what a VALID schedule may look like follows from it. + +A component using this mixin must provide: + +* ``TOU_SLOTS`` - how many slots the programme holds. +* ``derive_control_state(intent, current_soc)`` - turn an intent into a control state. +* ``_slot_for(start_time, state, reserve, self_use_power)`` - encode one slot. +""" + +# Minutes between consecutive padding slots. Padding slots are REAL intervals rather +# than zero-length ones: two slots sharing a start time is not a chronological +# programme, and the firmware's response to it is not consistent enough to rely on +# (an inverter can be found sitting with all six slots at 00:00, but that is an +# unconfigured default, not a schedule it was asked to run). Five minutes is below +# Predbat's own planning resolution, so a padding slot can never split a window it +# was not already inside. +TOU_PADDING_STEP = 5 + +# Minutes in a day. Slot times are wall-clock within a single day - the programme +# loops at midnight rather than running past it. +MINUTES_PER_DAY = 24 * 60 + + +class TouScheduleMixin: + """Shared Time-of-Use programme construction for the DEYE-family components.""" + + @staticmethod + def _to_slot_time(value): + """Normalise a schedule time to the HH:MM the TOU slots require. + + The control entities carry HH:MM:SS because that is what Predbat writes (see + INVERTER_DEF charge_time_format), so the seconds are dropped here - at the one + point a schedule time becomes a slot time. + """ + parts = str(value or "00:00").split(":") + if len(parts) < 2: + return "00:00" + try: + return f"{int(parts[0]):02d}:{int(parts[1]):02d}" + except ValueError: + return "00:00" + + @staticmethod + def _hm_to_minutes(hm): + """Convert an HH:MM string to minutes since midnight (0 on bad input).""" + try: + parts = str(hm).split(":") + return int(parts[0]) * 60 + int(parts[1]) + except (ValueError, IndexError): + return 0 + + @staticmethod + def _minutes_to_hm(minutes): + """Convert minutes since midnight to an HH:MM slot time.""" + minutes = int(minutes) % MINUTES_PER_DAY + return f"{minutes // 60:02d}:{minutes % 60:02d}" + + def _window_active(self, window, now_minutes): + """Return True if an enabled window covers now_minutes, handling a midnight wrap.""" + if not window.get("enable") or not window.get("start") or not window.get("end"): + return False + start = self._hm_to_minutes(self._to_slot_time(window["start"])) + end = self._hm_to_minutes(self._to_slot_time(window["end"])) + if start == end: + return False + if start < end: + return start <= now_minutes < end + return now_minutes >= start or now_minutes < end # window wraps past midnight + + def _active_state(self, schedule, current_soc, now_minutes): + """Derive the control state for the window active at now_minutes, else idle. + + These inverters have a single global work mode per schedule, so the top-level + mode must follow the window active RIGHT NOW rather than a static export-first + precedence: otherwise an export window enabled elsewhere in the day would pin + the mode to selling-first and block the charge window's grid charging. + """ + reserve = int(schedule.get("reserve", 0)) + charge = schedule.get("charge", {}) + export = schedule.get("export", {}) + # The rates are carried even when no window is active: with both windows shut a zero + # charge rate is Predbat's Freeze Export, and the mode has to follow it rather than + # sit in demand (see derive_control_state). + intent = {"reserve": reserve, "charge": {"enable": False, "power": int(charge.get("power", 0))}, "export": {"enable": False, "power": int(export.get("power", 0))}} + if self._window_active(export, now_minutes): + intent["export"] = {"enable": True, "soc": export.get("soc", 0), "power": export.get("power", 0)} + elif self._window_active(charge, now_minutes): + intent["charge"] = {"enable": True, "soc": charge.get("soc", 0), "power": charge.get("power", 0)} + return self.derive_control_state(intent, current_soc) + + @staticmethod + def _split_at_midnight(start_time, end_time): + """Split a window into (start, end) halves that never span midnight. + + A slot runs from its start until the next slot begins, and the programme loops at + midnight rather than running through it, so a window from 22:00 to 02:00 cannot be + one slot. It becomes two: 00:00-02:00 and 22:00-00:00. Both carry the window's + action state, and because slot 1 at 00:00 is one of them the charge runs unbroken + across midnight. + + A window ending exactly at midnight has no post-midnight half - it already stops + where the programme loops. + """ + if end_time == "00:00": + return [(start_time, "00:00")] + if start_time < end_time: + return [(start_time, end_time)] + return [("00:00", end_time), (start_time, "00:00")] + + def _padding_segments(self, ordered, needed, baseline): + """Place `needed` extra slots where they cannot disturb the programme. + + Every one of the inverter's slots is always in force, so a slot Predbat has no use + for still has to say something. It says the SAME THING as the slot it follows, and + sits just after it, which makes it a real but inert interval: the state does not + change at its boundary, so it cannot end a window early (GH#5156, where padding at + fixed clock hours truncated any window crossing 04:00/08:00/12:00). + + They go after the LAST slot by preference, keeping them out of the way at the end + of the day. When the last slot is too close to midnight to fit them - a window + ending at 23:50, say - they go into the largest gap instead, which is always wide + enough: with at most 5 slots in the programme the widest gap is at least + 1440/5 minutes, against the 25 minutes five padding slots need. + """ + if needed <= 0: + return [] + if not ordered: + return [(self._minutes_to_hm(index * TOU_PADDING_STEP), dict(baseline)) for index in range(needed)] + span = TOU_PADDING_STEP * needed + starts = [self._hm_to_minutes(start_time) for start_time, _ in ordered] + # Candidate anchors are every slot, each paired with the room before the next one + # (the last slot's room running to midnight). Prefer the last, then the widest. + room = [(starts[index + 1] if index + 1 < len(starts) else MINUTES_PER_DAY) - starts[index] for index in range(len(starts))] + anchor = len(starts) - 1 + if room[anchor] <= span: + anchor = room.index(max(room)) + state = ordered[anchor][1] + return [(self._minutes_to_hm(starts[anchor] + step * TOU_PADDING_STEP), dict(state)) for step in range(1, needed + 1)] + + def build_tou_slots(self, schedule, current_soc, self_use_power): + """Build exactly TOU_SLOTS ordered slots covering 24h from the schedule windows.""" + reserve = int(schedule.get("reserve", 0)) + # Collect (start_time, state) segment boundaries, from a DERIVED baseline at 00:00. + # "No window is active" is not always demand - a freeze export is exactly that state + # plus a zero charge rate (see derive_control_state) - so every slot the schedule + # does not otherwise claim carries the baseline, and a freeze covers the whole + # programme. Predbat never says when a freeze ends, so there is no boundary to write. + baseline = self.derive_control_state({"reserve": reserve, "charge": {"enable": False, "power": int(schedule.get("charge", {}).get("power", 0))}, "export": {"enable": False, "power": int(schedule.get("export", {}).get("power", 0))}}, current_soc) + segments = {"00:00": dict(baseline)} + for direction in ("charge", "export"): + window = schedule.get(direction, {}) + if not (window.get("enable") and window.get("start") and window.get("end")): + continue + # Normalised to HH:MM here: these strings become slot times, and the entities + # they came from carry seconds. + start_time = self._to_slot_time(window["start"]) + end_time = self._to_slot_time(window["end"]) + if start_time == end_time: + # Mirrors the guard in _window_active: a zero-length window has no interval + # to act over. Compared on the NORMALISED times, so "02:00:00" against + # "02:00" is caught too. Without this, an enable event arriving before the + # time fields (both still the "00:00:00" default) would add an action + # segment whose matching return-to-baseline segment cannot be added at the + # same key - an unterminated, multi-hour full-power grid-charge/export + # slot, even though _active_state correctly reports the window inactive. + continue + intent = {"reserve": reserve, "charge": {"enable": False}, "export": {"enable": False}} + intent[direction] = {"enable": True, "soc": window.get("soc", 0), "power": window.get("power", 0)} + action = self.derive_control_state(intent, current_soc) + for segment_start, segment_end in self._split_at_midnight(start_time, end_time): + segments[segment_start] = dict(action) + # After the window, return to the baseline. setdefault so a window ending + # where another starts does not overwrite that window's action, and so the + # midnight end of a wrapping window leaves its own post-midnight half alone. + segments.setdefault(segment_end, dict(baseline)) + ordered = sorted(segments.items(), key=lambda kv: kv[0]) + ordered = ordered + self._padding_segments(ordered, self.TOU_SLOTS - len(ordered), baseline) + ordered = sorted(ordered, key=lambda kv: kv[0])[: self.TOU_SLOTS] + return [self._slot_for(start_time, state, reserve, self_use_power) for start_time, state in ordered] diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index 6fe941c6b..ab9f024fc 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -205,6 +205,7 @@ from tests.test_deye_api import run_deye_api_tests from tests.test_deye_oauth import run_deye_oauth_tests from tests.test_deye_control import run_deye_control_tests +from tests.test_tou_schedule import run_tou_schedule_tests from tests.test_deye_publish import run_deye_publish_tests from tests.test_deye_storage import run_deye_storage_tests from tests.test_sunsynk_const import run_sunsynk_const_tests @@ -581,6 +582,7 @@ def main(): ("saving_session_zero_octopoints_free_slot", test_saving_session_zero_octopoints_joined_is_free_slot, "Joined zero octopoints session becomes a free import slot test (issue #4851)", False), ("alert_feed", test_alert_feed, "Alert feed tests", False), ("fox_api", run_fox_api_tests, "Fox API tests", False), + ("tou_schedule", run_tou_schedule_tests, "Shared TOU slot programme tests", False), ("deye_const", run_deye_const_tests, "DEYE constants tests", False), ("deye_config", run_deye_config_tests, "DEYE config/INVERTER_DEF tests", False), ("deye_api", run_deye_api_tests, "DEYE API tests", False), diff --git a/tools/debug-journal.md b/tools/debug-journal.md index f24c7b2f0..108036762 100644 --- a/tools/debug-journal.md +++ b/tools/debug-journal.md @@ -101,7 +101,7 @@ Grep for the named symbol rather than trusting a line number. | HA write/verify (`ha.py`, `inverter.py`) | `get_state(refresh=True)` is a no-op in a normal HA add-on install: it only re-reads when `not self.ha_key`, and `ha_key` is set in that case, so verification reads the websocket cache, never a fresh value (`ha.py:798`, since PR #2342). Separately, `write_and_poll_value()` treats `domain == "sensor"` as Predbat-owned and POSTs `/api/states/` directly (`inverter.py:2155`, `ha.py:1071-1077`) instead of calling the integration. Probed empirically across all four write-path/domain combinations (GH#4738): a control entity that resolves to a `sensor.*` domain by mistake gets a direct state overwrite that reads back as success, with zero corresponding lines in the real integration's log. `call_service_wrapper()`'s return value used to be discarded at every `inverter.py` write site, so a rejected HA service call (e.g. a Modbus "Illegal Function" error) never reached the verify logic. That is no longer true everywhere: PR #4878 made `call_service_template()` check it (`inverter.py:2817`) and log "was not accepted, it will be retried next cycle". The other eleven `call_service_wrapper()` call sites in that file still discard it, so check the specific write path before repeating the general claim. Later triage extended this row rather than replacing it. `write_and_poll_switch()` builds its service name as `domain + "/turn_..."` straight from the entity's own domain with no validation (`inverter.py:2102`), and `adjust_inverter_mode()` routes every `has_ge_eco_toggle` inverter through it - so an `inverter_mode` pointed at a `select.*` entity produces `select.turn_on`, an HA-invalid service retried 10x at 10s per attempt, and the string coercion just below (`inverter.py:2110-2111`) maps any select state to False, so the force-export direction reports success having written nothing (GH#4909, still live — though on 3-phase GEC the real problem is one stage deeper: no eco register exists at all, and the write path is skipped with `No entity_id for ECO Toggle`; see the GE Cloud row). `adjust_force_export`/`adjust_charge_window` format their times at plan precision with no snapping (`TIME_FORMAT_HMS`, `inverter.py:2608`/`:2615`), so against an integration whose select only offers 15-minute options (Solar Assistant/Deye) Predbat writes e.g. `20:55` and `write_and_poll_option` retries the impossible value 10x every cycle; nothing anywhere reads a select's `options` attribute, and `charge_time_entity_is_option` only decides *how* an entity is written, never which values are legal. The fix precedent is already in tree - AlphaESS's `snap_time_grid()` (`alphaess_const.py:215`), snap start forward and end backward with a collapse check (GH#4947). A second, independent way plan-precision times fail on the same select path: `write_and_poll_option()` only strips the seconds from an 8-char `HH:MM:SS` write when the *previous* state is a healthy 5-char `HH:MM:SS` read — the read feeding it returns `None` for an entity missing from the state cache (`ha.py:817`) or the literal `"unknown"`/`"unavailable"` when degraded (`ha.py:815`), either skips the strip and the raw `HH:MM:SS` goes to `select_option`, which HA rejects; all 10 retries re-read and fail identically, and the log reads `didn't complete got unknown` (GH#5009, probe-verified on main). The fix that serves both is reading the select's `options` attribute — nothing in the codebase does today. Six of the eight `adjust_*` writers discard the write helper's boolean and fire `call_notify()`/`mqtt_message()` regardless, and `rest_setReserve()` has the same gap (GH#4845). `call_service_template()` used to record its dedup hash *before* the call, so a silently failed `select_option` was skipped as "previously called" on every later cycle; fixed (it now records only on success and drops the record on failure), though it still deliberately returns True so callers cannot downgrade a freeze into a stop (GH#4876). From that same report, a false-alarm class worth recognising: a `didn't complete got X` where X is exactly the *previous* cycle's target, on an integration that republishes its entities slowly, is the ~50s verify window reading HA's cache - the write had already landed by the next cycle. | `inverter` | | GE Cloud (`gecloud.py`) | Gateway fields can come back null. `merge_non_null` stops nulls overwriting good values, and the publish path guards null containers — GH#4656 was a null crash in that area. `number_event` clamps an out-of-range write silently and does not report the clamped value back, while `write_and_poll_value()` keeps comparing against the original pre-clamp target - a permanent false `had_errors` and an unbounded retry (GH#4826). PR #4828 clamped inside `adjust_reserve()` only; the shared write helper still has no min/max awareness and the other `inverter.py` write sites still pass unclamped targets (GH#4831). The error-code path has since been reworked (`9560c61a`, `d6e5998c`) after GH#4896 showed `success: false` nulled `data` before the code branch could read it, so codes GivEnergy documents as never succeeding on a repeat (-3/-4/-7) were retried the full budget and `message` was never surfaced. Separately, **fixed in PR #4954 (merged 2026-09-06)** - keep the mechanism for anyone holding an older log: for `GE`/`GEC`/`GEE` the max battery rate was taken solely from the `charge_rate` entity's `max` attribute, and the `elif "battery_rate_max"` branch below it was unreachable for those types. Percentage-rated models (the 3-phase units) have no absolute charge power register, so `ge_cloud_automatic` left `charge_rate` unset and the attribute read returned the 2600 W default; both the correct rate GECloud writes into `battery_rate_max` and the user's own apps.yaml value went unread, and every rate was clamped to 2600 W. On those models it also corrupted the writes, since the percentage registers are set as `new_rate / battery_rate_max_raw * 100` against a denominator 3.8x too small. The fix falls back to `battery_rate_max` only when `charge_rate` resolves to nothing (GH#4908). GH#4198 is the same code shape on SolaX and is not covered by that fix. One API-shape trap on the site endpoint (`GE_API_SITE`, `site/{uuid}`, added for PR #4978): the published OpenAPI spec types `limits.import`/`limits.export` as a string with a null example and never shows a populated one, so the shape had to be learned from live accounts. A populated limit is `{'enabled': bool, 'power': {'watts': int, 'amps': float}}`, and **`enabled` is the thing that matters** - a site describes its connection's declared capacity in exactly the same structure as a curtailment, marked disabled, so `power.watts` alone tells you nothing about whether anything is being restricted. Site 64744 gave `{'import': {'enabled': False, 'power': {'watts': 46000, 'amps': 200}}, 'export': {'enabled': True, 'power': {'watts': 4500, 'amps': 19.6}}}` - a 200A supply whose export really is curtailed to 4.5kW - while site 46714's disabled 6kW export is its declared capacity and no restriction at all. A fleet survey confirmed the flag across accounts, so only an enabled limit is applied; an enabled zero is a real zero-export connection. A site with no limit reports a null import/export. For a fleet survey use `GET /v1/site`, the list endpoint, which returns every site's `limits` and the inverters under it (with `info.max_discharge_rate`) in one response - no second call per site. Two GEC triage tools and one clobber trap (GH#5017). The component publishes an entity for every `/settings` entry with no name filter (`async_get_inverter_settings()`/`publish_registers()`, `regname_to_ha()` is plain lowercasing), so **absence of an entity in the reporter's debug yaml is proof the API never listed the register** — "the app can control it" does not imply the settings API exposes it, because the app/GivTCP write via local control paths rather than the `/settings` list (a GIV-3HY-20 had slot 1/2 lower-SoC registers absent while 3..10 existed). Caveat: the settings list is fetched once per process start (`register_list` cache) and only the parsed flags are logged, so the debug yaml is the only observable. To capture the raw register list, `python3 apps/predbat/gecloud.py --api-key --write-entity number.probe_nothing 0` makes `test_gecloud_direct()` print the full "Available entities/registers" list with raw cloud names and setting IDs and performs no write — but the harness's startup pass also runs `enable_default_options()` against the real inverter (floors reset, unused windows zeroed), so say so when asking a user to run it. Clobber trap: the missing-feature branch `set_arg("discharge_target_soc", None)` *deletes* the key, including an explicit apps.yaml entry the user wrote; `set_arg_auto(..., overwrite=False)` (component_base.py) is the in-repo pattern for letting an explicit value win — check the `has_*` detection branch on any "automatic config deleted my setting" report. The `inverter_mode` half of this class was fixed in PR #5059 (`885c637d`) — it now binds via `set_arg_auto(..., overwrite=False)` like `export_limit` — but the other missing-feature deletes are still plain `set_arg(..., None)` on main: `pause_mode`, `pause_start_time`, `pause_end_time`, `discharge_target_soc`, `charge_rate_percent`, `discharge_rate_percent`, `givtcp_rest`. A companion test-fidelity trap (GH#5056, verified on main): the component test mocks (`MockBase.set_arg` in `test_ge_cloud.py`; same shape in `test_ohme.py` and `test_alphaess_api.py`, whose mock `set_arg_auto` also writes unconditionally) store the value instead of mirroring the real `set_arg()`'s None-deletes, and their `args_from_apps_yaml` is empty so `set_arg_auto()` falls through to the plain path — so the "feature not detected" assertions (`config_args.get(...) is None`) pass identically whether the code stored None or deleted the key, and a green suite cannot distinguish fixed from unfixed. Any test for this class must populate `args_from_apps_yaml` and assert the manual value survives — #5059's Test 10 in `test_ge_cloud.py` is now the in-repo precedent. And the deeper half of GH#4909 (still live): the 3-phase GEC class exposes **no** eco/demand/mode register in `inverter/{sn}/settings` — verified by enumerating the `predbat_gecloud_*` entities in the reporter's debug dump — yet the `GEC` def sets `has_ge_eco_toggle: True` unconditionally (`config.py`) and auto-config maps `inverter_mode` via `build_entities("switch", ["enable_eco_mode"])` (gecloud.py), which emits nothing when the register is missing, so `adjust_inverter_mode()` logs `No entity_id for ECO Toggle` and returns without writing every plan cycle and freeze export silently no-ops (force-discharge paths work fine). Fix wrinkle for the maintainer: the flag is one per-inverter-type bool while `build_entities` works per-device. The neighbouring **charge-side** gap (GH#5040: `scheduled_charge_enable` had no `enable_force_charge` candidate, so a 3-phase GEC that gates timed grid charge behind both charge switches logged real charge states while importing nothing) is **fixed in PR #5042** (`a7121135`, v9.0.2): auto-config now binds `scheduled_charge_enable` to `enable_force_charge` with `ac_charge_enable`/`enable_ac_charge` as fallbacks, and `enable_default_options()` holds `enable_ac_charge` on as a static enable when the force register exists. Keep the mechanism for pre-v9.0.2 logs, where the signature is a 3-phase GEC showing plan charge states against flat grid power with no warning anywhere. | `ge_cloud` | | Teslemetry / Powerwall (`teslemetry.py`) | Battery model is inferred from site `nameplate_power / battery_count`. A nameplate fallback combined with `inverter_hybrid: True` once produced a false 5 kW inverter limit and spurious morning export. Tariff writes push a whole TOU schedule every cycle, so a setting changed by hand in the Tesla app is reverted on the next cycle (GH#4600, GH#4610). Three later findings. `build_tariff()`/`_render_side` carve a *single* ON_PEAK interval priced with one scalar (`teslemetry.py:1117`/`:1036`/`:1074`), so no intra-window price gradient can exist and Tesla's own TBC is free to defer the whole export to the end of the window; the no-rate-data fallback branch also skips the boost entirely (GH#4887). The charge path sends no rate at all - `evaluate_schedule` (`teslemetry.py:539`) asserts mode `backup` plus reserve = charge target plus grid charging - so a slow charge ramp is Tesla firmware, not a Predbat write bug (GH#4892); Tesla now snaps a backup reserve of 81-99% down to 80% while Predbat still forwards any 0-100 target, and `automatic_config()` overwrites a manually set `battery_rate_max` whenever `teslemetry_automatic` is on. Since PR #4976 there is a Predbat-side lever for exactly that slow ramp: the `teslemetry_tbc_control` trial setting switches `evaluate_schedule()` to `evaluate_schedule_tbc()`, which pushes a **signal tariff** (`build_signal_tariff()` - fixed 0/50/100p bands over the committed charge and export windows) so Tesla's own optimiser runs the charge and reaches the full rate reserve-driven charging cannot; mode goes to `autonomous` in every state, reserve becomes an actual reserve rather than a charge signal, and `_settable_reserve()` maps requests onto reserves the Powerwall honours - an 81-99% request now rounds **up** to 100 rather than being forwarded raw into the snap-to-80 band. The same PR stopped `plan.py` planning a manual freeze export on inverters that cannot do it. A related pricing bug in the same `build_tariff()` family was fixed in PR #4977: an export window ending at midnight priced the whole of tomorrow at peak. And `optimization_strategy: "economics"` *is* pushed with `tariff_content_v2` on every `set_tariff()`, unconditionally, since v8.49.0 / PR #4603 - a request to "add" it is asking for something already shipped (GH#4918). | `teslemetry` | -| Sunsynk / DEYE (`sunsynk.py`, `deye.py`) | Freeze export is gated by the per-slot power register (`sellTime{n}Pac` on Sunsynk), confirmed on live hardware — setting the energy mode alone had no effect, and with slot power at zero the battery still charged. `read_only` was ignored by the reconcile loops until `_is_read_only()` gated `_reconcile_control()` (GH#4436). DEYE cloud telemetry keys are one hardcoded spelling per metric — `DEYE_TELEMETRY_KEYS["pv_power"] = "TotalSolarPower"` (`deye_const.py`), confirmed against a single live 3-phase hybrid — and a SUN-8K-SG05LP1-EU-AM2-P instead exposes `TotalDCInputPower` + `DCPowerPV1/2/3`, neither anywhere in the tree. Because `pv_power` is in `DEYE_TELEMETRY_REQUIRED`, the miss fires the `device/latest missing expected keys [...] - telemetry will read zero` warning every cycle and the Cloud PV sensor publishes 0.0 while a manual `pv_power` arg to a local sensor may still protect the planner. Reproduced by swapping the test fixture's single PV key (`tests/test_deye_api.py` `LIVE_DATA_LIST`). Whether the alternate key is unit-equivalent (vs summing `DCPowerPV1-3`) needs a live capture from an affected model — suspected, not verified. | `sunsynk_control`, `deye_control` | +| Sunsynk / DEYE (`sunsynk.py`, `deye.py`) | Freeze export is gated by the per-slot power register (`sellTime{n}Pac` on Sunsynk), confirmed on live hardware — setting the energy mode alone had no effect, and with slot power at zero the battery still charged. `read_only` was ignored by the reconcile loops until `_is_read_only()` gated `_reconcile_control()` (GH#4436). DEYE cloud telemetry keys are one hardcoded spelling per metric — `DEYE_TELEMETRY_KEYS["pv_power"] = "TotalSolarPower"` (`deye_const.py`), confirmed against a single live 3-phase hybrid — and a SUN-8K-SG05LP1-EU-AM2-P instead exposes `TotalDCInputPower` + `DCPowerPV1/2/3`, neither anywhere in the tree. Because `pv_power` is in `DEYE_TELEMETRY_REQUIRED`, the miss fires the `device/latest missing expected keys [...] - telemetry will read zero` warning every cycle and the Cloud PV sensor publishes 0.0 while a manual `pv_power` arg to a local sensor may still protect the planner. Reproduced by swapping the test fixture's single PV key (`tests/test_deye_api.py` `LIVE_DATA_LIST`). Whether the alternate key is unit-equivalent (vs summing `DCPowerPV1-3`) needs a live capture from an affected model — suspected, not verified. CONFIRMED live (Sunsynk 2211093089, 2026-09-19, via `sunsynk.py --tou-test`) that the settings object exposes the TOU register block a SECOND time under unrelated names, so several keys that read like independent settings are the same memory at a different scale: `volt1`-`volt4` are `sellTime3`-`sellTime6` as minutes-since-midnight / 50, `volt5`-`volt10` are `sellTime1Pac`-`sellTime6Pac` / 10, and `volt11`, `volt12`, `current1`-`current4` are `sellTime1Volt`-`sellTime6Volt` x 10 — a contiguous 16-value window, matched 16/16 in both the before and after reads of one write. Two consequences. It is independent proof a TOU write reached the real registers rather than a cloud-side echo, which is the cheapest hardware verification available. And it is a trap: anything that wrote `volt*`/`current*` would silently corrupt the charge programme. Predbat is safe today only because `SUNSYNK_SYSTEM_MODE_FIELDS` excludes them, so treat that exclusion as load-bearing rather than incidental. The alias does NOT extend to what the write payload must contain — `time{n}On` (capital O) is a derived echo of `time{n}on` and is correctly excluded via `SUNSYNK_DERIVED_SLOT_FIELDS`. | `sunsynk_control`, `deye_control` | | Grid sign / arrow direction | `grid_power_invert` is owned by some integrations and not others. With two systems configured, one integration setting it `True` bleeds into the other's entities and inverts the arrows. The fix is an explicit `False` in the automatic config of both. | `sunsynk_config`, `teslemetry` | | Enphase (`enphase.py`) | Unofficial Enlighten endpoints. Accounts with MFA cannot log in at all. Discharge-to-grid schedules are required for export control. Writes need a double-submit CSRF token or return 403. Using the Enphase app at the same time can trip session limits. | `enphase_api` | | myenergi (`myenergi.py`) | `myenergi_automatic_zappi` (PR #4997) gates the Zappi half of automatic config; with it off, Zappis stay monitor-only and `control_active` is never set. `release_zappis()` has exactly one caller — `control_tick()` — reached only under `if self.control_active:` in `run()`, and `control_active` is latched at the first tick: every `enable_control()` refusal (zappi_control → automatic → automatic_zappi → enable_controls) returns *before* setting it, and the gate re-runs only when a fresh instance starts. So after a component restart with any prerequisite off, a Zappi already held in 'Stopped' stays Stopped and the `switch.*_myenergi_zappi_control` entity is not even republished (it is gated on `control_active` too) — the symptom for a future report is "car won't charge after I turned off Predbat" with a Zappi. Note `control_tick()` itself *does* release on read-only mode and on the control switch being turned off; it is the config-gate half that strands. Two adjacent lifecycle facts: published entities are **never removed** — `dashboard_item()`/`set_state()` are upsert-only (`output.py`, `ha.py` POST `/api/states`), so when a gate stops `publish_data()` publishing a switch the old entity lingers frozen at its last state until HA restarts, and there is no entity-removal path anywhere in the codebase; and `set_arg`/`set_arg_auto` writes survive a component-only restart, because `Components.restart()` stops and re-creates only the instance and never resets the shared `base.args` from apps.yaml — auto-wired values (e.g. myenergi's `car_charging_*`) persist until a full process restart. Distinguish the two restarts when checking arg lifecycle. | `myenergi` |