Skip to content

Commit 9df6f46

Browse files
mgazzaclaude
andcommitted
fix(gecloud): address review of the settings refresh isolation
- A write now wins over a settings read only when the write was confirmed after that register's read was issued, using a write sequence number. pending_writes is cleared every cycle again, as before, so Predbat's own earlier write no longer masks a change made in the GivEnergy app for the length of a slow refresh. - A read failure only counts towards backoff when the device answered other reads in the same refresh. An inverter or GE Cloud outage no longer backs every register off and blinds the refreshes after it. - Devices that were due a read while a refresh was still running (such as the hourly re-read of EMS member inverters) are read by the next refresh instead of waiting for their next slot. - The default options pass and the AC charge gate run on fresh settings. When the refresh outlasts the wait, it runs them itself on completion instead of the cycle using the old snapshot. - A failed background refresh logs a traceback and fails the next cycle, so it is counted as an error as it was when the refresh ran inline. - The register list of a device removed during a refresh is dropped. - Test: the held refresh in the serial loop test now runs to completion and must store the new value it reads. Previously it was cancelled at stop and the test could not see the merge. Added outage, owed-device, deferred-defaults, failure-reporting and external-change cases. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf
1 parent a3f8387 commit 9df6f46

2 files changed

Lines changed: 204 additions & 46 deletions

File tree

‎apps/predbat/gecloud.py‎

Lines changed: 91 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import math
2424
import json
2525
import random
26+
import traceback
2627
from component_base import ComponentBase
2728
from coordinator import inverter_record
2829
from mock_base import MockBase as SharedMockBase
@@ -642,6 +643,16 @@ def initialize(self, ge_cloud_direct, api_key, automatic, automatic_evc=False, e
642643
# The periodic settings refresh running in the background, and when it started
643644
self.settings_refresh_task = None
644645
self.settings_refresh_started = None
646+
# Devices due a settings read while a refresh was still running, read by the next one
647+
self.settings_refresh_owed = set()
648+
# Set when the cycle that started the refresh stopped waiting for it
649+
self.settings_refresh_detached = False
650+
# Set when a background refresh failed, so the next cycle reports the failure
651+
self.settings_refresh_error = False
652+
# Every confirmed write is numbered; serial -> setting id -> (number, value) of the latest one, so
653+
# a settings read can tell whether a write landed after it was issued
654+
self.write_seq = 0
655+
self.recent_writes = {}
645656
# serial -> setting id -> {"failures": consecutive failed reads, "skip": refreshes left to skip}
646657
self.setting_read_backoff = {}
647658
self.default_options_stamp = None
@@ -2260,7 +2271,7 @@ def apply_devices(self, devices, evc_devices):
22602271

22612272
# Every store rather than just the devices polled until now, so entries that never were polled
22622273
# (settings restored from the storage cache for hardware that has since gone) are dropped too
2263-
stores = (self.status, self.meter, self.info, self.settings, self.pending_writes, self.register_list, self.setting_read_backoff)
2274+
stores = (self.status, self.meter, self.info, self.settings, self.pending_writes, self.register_list, self.setting_read_backoff, self.recent_writes)
22642275
for device in set().union(*stores, self.ems_slot_warned) - set(device_list):
22652276
for store in stores:
22662277
store.pop(device, None)
@@ -2459,18 +2470,29 @@ async def run(self, seconds, first):
24592470
# back stays invisible until a restart (#5103). `first` rather than `seconds == 0` so a startup
24602471
# that only succeeds on a backoff retry still takes the snapshot.
24612472
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)]
2473+
settings_fresh = True
24622474
if first or devices_changed:
24632475
# Automatic config below is built from these settings, so they are read before carrying on. A
24642476
# periodic refresh still running was started for the old device list, so it is stopped first.
24652477
await self.cancel_settings_refresh()
24662478
await self.refresh_settings(refresh_devices, use_cache=first, backoff=False)
24672479
elif self.settings_refresh_running():
2480+
# Remember what was due (the hourly re-read of EMS member inverters, say), so it is read by the
2481+
# next refresh rather than waiting for its next turn
2482+
self.settings_refresh_owed.update(refresh_devices)
2483+
settings_fresh = False
24682484
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()))
24692485
else:
2486+
refresh_devices = [device for device in self.device_list if device in refresh_devices or device in self.settings_refresh_owed]
2487+
self.settings_refresh_owed.clear()
24702488
self.settings_refresh_started = self.now_utc_exact
2489+
self.settings_refresh_detached = False
24712490
self.settings_refresh_task = asyncio.get_running_loop().create_task(self.refresh_settings_background(refresh_devices), name="GECloudSettingsRefresh")
24722491
done, _ = await asyncio.wait({self.settings_refresh_task}, timeout=SETTINGS_REFRESH_WAIT_SECONDS)
24732492
if not done:
2493+
# The refresh applies the default options and AC charge gate itself once it has fresh settings
2494+
self.settings_refresh_detached = True
2495+
settings_fresh = False
24742496
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))
24752497

24762498
# Run at startup, and again whenever the periodic device check finds the hardware has changed
@@ -2489,39 +2511,51 @@ async def run(self, seconds, first):
24892511
# so a failed report is retried rather than lost - see refresh_discovery().
24902512
self.refresh_discovery()
24912513

2492-
now_utc = self.now_utc_exact
2493-
options_due = self.default_options_stamp is None or (now_utc - self.default_options_stamp) >= timedelta(hours=24)
2494-
# read_only_now() rather than the switch alone, so an active Axle event (which forces read only
2495-
# through the attribute) defers the pass instead of writing over Axle's control (GH#5269)
2496-
if not self.read_only_now():
2497-
if options_due:
2498-
self.default_options_stamp = now_utc
2499-
option_devices = self.device_list
2500-
else:
2501-
# A device found by the periodic device check is set up now, not at the next 24 hour pass -
2502-
# or, when it was found while read only, on the first settings cycle after read only ends
2503-
option_devices = [device for device in self.device_list if device in self.pending_default_options]
2504-
self.pending_default_options.clear()
2505-
for device in option_devices:
2506-
await self.enable_default_options(device, self.settings[device])
2507-
2508-
# The 24 hour pass above is only a backstop for the AC charge gate: while a force charge is
2509-
# running, re-assert it on every settings refresh in case it was cleared mid-charge (GH#5269).
2510-
# Nothing is written while Predbat is not force charging, so an external controller is not fought.
2511-
for device in self.device_list:
2512-
if self.force_charge_on(device):
2513-
await self.ensure_ac_charge_gate(device)
2514+
if settings_fresh:
2515+
await self.apply_settings_defaults()
25142516

2515-
# Clear pending writes - unless a settings refresh is still reading, as it needs them to tell a write
2516-
# made while it was in flight from the older value it read back
2517-
if not self.settings_refresh_running():
2518-
for device in self.device_list:
2519-
if device in self.pending_writes:
2520-
self.pending_writes[device] = []
2517+
# Clear pending writes
2518+
for device in self.device_list:
2519+
if device in self.pending_writes:
2520+
self.pending_writes[device] = []
2521+
2522+
if self.settings_refresh_error:
2523+
# A background settings refresh failed since the last cycle: report it as this cycle failing, as it
2524+
# was when the refresh ran inline
2525+
self.settings_refresh_error = False
2526+
return False
25212527

25222528
self.update_success_timestamp()
25232529
return True
25242530

2531+
async def apply_settings_defaults(self):
2532+
"""
2533+
Once fresh settings are in hand: set the default options (every 24 hours, or for a new device) and
2534+
re-assert the AC charge gate while force charging
2535+
"""
2536+
now_utc = self.now_utc_exact
2537+
options_due = self.default_options_stamp is None or (now_utc - self.default_options_stamp) >= timedelta(hours=24)
2538+
# read_only_now() rather than the switch alone, so an active Axle event (which forces read only
2539+
# through the attribute) defers the pass instead of writing over Axle's control (GH#5269)
2540+
if not self.read_only_now():
2541+
if options_due:
2542+
self.default_options_stamp = now_utc
2543+
option_devices = self.device_list
2544+
else:
2545+
# A device found by the periodic device check is set up now, not at the next 24 hour pass -
2546+
# or, when it was found while read only, on the first settings cycle after read only ends
2547+
option_devices = [device for device in self.device_list if device in self.pending_default_options]
2548+
self.pending_default_options.clear()
2549+
for device in option_devices:
2550+
await self.enable_default_options(device, self.settings[device])
2551+
2552+
# The 24 hour pass above is only a backstop for the AC charge gate: while a force charge is
2553+
# running, re-assert it on every settings refresh in case it was cleared mid-charge (GH#5269).
2554+
# Nothing is written while Predbat is not force charging, so an external controller is not fought.
2555+
for device in self.device_list:
2556+
if self.force_charge_on(device):
2557+
await self.ensure_ac_charge_gate(device)
2558+
25252559
def settings_refresh_running(self):
25262560
"""Is a periodic settings refresh still running in the background"""
25272561
task = self.settings_refresh_task
@@ -2558,6 +2592,7 @@ async def refresh_settings(self, devices, use_cache=False, backoff=False):
25582592
results = await self.async_get_inverter_settings(device, first=False, previous=self.settings.get(device, {}), backoff=backoff)
25592593
if device not in self.device_list:
25602594
# Gone from the account while it was being read: storing it would bring it back (#5294)
2595+
self.register_list.pop(device, None)
25612596
continue
25622597
self.settings[device] = results
25632598
await self.publish_registers(device, results)
@@ -2574,14 +2609,17 @@ async def refresh_settings_background(self, devices):
25742609
"""The periodic settings refresh, run as its own task so a slow one does not hold up live data"""
25752610
try:
25762611
await self.refresh_settings(devices, backoff=True)
2612+
elapsed = (self.now_utc_exact - self.settings_refresh_started).total_seconds() if self.settings_refresh_started else 0
2613+
if self.settings_refresh_detached:
2614+
self.log("GECloud: Settings refresh of {} completed after {:.0f} seconds".format(devices, elapsed))
2615+
# The cycle that started this refresh moved on without it, so apply the defaults here
2616+
await self.apply_settings_defaults()
25772617
except asyncio.CancelledError:
25782618
raise
25792619
except Exception as e:
25802620
self.log("GECloud: Error: Settings refresh failed: {}".format(e))
2581-
return
2582-
elapsed = (self.now_utc_exact - self.settings_refresh_started).total_seconds() if self.settings_refresh_started else 0
2583-
if elapsed > SETTINGS_REFRESH_WAIT_SECONDS:
2584-
self.log("GECloud: Settings refresh of {} completed after {:.0f} seconds".format(devices, elapsed))
2621+
self.log("Error: " + traceback.format_exc())
2622+
self.settings_refresh_error = True
25852623

25862624
async def async_send_evc_command(self, uuid, command, params):
25872625
"""
@@ -2678,6 +2716,8 @@ async def async_write_inverter_setting(self, serial, setting_id, value):
26782716
await asyncio.sleep(random.random() * (3 + retry))
26792717
else:
26802718
self.pending_writes[serial].append({"setting_id": setting_id, "value": data_value})
2719+
self.write_seq += 1
2720+
self.recent_writes.setdefault(serial, {})[setting_id] = (self.write_seq, data_value)
26812721
# Update the register held in the cache now rather than leaving it to the caller: a settings
26822722
# refresh can replace the cached register while this write is in flight, and the caller
26832723
# would then update the replaced one and the cache would keep the value read before the write
@@ -2716,6 +2756,8 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None,
27162756
complete = []
27172757
loop = asyncio.get_running_loop()
27182758
device_backoff = self.setting_read_backoff.setdefault(serial, {})
2759+
# Writes numbered after this were confirmed after the reads below began
2760+
refresh_seq = self.write_seq
27192761

27202762
# Create the read tasks
27212763
for setting in self.register_list[serial]:
@@ -2740,6 +2782,7 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None,
27402782
future["validation_rules"] = validation_rules
27412783
future["validation"] = validation
27422784
future["skipped"] = False
2785+
future["issue_seq"] = refresh_seq
27432786
state = device_backoff.get(sid, None)
27442787
if backoff and not first and state and state["skip"] > 0:
27452788
state["skip"] -= 1
@@ -2754,6 +2797,7 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None,
27542797
while pending or futures:
27552798
while len(futures) < MAX_THREADS and pending:
27562799
future = pending.pop(0)
2800+
future["issue_seq"] = self.write_seq
27572801
if not first:
27582802
future["future"] = loop.create_task(self.async_read_inverter_setting(future["serial"], future["sid"]), name=f"ReadSetting-{future['sid']}")
27592803
futures.append(future)
@@ -2777,6 +2821,10 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None,
27772821
await asyncio.wait(outstanding)
27782822
raise
27792823

2824+
# A failure only counts towards backing a register off when the device answered other reads in the
2825+
# same refresh: when nothing reads at all the device or GE Cloud is down, not that one register
2826+
device_answered = any(future["data"] and ("value" in future["data"]) for future in complete)
2827+
27802828
# Wait for all the futures to complete and store results
27812829
for future in complete:
27822830
sid = future["sid"]
@@ -2789,7 +2837,7 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None,
27892837
else:
27902838
value = None
27912839

2792-
if not first and not future["skipped"]:
2840+
if not first and not future["skipped"] and (value is not None or device_answered):
27932841
self.record_setting_read(device_backoff, serial, sid, name, value is not None)
27942842

27952843
if value is None:
@@ -2810,15 +2858,15 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None,
28102858
"validation": validation,
28112859
}
28122860

2813-
# A write confirmed while these reads were in flight is newer than what they read back, so it wins
2814-
# (the same rule async_read_inverter_setting applies to a read issued after the write)
2815-
for pending_write in self.pending_writes.get(serial, []):
2816-
sid = pending_write["setting_id"]
2817-
if sid in results:
2818-
written = dict(results[sid])
2819-
written["value"] = pending_write["value"]
2820-
written.pop("read_failed", None)
2821-
results[sid] = written
2861+
# A write confirmed after a register's read was issued is newer than what that read returns, so it wins
2862+
# (a read issued after the write already reads back the pending write, see async_read_inverter_setting)
2863+
issued = {future["sid"]: future["issue_seq"] for future in complete}
2864+
for sid, (seq, value) in self.recent_writes.get(serial, {}).items():
2865+
if sid in results and seq > issued.get(sid, refresh_seq):
2866+
written = dict(results[sid])
2867+
written["value"] = value
2868+
written.pop("read_failed", None)
2869+
results[sid] = written
28222870
return results
28232871

28242872
def record_setting_read(self, device_backoff, serial, sid, name, success):

0 commit comments

Comments
 (0)