diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index 76455a365..2bf32ac40 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -23,6 +23,7 @@ import math import json import random +import traceback from component_base import ComponentBase from coordinator import inverter_record from mock_base import MockBase as SharedMockBase @@ -72,6 +73,18 @@ # from the account is picked up without a restart (#5294). Must be a multiple of 60, as run() is only # called once a minute; a cycle that finds a change polls and reads settings whatever else is due. DEVICE_REFRESH_SECONDS = 60 * 60 +# A periodic settings refresh reads every register one by one, and a register the inverter cannot serve +# burns its whole retry budget, so a refresh can run for many minutes. It runs as its own task and the +# cycle waits this long for it before carrying on, so live status and meter data keep being published. +SETTINGS_REFRESH_WAIT_SECONDS = 20 +# A setting whose read has failed this many periodic refreshes in a row, while the device answered its +# other reads, is backed off: not read for SETTING_READ_BACKOFF_MINUTES, doubling with each further +# failure up to SETTING_READ_BACKOFF_MAX_MINUTES, then tried again. Its last known value is kept and +# published as a failed readback. Time-based, so no register goes unread for more than the cap whatever +# the device's refresh cadence. +SETTING_READ_BACKOFF_FAILURES = 3 +SETTING_READ_BACKOFF_MINUTES = 15 +SETTING_READ_BACKOFF_MAX_MINUTES = 60 # An inverter under EMS control is expected to leave its own slot 1 windows spanning the whole day so # that they never override a command from the EMS (see #3781). Anything else is reported. @@ -630,6 +643,21 @@ def initialize(self, ge_cloud_direct, api_key, automatic, automatic_evc=False, e self.evc_devices_dict = {} self.evc_device_list = [] self.settings_from_cache = False + # The periodic settings refresh running in the background, and when it started + self.settings_refresh_task = None + self.settings_refresh_started = None + # Devices due a settings read while a refresh was still running, read by the next one + self.settings_refresh_owed = set() + # Set when the cycle that started the refresh stopped waiting for it + self.settings_refresh_detached = False + # Set when a background refresh failed, so the next cycle reports the failure + self.settings_refresh_error = False + # Every confirmed write is numbered; serial -> setting id -> (number, value) of the latest one, so + # a settings read can tell whether a write landed after it was issued + self.write_seq = 0 + self.recent_writes = {} + # serial -> setting id -> {"failures": consecutive failed reads, "until": not read again before this time} + self.setting_read_backoff = {} self.default_options_stamp = None # Devices found by the periodic device check still waiting for their default options, kept across # cycles so a device found while read only is set up once read only ends @@ -1445,6 +1473,9 @@ async def publish_registers(self, device, registers, select_key=None): ha_name = regname_to_ha(reg_name) attributes = {} attributes["friendly_name"] = reg_name + if registers[key].get("read_failed", False): + # The last read of this register failed, so the state is its last known value, not a fresh readback + attributes["read_failed"] = True is_select_time = False is_select_options = False @@ -2243,7 +2274,7 @@ def apply_devices(self, devices, evc_devices): # Every store rather than just the devices polled until now, so entries that never were polled # (settings restored from the storage cache for hardware that has since gone) are dropped too - stores = (self.status, self.meter, self.info, self.settings, self.pending_writes, self.register_list) + stores = (self.status, self.meter, self.info, self.settings, self.pending_writes, self.register_list, self.setting_read_backoff, self.recent_writes) for device in set().union(*stores, self.ems_slot_warned) - set(device_list): for store in stores: store.pop(device, None) @@ -2437,24 +2468,38 @@ async def run(self, seconds, first): if first or devices_changed or (seconds % (10 * 60) == 0): # Get All registers every now and again in case user changes them - settings_updated = False - for device in self.device_list: - # Under EMS control (polling_mode off) the battery inverters are not the control device, but - # their own registers still have to be re-read periodically or a slot changed behind Predbat's - # back stays invisible until a restart (#5103). `first` rather than `seconds == 0` so a startup - # that only succeeds on a backoff retry still takes the snapshot. - if first or (device in new_devices) or (seconds % SETTINGS_SLOW_REFRESH_SECONDS == 0) or self.polling_mode or (device == self.ems_device) or (device == self.gateway_device): - if first and self.settings_from_cache and device in self.settings: - # Fresh cache loaded on startup — skip the slow poll for this device - await self.publish_registers(device, self.settings[device]) - else: - self.settings[device] = await self.async_get_inverter_settings(device, first=False, previous=self.settings.get(device, {})) - await self.publish_registers(device, self.settings[device]) - settings_updated = True - self.check_ems_inverter_slots(device, self.settings.get(device, {})) - - if settings_updated and self.storage: - await self.storage.save("gecloud", "settings", self.settings, format="json", expiry=None) + # Under EMS control (polling_mode off) the battery inverters are not the control device, but + # their own registers still have to be re-read periodically or a slot changed behind Predbat's + # back stays invisible until a restart (#5103). `first` rather than `seconds == 0` so a startup + # that only succeeds on a backoff retry still takes the snapshot. + refresh_devices = [device for device in self.device_list if first or (device in new_devices) or (seconds % SETTINGS_SLOW_REFRESH_SECONDS == 0) or self.polling_mode or (device == self.ems_device) or (device == self.gateway_device)] + settings_fresh = True + if first or devices_changed: + # Automatic config below is built from these settings, so they are read before carrying on. A + # periodic refresh still running was started for the old device list, so it is stopped first. + await self.cancel_settings_refresh() + await self.refresh_settings(refresh_devices, use_cache=first, backoff=False) + elif self.settings_refresh_running(): + # Remember what was due (the hourly re-read of EMS member inverters, say), so it is read by the + # next refresh rather than waiting for its next turn + self.settings_refresh_owed.update(refresh_devices) + settings_fresh = False + self.log("GECloud: Warn: Settings refresh started {:.0f} seconds ago is still running, not starting another".format((self.now_utc_exact - self.settings_refresh_started).total_seconds())) + else: + refresh_devices = [device for device in self.device_list if device in refresh_devices or device in self.settings_refresh_owed] + self.settings_refresh_owed.clear() + self.settings_refresh_started = self.now_utc_exact + self.settings_refresh_detached = False + self.settings_refresh_task = asyncio.get_running_loop().create_task(self.refresh_settings_background(refresh_devices), name="GECloudSettingsRefresh") + done, _ = await asyncio.wait({self.settings_refresh_task}, timeout=SETTINGS_REFRESH_WAIT_SECONDS) + if not done: + # The refresh applies the default options and AC charge gate itself once it has fresh settings + self.settings_refresh_detached = True + settings_fresh = False + self.log("GECloud: Settings refresh of {} is taking longer than {} seconds, carrying on with live data while it completes".format(refresh_devices, SETTINGS_REFRESH_WAIT_SECONDS)) + elif self.settings_refresh_error: + # It failed, so the settings are the old ones: leave the defaults to a cycle with fresh settings + settings_fresh = False # Run at startup, and again whenever the periodic device check finds the hardware has changed if first or devices_changed: @@ -2472,37 +2517,116 @@ async def run(self, seconds, first): # so a failed report is retried rather than lost - see refresh_discovery(). self.refresh_discovery() - now_utc = self.now_utc_exact - options_due = self.default_options_stamp is None or (now_utc - self.default_options_stamp) >= timedelta(hours=24) - # read_only_now() rather than the switch alone, so an active Axle event (which forces read only - # through the attribute) defers the pass instead of writing over Axle's control (GH#5269) - if not self.read_only_now(): - if options_due: - self.default_options_stamp = now_utc - option_devices = self.device_list - else: - # A device found by the periodic device check is set up now, not at the next 24 hour pass - - # or, when it was found while read only, on the first settings cycle after read only ends - option_devices = [device for device in self.device_list if device in self.pending_default_options] - self.pending_default_options.clear() - for device in option_devices: - await self.enable_default_options(device, self.settings[device]) - - # The 24 hour pass above is only a backstop for the AC charge gate: while a force charge is - # running, re-assert it on every settings refresh in case it was cleared mid-charge (GH#5269). - # Nothing is written while Predbat is not force charging, so an external controller is not fought. - for device in self.device_list: - if self.force_charge_on(device): - await self.ensure_ac_charge_gate(device) + if settings_fresh: + await self.apply_settings_defaults() # Clear pending writes for device in self.device_list: if device in self.pending_writes: self.pending_writes[device] = [] + if self.settings_refresh_error: + # A background settings refresh failed since the last cycle: report it as this cycle failing, as it + # was when the refresh ran inline + self.settings_refresh_error = False + return False + self.update_success_timestamp() return True + async def apply_settings_defaults(self): + """ + Once fresh settings are in hand: set the default options (every 24 hours, or for a new device) and + re-assert the AC charge gate while force charging + """ + now_utc = self.now_utc_exact + options_due = self.default_options_stamp is None or (now_utc - self.default_options_stamp) >= timedelta(hours=24) + # read_only_now() rather than the switch alone, so an active Axle event (which forces read only + # through the attribute) defers the pass instead of writing over Axle's control (GH#5269) + if not self.read_only_now(): + if options_due: + self.default_options_stamp = now_utc + option_devices = self.device_list + else: + # A device found by the periodic device check is set up now, not at the next 24 hour pass - + # or, when it was found while read only, on the first settings cycle after read only ends + option_devices = [device for device in self.device_list if device in self.pending_default_options] + self.pending_default_options.clear() + for device in option_devices: + await self.enable_default_options(device, self.settings[device]) + + # The 24 hour pass above is only a backstop for the AC charge gate: while a force charge is + # running, re-assert it on every settings refresh in case it was cleared mid-charge (GH#5269). + # Nothing is written while Predbat is not force charging, so an external controller is not fought. + for device in self.device_list: + if self.force_charge_on(device): + await self.ensure_ac_charge_gate(device) + + def settings_refresh_running(self): + """Is a periodic settings refresh still running in the background""" + task = self.settings_refresh_task + return task is not None and not task.done() + + async def cancel_settings_refresh(self): + """Stop a background settings refresh and wait for it, and its reads, to finish""" + task = self.settings_refresh_task + self.settings_refresh_task = None + if task is not None and not task.done(): + task.cancel() + # asyncio.wait rather than awaiting the task, so cancelling this caller is not swallowed + await asyncio.wait({task}) + + async def final(self): + """Stop the background settings refresh when the component stops""" + await self.cancel_settings_refresh() + + async def refresh_settings(self, devices, use_cache=False, backoff=False): + """ + Read, cache and publish the settings of devices, then save them to storage + + use_cache publishes the settings restored from storage at startup in place of reading them. A device + dropped by a device list refresh while its settings were being read is not put back. + """ + settings_updated = False + for device in devices: + if device not in self.device_list: + continue + if use_cache and self.settings_from_cache and device in self.settings: + # Fresh cache loaded on startup — skip the slow poll for this device + await self.publish_registers(device, self.settings[device]) + else: + results = await self.async_get_inverter_settings(device, first=False, previous=self.settings.get(device, {}), backoff=backoff) + if device not in self.device_list: + # Gone from the account while it was being read: storing it would bring it back (#5294) + self.register_list.pop(device, None) + continue + self.settings[device] = results + await self.publish_registers(device, results) + settings_updated = True + failed = sorted((sid for sid, register in results.items() if register.get("read_failed", False)), key=str) + if failed: + self.log("GECloud: Warn: Device {} settings refresh could not read setting ids {}, their last known values are kept and marked read_failed".format(device, failed)) + self.check_ems_inverter_slots(device, self.settings.get(device, {})) + + if settings_updated and self.storage: + await self.storage.save("gecloud", "settings", self.settings, format="json", expiry=None) + + async def refresh_settings_background(self, devices): + """The periodic settings refresh, run as its own task so a slow one does not hold up live data""" + try: + await self.refresh_settings(devices, backoff=True) + elapsed = (self.now_utc_exact - self.settings_refresh_started).total_seconds() if self.settings_refresh_started else 0 + if self.settings_refresh_detached: + self.log("GECloud: Settings refresh of {} completed after {:.0f} seconds".format(devices, elapsed)) + # The cycle that started this refresh moved on without it, so apply the defaults here + await self.apply_settings_defaults() + except asyncio.CancelledError: + raise + except Exception as e: + self.log("GECloud: Error: Settings refresh failed: {}".format(e)) + self.log("Error: " + traceback.format_exc()) + self.settings_refresh_error = True + async def async_send_evc_command(self, uuid, command, params): """ Send a command to the EVC @@ -2545,7 +2669,8 @@ async def async_read_inverter_setting(self, serial, setting_id): """ if serial in self.pending_writes: - for pending in self.pending_writes[serial]: + # Newest first, so a setting written twice in a cycle reads back as its latest value + for pending in reversed(self.pending_writes[serial]): if pending["setting_id"] == setting_id: return {"value": pending["value"], "context": "predbat"} @@ -2597,15 +2722,29 @@ async def async_write_inverter_setting(self, serial, setting_id, value): await asyncio.sleep(random.random() * (3 + retry)) else: self.pending_writes[serial].append({"setting_id": setting_id, "value": data_value}) + self.write_seq += 1 + self.recent_writes.setdefault(serial, {})[setting_id] = (self.write_seq, data_value) + # Update the register held in the cache now rather than leaving it to the caller: a settings + # refresh can replace the cached register while this write is in flight, and the caller + # would then update the replaced one and the cache would keep the value read before the write + setting = self.settings.get(serial, {}).get(setting_id, None) + if setting is not None: + setting["value"] = data_value + setting.pop("read_failed", None) break await asyncio.sleep(RETRY_FACTOR * (retry + 1)) if data is None: self.log("GECloud: Warn: Failed to write setting id {}, value {}".format(setting_id, value)) return data - async def async_get_inverter_settings(self, serial, first=False, previous=None): + async def async_get_inverter_settings(self, serial, first=False, previous=None, backoff=False): """ Get settings for account + + A register whose read fails keeps its previous value, marked read_failed so it is not passed off as a + fresh readback. With backoff, a register that has failed SETTING_READ_BACKOFF_FAILURES refreshes in a + row is skipped for a growing number of refreshes rather than costing its whole retry budget each time. + A setting written while the read was in flight keeps the written value, not the older one read back. """ if previous is None: previous = {} @@ -2622,6 +2761,9 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None): pending = [] complete = [] loop = asyncio.get_running_loop() + device_backoff = self.setting_read_backoff.setdefault(serial, {}) + # Writes numbered after this were confirmed after the reads below began + refresh_seq = self.write_seq # Create the read tasks for setting in self.register_list[serial]: @@ -2645,25 +2787,51 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None): future["name"] = name future["validation_rules"] = validation_rules future["validation"] = validation - pending.append(future) + future["skipped"] = False + future["issue_seq"] = refresh_seq + state = device_backoff.get(sid, None) + if backoff and not first and state and state.get("until") and self.now_utc_exact < state["until"]: + future["skipped"] = True + future["data"] = None + complete.append(future) + else: + pending.append(future) # Perform all the reads in parallel - while pending or futures: - while len(futures) < MAX_THREADS and pending: - future = pending.pop(0) + try: + while pending or futures: + while len(futures) < MAX_THREADS and pending: + future = pending.pop(0) + future["issue_seq"] = self.write_seq + if not first: + future["future"] = loop.create_task(self.async_read_inverter_setting(future["serial"], future["sid"]), name=f"ReadSetting-{future['sid']}") + futures.append(future) + if futures: + future = futures[0] + if first: + future["data"] = None + else: + future["data"] = await future["future"] + futures.pop(0) + future["future"] = None + complete.append(future) if not first: - future["future"] = loop.create_task(self.async_read_inverter_setting(future["serial"], future["sid"]), name=f"ReadSetting-{future['sid']}") - futures.append(future) - if futures: - future = futures.pop(0) - if first: - future["data"] = None - else: - future["data"] = await future["future"] - future["future"] = None - complete.append(future) - if not first: - await asyncio.sleep(0.2) + await asyncio.sleep(0.2) + except asyncio.CancelledError: + # Stopped part way (shutdown, or the device list changed): do not leave reads running unowned + outstanding = [future["future"] for future in futures if future.get("future", None) is not None] + for task in outstanding: + task.cancel() + if outstanding: + await asyncio.wait(outstanding) + raise + + # A failure only counts towards backing a register off when the device answered other reads in the + # same refresh: when nothing reads at all the device or GE Cloud is down, not that one register + device_answered = any(future["data"] and ("value" in future["data"]) for future in complete) + if not first and not device_answered and any(not future["skipped"] for future in complete): + # Nothing answered: start every register afresh, so all are read as soon as the device is back + device_backoff.clear() # Wait for all the futures to complete and store results for future in complete: @@ -2677,9 +2845,19 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None): else: value = None - if value is None and sid in results: - # Keep previous failure on read failure - pass + if not first and not future["skipped"] and (value is not None or device_answered): + self.record_setting_read(device_backoff, serial, sid, name, value is not None) + + if value is None: + # Keep the previous value on a read failure, but say it is not a fresh readback (with first + # nothing was read at all, so there is no failure to report) + if sid in results: + failed = dict(results[sid]) + else: + failed = {"name": name, "value": None, "validation_rules": validation_rules, "validation": validation} + if not first: + failed["read_failed"] = True + results[sid] = failed else: results[sid] = { "name": name, @@ -2687,8 +2865,40 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None): "validation_rules": validation_rules, "validation": validation, } + + # A write confirmed after a register's read was issued is newer than what that read returns, so it wins + # (a read issued after the write already reads back the pending write, see async_read_inverter_setting) + issued = {future["sid"]: future["issue_seq"] for future in complete} + for sid, (seq, value) in self.recent_writes.get(serial, {}).items(): + if sid in results and seq > issued.get(sid, refresh_seq): + written = dict(results[sid]) + written["value"] = value + written.pop("read_failed", None) + results[sid] = written return results + def record_setting_read(self, device_backoff, serial, sid, name, success): + """ + Track consecutive failed reads of a setting and back it off once it keeps failing + + An inverter can list a register it cannot serve (for example AC charge slots 8 and 9), and every + read of it times out through the whole retry budget. Backed off, it is not read for 15, 30, then + at most 60 minutes before it is tried again, so it is never hidden for good. + device_backoff is the device's entry in setting_read_backoff taken when the reads started, so a + device dropped by a device list refresh meanwhile is not added back. + """ + if success: + state = device_backoff.pop(sid, None) + if state and state["failures"] >= SETTING_READ_BACKOFF_FAILURES: + self.log("GECloud: Device {} setting id {} ({}) read again after {} failed refreshes".format(serial, sid, name, state["failures"])) + return + state = device_backoff.setdefault(sid, {"failures": 0, "until": None}) + state["failures"] += 1 + if state["failures"] >= SETTING_READ_BACKOFF_FAILURES: + minutes = min(SETTING_READ_BACKOFF_MINUTES * 2 ** (state["failures"] - SETTING_READ_BACKOFF_FAILURES), SETTING_READ_BACKOFF_MAX_MINUTES) + state["until"] = self.now_utc_exact + timedelta(minutes=minutes) + self.log("GECloud: Warn: Device {} setting id {} ({}) has failed {} refreshes in a row, not reading it for {} minutes and keeping its last known value".format(serial, sid, name, state["failures"], minutes)) + async def async_get_smart_device_data(self, uuid): """ Get smart device data points diff --git a/apps/predbat/tests/test_ge_cloud.py b/apps/predbat/tests/test_ge_cloud.py index 9d4d9d55e..23eca895f 100644 --- a/apps/predbat/tests/test_ge_cloud.py +++ b/apps/predbat/tests/test_ge_cloud.py @@ -11,6 +11,7 @@ from gecloud import GECloudDirect, GECloudData, regname_to_ha from gecloud import GE_API_ACCOUNT, GE_API_DEVICES, GE_API_EVC_SEND_COMMAND, GE_API_INVERTER_WRITE_SETTING, GE_API_SITE +from gecloud import GE_API_INVERTER_READ_SETTING, GE_API_INVERTER_SETTINGS, SETTING_READ_BACKOFF_MAX_MINUTES from gecloud import EVCCommandFailed, GECloudTerminalError, SITE_MAX_AGE_MINUTES, parse_site_export_limit from gecloud import DEVICE_REFRESH_SECONDS, SETTINGS_SLOW_REFRESH_SECONDS, find_ems_slot_overrides, normalise_register_time from utils import dp4 @@ -76,6 +77,14 @@ def __init__(self): self._discovery_report = None self._now_utc_exact = datetime.now(timezone.utc) self.settings_from_cache = False + self.settings_refresh_task = None + self.settings_refresh_started = None + self.setting_read_backoff = {} + self.settings_refresh_owed = set() + self.settings_refresh_detached = False + self.settings_refresh_error = False + self.write_seq = 0 + self.recent_writes = {} self.default_options_stamp = None self.pending_default_options = set() self.ems_disabled_ge_cloud_data = False @@ -292,6 +301,12 @@ def test_ge_cloud(my_predbat=None): ("settings_refetch_failed_register_list", _test_inverter_settings_refetch_after_failed_register_list, "Failed register list fetch is retried"), ("settings_saved_to_storage", _test_settings_saved_to_storage, "Settings saved to storage after poll"), ("settings_restored_from_cache", _test_settings_restored_from_fresh_cache, "Settings restored from fresh storage cache"), + ("settings_refresh_live_data", _test_settings_refresh_does_not_hold_up_live_data, "Slow settings refresh does not hold up live data in the serial loop"), + ("settings_refresh_cancelled_on_stop", _test_settings_refresh_cancelled_on_stop, "Stopping cancels and awaits the background settings refresh"), + ("settings_read_vs_write", _test_settings_read_does_not_overwrite_newer_write, "Settings read in flight does not overwrite a newer write"), + ("settings_read_backoff", _test_settings_read_backoff, "Persistently failing setting read is backed off and marked read_failed"), + ("settings_refresh_departed_device", _test_settings_refresh_departed_device, "Departed device does not come back from a background refresh"), + ("settings_refresh_follow_ups", _test_settings_refresh_follow_ups, "Owed devices, deferred default options and failure reporting for the background refresh"), ("inverter_status", _test_async_get_inverter_status, "Get inverter status"), ("inverter_meter", _test_async_get_inverter_meter, "Get inverter meter"), ("status_null_leaves", _test_inverter_status_null_leaves_retained, "Null status leaves retain previous reading"), @@ -2559,7 +2574,7 @@ async def mock_get_device_info(device, previous): async def mock_publish_info(device, info): pass - async def mock_get_inverter_settings(device, first, previous): + async def mock_get_inverter_settings(device, first, previous, backoff=False): return expected_settings async def mock_publish_registers(device, settings): @@ -2641,7 +2656,7 @@ async def mock_get_device_info(device, previous): async def mock_publish_info(device, info): pass - async def mock_get_inverter_settings(device, first, previous): + async def mock_get_inverter_settings(device, first, previous, backoff=False): poll_calls.append(device) return {} @@ -2749,7 +2764,7 @@ async def mock_publish_evc_data(serial, data): async def mock_publish_evc_device(serial, device): call_order.append(f"publish_evc_device:{serial}") - async def mock_get_inverter_settings(device, first, previous): + async def mock_get_inverter_settings(device, first, previous, backoff=False): call_order.append(f"async_get_inverter_settings:{device}") return {} @@ -3022,7 +3037,7 @@ def capture_status(message, had_errors=False, **kwargs): async def benign(*args, **kwargs): return {} - async def mock_get_inverter_settings(device, first, previous): + async def mock_get_inverter_settings(device, first, previous, backoff=False): settings_reads.append(device) return dict(inverter_registers) if device == "inv001" else {} @@ -3157,7 +3172,7 @@ async def get_status(device, previous): calls["status"].append(device) return {"device": device} - async def get_settings(device, first, previous): + async def get_settings(device, first, previous, backoff=False): calls["settings"].append(device) return {} @@ -5509,7 +5524,7 @@ async def mock_publish_info(device, device_info): """No-op info publish.""" return None - async def mock_get_inverter_settings(device, first=False, previous=None): + async def mock_get_inverter_settings(device, first=False, previous=None, backoff=False): """Return this device's fixture register settings, the same shape self.settings[device] holds for real.""" return settings.get(device, {}) @@ -8510,3 +8525,548 @@ async def run_all(): return await check() + await check_storage() + await check_auth() return run_async(run_all()) + + +# ============================================================================= +# Settings refresh isolation: a slow settings read must not hold up live data, and must not overwrite a +# newer confirmed write in the cache + + +ISOLATION_REGISTERS = [ + {"id": 17, "name": "Enable Discharge", "validation_rules": ["boolean"], "validation": None}, + {"id": 120, "name": "AC Charge 8 Start Time", "validation_rules": ["date_format:H:i"], "validation": None}, +] +ISOLATION_SWITCH = "switch.predbat_gecloud_inv001_enable_discharge" + + +def _make_isolation_ge_cloud(devices=("inv001",)): + """ + A component with the real settings, read, write, refresh and publish code, over a fake GE Cloud API. + + Reads of setting id ge.hold_id (120) can be held (ge.hold_reads) and writes held (ge.hold_writes) to order a read + against a write; ge.read_values gives what each setting id reads back, None for a failed read. + """ + ge = MockGECloudDirect() + ge.automatic = False + ge._read_only = True + ge.device_list_reported = list(devices) + ge.read_values = {17: False, 120: "02:00"} + ge.read_calls = [] + ge.read_cancelled = [] + ge.status_calls = [] + ge.hold_id = 120 + ge.hold_reads = None + ge.hold_writes = None + ge.read_started = asyncio.Event() + + async def get_inverter_data(endpoint, serial="", setting_id="", post=False, datain=None, *args, **kwargs): + if endpoint == GE_API_INVERTER_SETTINGS: + return copy.deepcopy(ISOLATION_REGISTERS) + if endpoint == GE_API_INVERTER_READ_SETTING: + ge.read_calls.append((serial, setting_id)) + if setting_id == ge.hold_id and ge.hold_reads is not None: + ge.read_started.set() + try: + await ge.hold_reads.wait() + except asyncio.CancelledError: + ge.read_cancelled.append((serial, setting_id)) + raise + value = ge.read_values.get(setting_id, None) + return None if value is None else {"value": value} + if endpoint == GE_API_INVERTER_WRITE_SETTING: + if ge.hold_writes is not None: + await ge.hold_writes.wait() + return {"value": datain["value"] == "True"} + return None + + async def get_devices(): + return {"battery": list(ge.device_list_reported), "ems": None, "gateway": None, "pv": [], "battery_meters": {}} + + async def get_evc_devices(previous=None): + return [] + + async def get_inverter_status(device, previous): + ge.status_calls.append(device) + return {"time": "fresh"} + + async def get_empty(device, previous): + return {} + + async def nothing(*args, **kwargs): + return None + + ge.async_get_inverter_data = get_inverter_data + ge.async_get_devices = get_devices + ge.async_get_evc_devices = get_evc_devices + ge.async_get_inverter_status = get_inverter_status + ge.async_get_inverter_meter = get_empty + ge.async_get_device_info = get_empty + ge.publish_status = nothing + ge.publish_meter = nothing + ge.publish_info = nothing + ge.publish_site_export_limit = nothing + ge.update_account = nothing + ge.update_site = nothing + return ge + + +class _FastComponentLoop: + """ + Stands in for the asyncio module inside component_base, so ComponentBase.start() runs its real serial + loop with each 5 second tick taking a millisecond. on_tick(seconds) is awaited once per tick. + """ + + def __init__(self, component, on_tick): + self.component = component + self.on_tick = on_tick + self.seconds = 0 + self.real_sleep = asyncio.sleep + + def __getattr__(self, name): + return getattr(asyncio, name) + + async def sleep(self, delay): + if delay == 5: + self.seconds += 5 + await self.on_tick(self.seconds) + await self.real_sleep(0.001) + + +async def _start_component_loop(ge, on_tick, limit=10): + """Run the real ComponentBase.start() loop for ge until on_tick sets api_stop; False if it never finishes""" + import component_base + + ge.api_stop = False + ge.api_started = False + ge.run_timeout = 60 * 60 + ge.count_errors = 0 + ge.base.fatal_error = False + ge.non_fatal_error_occurred = lambda: None + loop = _FastComponentLoop(ge, on_tick) + with patch.object(component_base, "asyncio", loop): + try: + await asyncio.wait_for(ge.start(), timeout=limit) + except asyncio.TimeoutError: + return False + return True + + +def _test_settings_refresh_does_not_hold_up_live_data(my_predbat): + """ + Through the real serial ComponentBase.start() loop: while a periodic settings refresh is stuck on a + register read, the following cycles still fetch live status, no second refresh is started on top of it, + and a write made meanwhile is still in the cache once the held refresh completes. Before the fix the + cycle at 600 seconds awaited the refresh, so the loop never got past it. + """ + + async def test(): + failed = 0 + ge = _make_isolation_ge_cloud() + ge.hold_reads = asyncio.Event() + marks = {} + + async def on_tick(seconds): + if seconds == 600: + # The first cycle has finished; hold the next refresh on setting id 120 + ge.hold_reads.clear() + marks["status_before"] = len(ge.status_calls) + if seconds == 900: + marks["status_during"] = len(ge.status_calls) + marks["reads_during"] = len([call for call in ge.read_calls if call[1] == 120]) + # Predbat turns discharge on while the refresh is still holding its older read of the switch + await ge.switch_event(ISOLATION_SWITCH, "turn_on") + if seconds == 1500: + marks["held_task"] = ge.settings_refresh_task + # The slot changed in the meantime, so the completed refresh must bring the new value in + ge.read_values[120] = "05:30" + ge.hold_reads.set() + if seconds > 1500 and not ge.settings_refresh_running(): + ge.api_stop = True + + ge.hold_reads.set() + with patch("gecloud.SETTINGS_REFRESH_WAIT_SECONDS", 0.05): + finished = await _start_component_loop(ge, on_tick) + if not finished: + print("ERROR: The component loop stalled behind the settings refresh") + return 1 + + # Status every 120 seconds from 600 to 900 is 720 and 840, so live data kept coming + if marks.get("status_during", 0) - marks.get("status_before", 0) < 2: + print("ERROR: Live status was not fetched while the settings refresh was held: {}".format(marks)) + failed += 1 + # Two reads of id 120: startup, then the refresh at 600 - the cycle at 1200 must not start another + if marks.get("reads_during") != 2 or len([call for call in ge.read_calls if call[1] == 120]) != 2: + print("ERROR: Expected one held read of id 120 after startup, got reads {}".format(ge.read_calls)) + failed += 1 + if not any("still running, not starting another" in message for message in ge.log_messages): + print("ERROR: The skipped overlapping refresh was not logged") + failed += 1 + if ge.settings["inv001"][17]["value"] is not True: + print("ERROR: The held refresh replaced the newer write: {}".format(ge.settings["inv001"][17])) + failed += 1 + if ge.dashboard_items.get(ISOLATION_SWITCH, {}).get("state") != "on": + print("ERROR: The switch was not left published on: {}".format(ge.dashboard_items.get(ISOLATION_SWITCH))) + failed += 1 + held = marks.get("held_task") + if held is None or held.cancelled() or not held.done(): + print("ERROR: The held refresh did not run to completion: {}".format(held)) + failed += 1 + if ge.settings["inv001"][120]["value"] != "05:30" or ge.settings["inv001"][120].get("read_failed"): + print("ERROR: The completed refresh did not store the value it read: {}".format(ge.settings["inv001"][120])) + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_refresh_cancelled_on_stop(my_predbat): + """Stopping the component cancels a held background refresh and waits for its read to finish""" + + async def test(): + failed = 0 + ge = _make_isolation_ge_cloud() + ge.hold_reads = asyncio.Event() + ge.hold_reads.set() + + async def on_tick(seconds): + if seconds == 600: + ge.hold_reads.clear() + if seconds == 720: + ge.api_stop = True + + with patch("gecloud.SETTINGS_REFRESH_WAIT_SECONDS", 0.05): + finished = await _start_component_loop(ge, on_tick) + if not finished: + print("ERROR: The component loop stalled behind the settings refresh") + return 1 + if ge.settings_refresh_task is not None: + print("ERROR: final() left the refresh task in place") + failed += 1 + if ge.read_cancelled != [("inv001", 120)]: + print("ERROR: The held read was not cancelled and awaited: {}".format(ge.read_cancelled)) + failed += 1 + leftover = [task for task in asyncio.all_tasks() if task is not asyncio.current_task() and task.get_name().startswith(("ReadSetting-", "GECloudSettingsRefresh"))] + if leftover: + print("ERROR: Tasks left running after stop: {}".format(leftover)) + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_read_does_not_overwrite_newer_write(my_predbat): + """ + A write confirmed while a settings read of the same register is in flight is not replaced in the cache + by the older value read back - whether the write lands before the read returns, or after the refresh + has stored its results + """ + + async def test(): + failed = 0 + + async def refresh(ge): + """Read the settings and store and publish them, as a settings refresh does""" + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings.get("inv001", {})) + await ge.publish_registers("inv001", ge.settings["inv001"]) + + # The write lands while the read of the register is held + ge = _make_isolation_ge_cloud() + ge.device_list = ["inv001"] + ge.pending_writes = {"inv001": []} + await refresh(ge) + ge.hold_id = 17 + ge.hold_reads = asyncio.Event() + ge.read_started.clear() + reading = asyncio.ensure_future(ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"])) + await asyncio.wait_for(ge.read_started.wait(), 1) + await ge.switch_event(ISOLATION_SWITCH, "turn_on") + ge.hold_reads.set() + ge.settings["inv001"] = await asyncio.wait_for(reading, 2) + if ge.settings["inv001"][17]["value"] is not True: + print("ERROR: Read in flight across a write replaced it: {}".format(ge.settings["inv001"][17])) + failed += 1 + + # The refresh stores its results while the write is still in flight + ge = _make_isolation_ge_cloud() + ge.device_list = ["inv001"] + ge.pending_writes = {"inv001": []} + await refresh(ge) + ge.hold_writes = asyncio.Event() + write = asyncio.ensure_future(ge.switch_event(ISOLATION_SWITCH, "turn_on")) + await asyncio.sleep(0) + await refresh(ge) + ge.hold_writes.set() + await asyncio.wait_for(write, 2) + if ge.settings["inv001"][17]["value"] is not True: + print("ERROR: A write completing after the refresh stored its results was lost: {}".format(ge.settings["inv001"][17])) + failed += 1 + if ge.dashboard_items.get(ISOLATION_SWITCH, {}).get("state") != "on": + print("ERROR: The switch was published from the replaced register: {}".format(ge.dashboard_items.get(ISOLATION_SWITCH))) + failed += 1 + + # Written twice in one cycle, the setting reads back as the later write + ge.pending_writes["inv001"] = [{"setting_id": 17, "value": True}, {"setting_id": 17, "value": False}] + data = await ge.async_read_inverter_setting("inv001", 17) + if data != {"value": False, "context": "predbat"}: + print("ERROR: Pending write read back as the older value: {}".format(data)) + failed += 1 + + # Once the write is no longer pending, a read issued after it is the inverter's own state: a change + # made outside Predbat since (in the GivEnergy app, say) is not masked by Predbat's earlier write + ge.pending_writes["inv001"] = [] + ge.read_values[17] = False + await refresh(ge) + if ge.settings["inv001"][17]["value"] is not False: + print("ERROR: An earlier write masked a later change read from the inverter: {}".format(ge.settings["inv001"][17])) + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_read_backoff(my_predbat): + """ + A setting whose reads keep failing is backed off on the periodic refresh, keeps its last known value + marked read_failed (and published so), is tried again after the backoff, and recovers on a good read + """ + + async def test(): + failed = 0 + ge = _make_isolation_ge_cloud() + ge.device_list = ["inv001"] + ge.pending_writes = {"inv001": []} + with patch("gecloud.RETRIES", 1), patch("gecloud.RETRY_FACTOR", 0): + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous={}) + ge.read_values[120] = None + read_pattern = [] + for _ in range(17): + # A refresh every 10 minutes, as in polling mode + ge._now_utc_exact += timedelta(minutes=10) + before = len(ge.read_calls) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + read_pattern.append(any(call[1] == 120 for call in ge.read_calls[before:])) + # Fails 1, 2, 3 (15 min off: skip 1), fails 4 (30 min: skip 2), fails 5 (60 min: skip 5), + # fails 6 (capped at 60 min: skip 5), read again + expected = [True, True, True, False, True, False, False, True, False, False, False, False, False, True, False, False, False] + if read_pattern != expected: + print("ERROR: Backoff read pattern {} expected {}".format(read_pattern, expected)) + failed += 1 + register = ge.settings["inv001"][120] + if register.get("value") != "02:00" or not register.get("read_failed"): + print("ERROR: Failed register should keep its last value marked read_failed: {}".format(register)) + failed += 1 + await ge.publish_registers("inv001", ge.settings["inv001"]) + published = ge.dashboard_items.get("select.predbat_gecloud_inv001_ac_charge_8_start_time", {}) + if not published.get("attributes", {}).get("read_failed"): + print("ERROR: read_failed not published: {}".format(published)) + failed += 1 + + # Without backoff (startup, a new device) every register is read + before = len(ge.read_calls) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"]) + if not any(call[1] == 120 for call in ge.read_calls[before:]): + print("ERROR: A read without backoff skipped the backed off register") + failed += 1 + + # Good again: read on the next try after the backoff, and the flag and backoff are cleared + ge.read_values[120] = "03:00" + for _ in range(SETTING_READ_BACKOFF_MAX_MINUTES // 10 + 1): + ge._now_utc_exact += timedelta(minutes=10) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + register = ge.settings["inv001"][120] + if register.get("value") != "03:00" or register.get("read_failed") or 120 in ge.setting_read_backoff.get("inv001", {}): + print("ERROR: Register did not recover: {} backoff {}".format(register, ge.setting_read_backoff)) + failed += 1 + if ge.settings["inv001"][17].get("read_failed"): + print("ERROR: A good register was marked read_failed") + failed += 1 + + # The device or GE Cloud being down fails every read: that is not a register it cannot serve, so + # nothing is backed off - even a register already backed off is read the moment the device is back + ge.read_values = {17: True, 120: None} + for _ in range(3): + ge._now_utc_exact += timedelta(minutes=10) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + if not ge.setting_read_backoff.get("inv001", {}).get(120, {}).get("until"): + print("ERROR: Expected id 120 backed off before the outage: {}".format(ge.setting_read_backoff)) + failed += 1 + ge.read_values = {17: None, 120: None} + for _ in range(8): + ge._now_utc_exact += timedelta(minutes=1) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + if ge.setting_read_backoff.get("inv001"): + print("ERROR: An outage backed registers off: {}".format(ge.setting_read_backoff)) + failed += 1 + ge.read_values = {17: True, 120: "04:00"} + before = len(ge.read_calls) + ge._now_utc_exact += timedelta(minutes=1) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + if sorted(call[1] for call in ge.read_calls[before:]) != [17, 120] or ge.settings["inv001"][120]["value"] != "04:00": + print("ERROR: Registers not read on the first refresh after an outage: {}".format(ge.read_calls[before:])) + failed += 1 + + # An EMS member inverter is only re-read once an hour: a register that keeps failing there is still read on + # every one of those refreshes, as the backoff never keeps a register unread for more than an hour + ge = _make_isolation_ge_cloud() + ge.device_list = ["inv001"] + ge.pending_writes = {"inv001": []} + with patch("gecloud.RETRIES", 1), patch("gecloud.RETRY_FACTOR", 0): + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous={}) + ge.read_values[120] = None + last_read = ge.now_utc_exact + longest_gap = timedelta(0) + for _ in range(8): + ge._now_utc_exact += timedelta(minutes=SETTING_READ_BACKOFF_MAX_MINUTES) + before = len(ge.read_calls) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + if any(call[1] == 120 for call in ge.read_calls[before:]): + longest_gap = max(longest_gap, ge.now_utc_exact - last_read) + last_read = ge.now_utc_exact + longest_gap = max(longest_gap, ge.now_utc_exact - last_read) + if longest_gap > timedelta(minutes=SETTING_READ_BACKOFF_MAX_MINUTES) or ge.setting_read_backoff["inv001"][120]["failures"] != 8: + print("ERROR: Hourly refreshed register went {} unread: {}".format(longest_gap, ge.setting_read_backoff)) + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_refresh_follow_ups(my_predbat): + """ + Devices due while a refresh is still running are read by the next one; the default options pass waits + for fresh settings rather than running on the old ones; a failed background refresh fails the cycle + """ + + async def test(): + failed = 0 + with patch("gecloud.SETTINGS_REFRESH_WAIT_SECONDS", 0.05): + # Under an EMS the member inverter inv001 is only re-read hourly; that slot falls while a refresh + # of the EMS is held, so the next refresh must read it + ge = _make_isolation_ge_cloud(devices=("inv001", "inv002")) + await ge.run(seconds=0, first=True) + ge.polling_mode = False + ge.ems_device = "inv002" + ge.hold_reads = asyncio.Event() + ge.read_started.clear() + await asyncio.wait_for(ge.run(seconds=SETTINGS_SLOW_REFRESH_SECONDS - 600, first=False), 5) + await asyncio.wait_for(ge.read_started.wait(), 1) + await asyncio.wait_for(ge.run(seconds=SETTINGS_SLOW_REFRESH_SECONDS, first=False), 5) + ge.hold_reads.set() + await asyncio.wait_for(ge.settings_refresh_task, 5) + ge.hold_reads = None + before = len(ge.read_calls) + await asyncio.wait_for(ge.run(seconds=SETTINGS_SLOW_REFRESH_SECONDS + 600, first=False), 5) + await asyncio.wait_for(ge.settings_refresh_task, 5) + if not any(call[0] == "inv001" for call in ge.read_calls[before:]): + print("ERROR: The hourly re-read skipped behind a running refresh was lost: {}".format(ge.read_calls[before:])) + failed += 1 + + # The default options pass is due; the refresh is slow, so it runs once that refresh completes + ge = _make_isolation_ge_cloud() + option_calls = [] + + async def record_options(device, registers): + option_calls.append((device, registers[120]["value"])) + + ge.enable_default_options = record_options + await ge.run(seconds=0, first=True) + ge._read_only = False + ge.default_options_stamp = ge.now_utc_exact - timedelta(hours=25) + ge.hold_reads = asyncio.Event() + ge.read_values[120] = "06:00" + await asyncio.wait_for(ge.run(seconds=600, first=False), 5) + if option_calls: + print("ERROR: Default options ran on the settings from before the refresh: {}".format(option_calls)) + failed += 1 + ge.hold_reads.set() + await asyncio.wait_for(ge.settings_refresh_task, 5) + if option_calls != [("inv001", "06:00")]: + print("ERROR: Default options did not run on the fresh settings: {}".format(option_calls)) + failed += 1 + + # A refresh that raises fails the cycle, with the traceback logged, and the next cycle is clean. + # The default options pass is due, but must not run on the settings the failed refresh left behind + ge = _make_isolation_ge_cloud() + option_calls = [] + ge.enable_default_options = record_options + await ge.run(seconds=0, first=True) + option_calls.clear() + ge._read_only = False + ge.default_options_stamp = ge.now_utc_exact - timedelta(hours=25) + + async def broken(*args, **kwargs): + raise ValueError("bad register list") + + ge.async_get_inverter_settings = broken + result = await asyncio.wait_for(ge.run(seconds=600, first=False), 5) + if result is not False or not any("Traceback" in message for message in ge.log_messages): + print("ERROR: A failed background refresh was not reported: result {}".format(result)) + failed += 1 + if option_calls: + print("ERROR: Default options ran on stale settings after the refresh failed: {}".format(option_calls)) + failed += 1 + if await ge.run(seconds=660, first=False) is not True: + print("ERROR: The cycle after a reported refresh failure should succeed") + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_refresh_departed_device(my_predbat): + """ + A device removed from the account while its settings are being read in the background is not put + back in the cache or published, and a device list change stops the background refresh before reading + the settings itself + """ + + async def test(): + failed = 0 + ge = _make_isolation_ge_cloud(devices=("inv001", "inv002")) + with patch("gecloud.SETTINGS_REFRESH_WAIT_SECONDS", 0.05): + await ge.run(seconds=0, first=True) + ge.hold_reads = asyncio.Event() + ge.read_started.clear() + try: + await asyncio.wait_for(ge.run(seconds=600, first=False), 5) + except asyncio.TimeoutError: + print("ERROR: run() at 600 seconds waited for the held settings refresh") + return 1 + await asyncio.wait_for(ge.read_started.wait(), 1) + if not ge.settings_refresh_running(): + print("ERROR: Expected the refresh to be held") + return 1 + held = ge.settings_refresh_task + + # inv001 is held mid read; take it off the account and let the read finish + ge.apply_devices({"battery": ["inv002"], "ems": None, "gateway": None, "pv": [], "battery_meters": {}}, []) + for entity_id in [entity_id for entity_id in ge.dashboard_items if "inv001" in entity_id]: + del ge.dashboard_items[entity_id] + ge.hold_reads.set() + await asyncio.wait_for(held, 2) + if "inv001" in ge.settings or "inv001" in ge.setting_read_backoff or any("inv001" in entity_id for entity_id in ge.dashboard_items) or any(entry["device"] == "inv001" for entry in ge.register_entity_map.values()): + print("ERROR: Departed device came back: settings {} entities {}".format(list(ge.settings), [e for e in ge.dashboard_items if "inv001" in e])) + failed += 1 + + # A device list change while a refresh is held cancels it, then reads the settings in line + ge.hold_reads = asyncio.Event() + ge.read_started.clear() + try: + await asyncio.wait_for(ge.run(seconds=1200, first=False), 5) + except asyncio.TimeoutError: + print("ERROR: run() at 1200 seconds waited for the held settings refresh") + return 1 + await asyncio.wait_for(ge.read_started.wait(), 1) + held = ge.settings_refresh_task + ge.hold_reads = None + ge.device_list_reported = ["inv002", "inv003"] + await asyncio.wait_for(ge.run(seconds=DEVICE_REFRESH_SECONDS, first=False), 5) + if not held.cancelled() or ge.read_cancelled[-1:] != [("inv002", 120)]: + print("ERROR: Device list change did not cancel the held refresh: {} {}".format(held, ge.read_cancelled)) + failed += 1 + if "inv003" not in ge.settings or ge.settings_refresh_running(): + print("ERROR: New device's settings were not read in line: {}".format(list(ge.settings))) + failed += 1 + return failed + + return run_async(test())