fix(enphase): age livestream readings out instead of dropping them

Clearing the reading on a failed read made a single missed connection flip
all four sensors onto the bucket fallback, which lags 15-30 minutes - a
bigger visible step than simply holding the last measurement a little
longer. A reading now stays in use for ENPHASE_LIVE_MAX_AGE_MINUTES and is
ignored after that, so a blip is absorbed while genuinely old data still
stops being presented as current.

Readings are stamped on arrival because DataMsg.timestamp is a constant
and the payload cannot date itself. They remain in-memory only, so nothing
survives a restart.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: springfall2008

apps/predbat/enphase.py

@@ -68,6 +68,10 @@
 # 1 would be the just-closed bucket, which the cloud is still back-filling; 2 is settled.
 ENPHASE_SETTLED_BUCKETS = 2
 ENPHASE_LIVESTREAM_TIMEOUT = 15  # seconds to wait for a livestream message before giving up
+# How long a livestream reading stays usable. Holding it over a missed poll avoids flipping the
+# sensors onto the 15-30 minute bucket fallback for a single blip, but it is instantaneous data
+# with no timestamp of its own, so it must not be published indefinitely either.
+ENPHASE_LIVE_MAX_AGE_MINUTES = 15
 LIVESTREAM_BOOTSTRAP = "/pv/aws_sigv4/livestream.json"  # returns the AWS IoT endpoint, topic and authorizer credentials
 
 # live_power is deliberately absent: livestream readings are instantaneous and carry no usable
@@ -603,6 +607,8 @@ async def publish_data(self, site_id):
         # instantaneous, where the buckets are a 15-minute average and load is only ever a residual.
         # The bucket values above stay as the fallback for when the stream is unavailable.
         live = self.live_power.get(site_id) or {}
+        if live and (now_ts - live.get("read_ts", 0)) > ENPHASE_LIVE_MAX_AGE_MINUTES * 60:
+            live = {}  # too old to present as current; fall back to the bucket values below
         if live:
             pv_power = live.get("pv", pv_power)
             # The livestream reports grid negative while exporting, the opposite of Predbat's sign.
@@ -1207,12 +1213,11 @@ async def get_live_power(self, site_id):
         """
         reading = await self._fetch_live_power(site_id)
         if reading:
+            # Stamped on arrival: DataMsg.timestamp is a constant, so the payload cannot date itself.
+            reading["read_ts"] = datetime.now(timezone.utc).timestamp()
             self.live_power[site_id] = reading
-        else:
-            # Drop any previous reading rather than let publish_data republish it. These values are
-            # instantaneous, so a stale measurement presented as current is worse than falling back
-            # to the (lagging but genuinely current) bucket values.
-            self.live_power.pop(site_id, None)
+        # A failure deliberately leaves any previous reading alone - publish_data ages it out after
+        # ENPHASE_LIVE_MAX_AGE_MINUTES rather than dropping to the lagging buckets over one blip.
         return reading
 
     async def _fetch_live_power(self, site_id):

apps/predbat/tests/test_enphase_api.py

@@ -669,9 +669,13 @@ def test_gateway_serial_read_from_today():
     assert gateway_serial({}) is None
 
 
+BUCKET_START = 1783724400  # local midnight of the day the /today buckets belong to
+FROZEN_NOW_TS = BUCKET_START + int(82.5 * 900)  # inside index 82, so the settled bucket read is 80
+
+
 def _publish_with_buckets(api, production=250, imp=100, exp=0, charge=200, discharge=0):
     """Publish a site from /today buckets; the defaults give pv 1000 W, importing 400 W, charging 800 W."""
-    start = 1783724400
+    start = BUCKET_START
 
     def bucket(value):
         """Return a 96-slot Wh array with the bucket read by publish_data set to value."""
@@ -693,7 +697,7 @@ class _Fixed(original):
         @classmethod
         def now(cls, tz=None):
             """Return a time inside index 82, so the settled bucket publish_data reads is 80."""
-            return original.fromtimestamp(start + int(82.5 * 900), tz)
+            return original.fromtimestamp(FROZEN_NOW_TS, tz)
 
     enphase_module.datetime = _Fixed
     try:
@@ -711,7 +715,7 @@ def test_publish_prefers_the_measured_livestream_reading():
     the 15-minute buckets on every count.
     """
     api = MockEnphaseAPI()
-    api.live_power["12345"] = {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100}
+    api.live_power["12345"] = {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100, "read_ts": FROZEN_NOW_TS - 60}
     published = _publish_with_buckets(api)
     assert published == {"pv": 4632.8, "grid": 2686.1, "battery": 32.0, "load": 1978.7}  # grid flipped to +export
 
@@ -728,18 +732,34 @@ def test_live_power_is_never_persisted():
     assert "live_power" not in ENPHASE_CACHE_KEYS
 
 
-def test_failed_live_read_drops_the_previous_reading():
-    """A failed livestream read clears the last reading rather than leaving it to be republished.
+def test_failed_live_read_keeps_the_recent_reading():
+    """A failed read leaves the last reading in place so a blip does not flip the sensors.
 
-    Falling back to the (lagging but current) bucket values beats presenting a stale measurement
-    as though it were live.
+    The bucket fallback lags 15-30 minutes, so bouncing onto it for one missed cycle would be a
+    bigger step than simply holding the measurement a little longer.
     """
     api = MockEnphaseAPI()
     api.today["12345"] = {"serial": "122530006866"}
-    api.live_power["12345"] = {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100}
+    reading = {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100, "read_ts": FROZEN_NOW_TS - 60}
+    api.live_power["12345"] = dict(reading)
     # No canned response for the bootstrap, so it 404s and the read fails
     assert run_async(api.get_live_power("12345")) is None
-    assert "12345" not in api.live_power
+    assert api.live_power["12345"] == reading
+
+
+def test_live_reading_older_than_the_window_falls_back_to_buckets():
+    """Once a reading passes ENPHASE_LIVE_MAX_AGE it is ignored in favour of the bucket values.
+
+    Livestream readings are instantaneous and carry no usable timestamp of their own, so an old one
+    must not keep being published as though it were current.
+    """
+    from enphase import ENPHASE_LIVE_MAX_AGE_MINUTES
+
+    api = MockEnphaseAPI()
+    api.live_power["12345"] = {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100, "read_ts": FROZEN_NOW_TS - (ENPHASE_LIVE_MAX_AGE_MINUTES * 60 + 1)}
+    published = _publish_with_buckets(api)
+    assert published["pv"] == 1000.0  # bucket value, not the stale 4632.8
+    assert published["grid"] == -400.0
 
 
 def test_grid_power_is_positive_when_exporting():
@@ -750,7 +770,7 @@ def test_grid_power_is_positive_when_exporting():
     exporting, and the /today buckets give import - export), so both paths have to be flipped.
     """
     api = MockEnphaseAPI()
-    api.live_power["12345"] = {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100}
+    api.live_power["12345"] = {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100, "read_ts": FROZEN_NOW_TS - 60}
     published = _publish_with_buckets(api)
     assert published["grid"] == 2686.1  # exporting 2.7 kW -> positive
 
@@ -2120,7 +2140,8 @@ def run_enphase_api_tests(my_predbat):
     test_publish_prefers_the_measured_livestream_reading()
     test_publish_falls_back_to_buckets_without_a_livestream_reading()
     test_live_power_is_never_persisted()
-    test_failed_live_read_drops_the_previous_reading()
+    test_failed_live_read_keeps_the_recent_reading()
+    test_live_reading_older_than_the_window_falls_back_to_buckets()
     test_grid_power_is_positive_when_exporting()
     test_grid_power_from_buckets_is_positive_when_exporting()
     test_published_power_satisfies_the_predbat_energy_balance()

docs/components.md

@@ -586,7 +586,8 @@ Connects Predbat to the Enphase Enlighten cloud for monitoring and battery contr
 - Predbat controls the battery by writing Enphase schedules: charge windows become charge-from-grid (CFG) schedules with a target SOC, export windows become discharge-to-grid (DTG) schedules, freeze-export windows use restrict-battery-discharge (RBD) schedules, and the reserve is set through the battery profile. `automatic_config` requires both CFG and DTG support and fails configuration if either is missing
 - On a successful write, Predbat optimistically updates its local cache and moves on rather than waiting to re-read the cloud - the periodic schedule/profile re-read (every 30 minutes) corrects the cache later if a write didn't actually land
 - The PV, grid, battery and load power sensors come from the Enlighten livestream: once per cycle Predbat connects to Enphase's AWS IoT broker over MQTT, takes one measured reading and disconnects. These are separately metered channels, so they are instantaneous and the house load is a real measurement rather than a calculation
-- If the livestream is unavailable, all four fall back to the same 15-minute energy bucket of the cloud's intra-day data, so they still agree with each other and a power-flow display still balances. The cloud keeps back-filling a bucket for several minutes after it closes, so Predbat reads a bucket that has settled - which means the fallback values lag real time by roughly 15 to 30 minutes
+- A livestream reading stays in use for up to 15 minutes, so a single failed connection does not disturb the sensors. Past that they fall back rather than keep presenting an old measurement as current, and readings are never carried across a restart
+- In that fallback, all four come from the same 15-minute energy bucket of the cloud's intra-day data, so they still agree with each other and a power-flow display still balances. The cloud keeps back-filling a bucket for several minutes after it closes, so Predbat reads a bucket that has settled - which means the fallback values lag real time by roughly 15 to 30 minutes
 - In that fallback, load power is the energy-balance residual (PV + grid + battery), which is how the Enphase cloud derives its own consumption figure. Because it is a small difference between much larger numbers, it becomes unreliable while the battery is charging or discharging hard - it is clamped at zero so it can never show a negative house load, but treat it as indicative only during battery activity. The energy (`*_today`) sensors are unaffected either way and remain accurate
 - **Predbat owns the battery schedules**: it drives exactly one window per direction, so unless it is in read-only mode it deletes any other CFG/DTG/RBD schedule it finds on the site, including ones you created in the Enlighten app. Do not add your own battery schedules while Predbat is in write mode - the Enphase cloud rejects any overlapping schedule with an HTTP 409 conflict, which would stop Predbat from controlling the battery. Set Predbat to read-only mode if you want to manage schedules yourself
 - A window that is no longer needed is deleted rather than disabled, because the Enphase cloud ignores a request to disable a schedule (it reports success but keeps enforcing the window)