From a96d778bd162d91d950de28bdd9ffd608a035553 Mon Sep 17 00:00:00 2001 From: Trefor Southwell Date: Mon, 3 Aug 2026 15:39:05 +0200 Subject: [PATCH 1/7] feat(enphase): read measured power from the Enlighten livestream The instantaneous power sensors were derived from the /today 15-minute energy buckets, which cannot produce a usable house load: consumption is the residual of much larger terms, so on a site cycling 30 kWh a day through the battery to serve a 5 kWh house load it is unphysical about a fifth of the time. load_power was instead published from get_latest_power, which reports PRODUCTION, not consumption - it reads 0-1 W all night and tracks the PV ramp by day. The Enlighten app streams a protobuf DataMsg once a second over MQTT-on-WebSockets from AWS IoT, carrying separately METERED pv, storage, grid and load channels plus SOC. Predbat now takes one reading per cycle - connect, first message, disconnect, the same lifecycle the web app uses - rather than holding the stream open and re-authorising every 900s. Credentials are bootstrapped from /pv/aws_sigv4/livestream.json using the gateway serial, which /today already carries, so no extra discovery call is needed. AWS IoT's custom authorizer is fed through the MQTT CONNECT username: the WebSocket takes no query string and no password, because a browser cannot set custom headers on a WebSocket. Verified against 379 frames captured from a real session: the channels satisfy load = pv + grid + battery to 0.0 W on every frame, load reads 154-1979 W where PV reads 4452-4933 W, and the signs already match Predbat's convention. One of those frames is committed as a test fixture. The bucket-derived values remain the fallback for pv/grid/battery when the stream is unavailable, so a failure degrades rather than blanking the sensors; load is left empty in that case rather than published wrong. Co-Authored-By: Claude Opus 5 (1M context) --- .cspell/custom-dictionary-workspace.txt | 4 + apps/predbat/enphase.py | 180 +++++++++++++++++++++++- apps/predbat/enphase_livestream.proto | 119 ++++++++++++++++ apps/predbat/enphase_livestream_pb2.py | 60 ++++++++ apps/predbat/predbat.py | 2 +- apps/predbat/tests/test_enphase_api.py | 132 +++++++++++++++++ docs/components.md | 5 +- 7 files changed, 497 insertions(+), 5 deletions(-) create mode 100644 apps/predbat/enphase_livestream.proto create mode 100644 apps/predbat/enphase_livestream_pb2.py diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index e274dc57c..c83a3324b 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -25,6 +25,7 @@ armhf armv ASHP asyncio +authoriser autodocstring autoflake automations @@ -76,6 +77,7 @@ corruptplans cprofile creds crosscharge +customauthorizer customisation Customise cvalue @@ -138,6 +140,7 @@ energythroughput enho Enlighten enlm +enph enphase enphaseenergy Enpower @@ -368,6 +371,7 @@ predheat preseed preseeded prevs +protobuf psum pvbat pvenergytotal diff --git a/apps/predbat/enphase.py b/apps/predbat/enphase.py index 734d6ed93..f37ee3508 100644 --- a/apps/predbat/enphase.py +++ b/apps/predbat/enphase.py @@ -21,7 +21,9 @@ import base64 import json import random +import ssl import uuid +from urllib.parse import urlencode import aiohttp @@ -29,6 +31,22 @@ from mock_base import MockBase from predbat_metrics import record_api_call +try: + import enphase_livestream_pb2 as livestream_pb + + HAS_LIVESTREAM_PROTOBUF = True +except (ImportError, Exception): + livestream_pb = None + HAS_LIVESTREAM_PROTOBUF = False + +try: + import aiomqtt + + HAS_AIOMQTT = True +except (ImportError, Exception): + aiomqtt = None + HAS_AIOMQTT = False + # Defined locally (not imported from utils) - every cloud component defines its own # copy of this table rather than sharing one, matching the pattern used by fox.py. BASE_TIME = datetime.strptime("00:00", "%H:%M") @@ -49,8 +67,10 @@ # How many intra-day buckets to step back when deriving power from the /today energy arrays. # 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 +LIVESTREAM_BOOTSTRAP = "/pv/aws_sigv4/livestream.json" # returns the AWS IoT endpoint, topic and authorizer credentials -ENPHASE_CACHE_KEYS = ["sites", "battery_status", "battery_settings", "profile", "schedules", "site_settings", "today", "latest_power"] +ENPHASE_CACHE_KEYS = ["sites", "battery_status", "battery_settings", "profile", "schedules", "site_settings", "today", "latest_power", "live_power"] ENPHASE_CACHE_VERSION = 2 # Battery profiles accepted by the profile endpoint @@ -170,6 +190,55 @@ def enphase_time_to_ha(value): return text + ":00" +def gateway_serial(today): + """Return the gateway (Envoy) serial recorded by get_today, or None.""" + return (today or {}).get("serial") + + +def livestream_username(boot, site_id): + """Build the MQTT CONNECT username that AWS IoT's custom authorizer expects. + + The livestream WebSocket carries no query parameters and no password - a browser cannot set + custom headers on a WebSocket - so the authorizer name, the token and the token's signature all + travel in the username as a leading-'?' query string. Field order matches the Enlighten web app. + """ + return "?" + urlencode( + [ + ("x-amz-customauthorizer-name", boot.get("aws_authorizer", "")), + (boot.get("aws_token_key", "enph_token"), boot.get("aws_token_value", "")), + ("site-id", str(site_id)), + ("x-amz-customauthorizer-signature", boot.get("aws_digest", "")), + ("evse-count", "0"), + ("env", "prod"), + ] + ) + + +def decode_livestream_message(payload): + """Decode one livestream DataMsg into per-channel watts plus battery SOC. + + ``agg_p_mw`` is real power in milliwatts. The channels are measured, not derived, and satisfy + load = pv + grid + battery exactly. Signs already match Predbat's convention (grid negative when + exporting, battery positive when discharging). Returns None if the payload will not decode. + """ + if not HAS_LIVESTREAM_PROTOBUF or not payload: + return None + try: + message = livestream_pb.DataMsg() + message.ParseFromString(payload) + except Exception: + return None + meters = message.meters + watts = lambda channel: round(channel.agg_p_mw / 1000.0, 1) # noqa: E731 - milliwatts -> watts + return { + "pv": watts(meters.pv), + "battery": watts(meters.storage), + "grid": watts(meters.grid), + "load": watts(meters.load), + "soc": int(meters.soc), + } + + def _schedule_id_of(entry): """Return the cloud id of a schedule detail entry ('scheduleId', or 'id' on older shapes).""" return entry.get("scheduleId") or entry.get("id") @@ -257,6 +326,7 @@ def initialize(self, username, password, site_id=None, automatic=False, automati self.site_settings = {} self.today = {} # per-site today totals (Wh) + intra-day 15-minute buckets, from /today self.latest_power = {} + self.live_power = {} # measured pv/grid/battery/load watts + soc from the Enlighten livestream # Local (HA-side) schedule model, written by events, applied on write switch self.local_schedule = {} @@ -367,6 +437,8 @@ async def run(self, seconds, first): await self.get_today(site_id) if self._needs_refresh("latest_power", ENPHASE_REFRESH_POWER): await self.get_latest_power(site_id) + # Measured instantaneous power; falls back to the /today buckets if unavailable. + await self.get_live_power(site_id) self.sync_local_schedule_from_cloud(site_id) await self.publish_data(site_id) await self.publish_schedule_settings_ha(site_id) @@ -496,7 +568,8 @@ async def publish_data(self, site_id): app="enphase", ) - # Instantaneous power from the most recent settled intra-day 15-minute energy bucket of + # Fallback when the livestream is unavailable: instantaneous power from the most recent + # settled intra-day 15-minute energy bucket of # the /today arrays (Wh per interval -> average watts over that interval). This reads a # single bucket value per poll, so it is inherently stable within an interval and needs no # cross-poll delta tracking. @@ -520,6 +593,16 @@ async def publish_data(self, site_id): # house load would render nonsensically. load_today remains the trustworthy energy figure. load_power = max(0.0, round(pv_power + grid_power + battery_power, 1)) + # Prefer the livestream when we have one: those four channels are separately metered and + # 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: + pv_power = live.get("pv", pv_power) + grid_power = live.get("grid", grid_power) + battery_power = live.get("battery", battery_power) + load_power = live.get("load", load_power) + self.dashboard_item( f"{entity_base}_load_power", state=load_power, @@ -1091,6 +1174,8 @@ async def get_today(self, site_id): "interval_length": stat.get("interval_length"), # Site health: siteStatus is "normal"/"comm" (communication fault) etc., with a # human-readable status description when there is a problem (e.g. gateway not reporting). + # Gateway (Envoy) serial, needed to bootstrap the livestream - saves a separate call. + "serial": ((data.get("connectionDetails") or [{}])[0] or {}).get("serial_num"), "site_status": data.get("siteStatus"), "status_severity": status_details.get("statusSeverity"), "status_desc": status_details.get("statusDesc"), @@ -1099,6 +1184,77 @@ async def get_today(self, site_id): await self._save_cache("today", self.today) return self.today[site_id] + async def get_live_power(self, site_id): + """Fetch one instantaneous, measured power reading from the Enlighten livestream. + + The Enlighten app streams a protobuf `DataMsg` once a second over MQTT-on-WebSockets from + AWS IoT, carrying separately METERED pv/storage/grid/load channels plus SOC. That is the + only source of a real house-load figure: the /today energy buckets can only yield load as + the residual of much larger numbers, which is unusable while the battery cycles, and + get_latest_power reports production rather than consumption. + + Predbat only needs one sample per cycle, so this connects, takes the first message and + disconnects - the same lifecycle the web app uses - rather than holding the stream open and + re-authorising every `live_stream_duration` (900s). Returns the reading, or None on any + failure, leaving the caller to fall back to the bucket-derived values. + """ + if not (HAS_AIOMQTT and HAS_LIVESTREAM_PROTOBUF): + return None + serial = gateway_serial(self.today.get(site_id, {})) + if not serial: + return None + boot = await self.request_json("GET", LIVESTREAM_BOOTSTRAP, params={"serial_num": serial}) + if not boot or not boot.get("aws_iot_endpoint") or not boot.get("live_stream_topic"): + return None + reading = await self._read_livestream(site_id, boot, serial) + if reading: + self.live_power[site_id] = reading + await self._save_cache("live_power", self.live_power) + return reading + + async def _read_livestream(self, site_id, boot, serial): + """Connect to AWS IoT, take the first livestream message for a site, then disconnect. + + Credentials ride in the MQTT CONNECT username (see livestream_username) because the + WebSocket carries no query string and no password. Any failure is logged and swallowed - + the livestream is an enhancement, never a reason to fail a cycle. + """ + timeout = safe_float(boot.get("timeout"), ENPHASE_LIVESTREAM_TIMEOUT) or ENPHASE_LIVESTREAM_TIMEOUT + topic = boot.get("live_stream_topic") + + async def consume(): + """Subscribe and return the first decodable reading.""" + async with aiomqtt.Client( + hostname=boot["aws_iot_endpoint"], + port=443, + transport="websockets", + websocket_path="/mqtt", + tls_context=ssl.create_default_context(), + identifier=f"em-paho-mqtt-{random.randint(10000, 99999)}-{serial}", + username=livestream_username(boot, site_id), + clean_session=True, + keepalive=60, + ) as client: + await client.subscribe(topic, qos=0) + async for message in client.messages: + reading = decode_livestream_message(bytes(message.payload)) + if reading: + return reading + return None + + try: + reading = await asyncio.wait_for(consume(), timeout=timeout) + except asyncio.TimeoutError: + self.log(f"Warn: Enphase: Livestream timed out after {timeout}s for site {site_id}") + record_api_call("enphase", False, "livestream_timeout") + return None + except Exception as error: + self.log(f"Warn: Enphase: Livestream failed for site {site_id}: {error}") + record_api_call("enphase", False, "livestream_error") + return None + record_api_call("enphase", True) + return reading + async def get_latest_power(self, site_id): """Fetch and normalise the latest instantaneous power reading for a site.""" data = await self.request_json("GET", f"/app-api/{site_id}/get_latest_power") @@ -1602,6 +1758,26 @@ async def test_enphase_api(username, password, site_id): # pragma: no cover if values: print(f" {channel}: len={len(values)} last5={values[-5:]}") + # Livestream: prove the connect -> read one message -> disconnect cycle works against the + # real account, and cross-check it against the bucket-derived values it replaces. + print(f"\ngateway serial: {gateway_serial(today)}") + print(f"protobuf available: {HAS_LIVESTREAM_PROTOBUF} aiomqtt available: {HAS_AIOMQTT}") + for attempt in range(1, 4): + started = datetime.now(timezone.utc) + reading = await api.get_live_power(sid) + elapsed = (datetime.now(timezone.utc) - started).total_seconds() + if not reading: + print(f" livestream attempt {attempt}: FAILED after {elapsed:.1f}s") + continue + balance = reading["pv"] + reading["grid"] + reading["battery"] - reading["load"] + print(f" livestream attempt {attempt} ({elapsed:.1f}s): pv={reading['pv']}W grid={reading['grid']}W battery={reading['battery']}W load={reading['load']}W soc={reading['soc']}%") + print(f" energy balance (pv+grid+battery-load) = {balance:.1f} W <- expect ~0") + arrays = today.get("arrays") or {} + now_ts = datetime.now(timezone.utc).timestamp() + bucket = {channel: interval_power(arrays.get(channel, []), today.get("start_time"), today.get("interval_length"), now_ts) for channel in ("production", "import", "export", "charge", "discharge")} + print(f" bucket-derived for comparison: pv={bucket['production']}W grid={bucket['import'] - bucket['export']}W battery={bucket['discharge'] - bucket['charge']}W") + print(" (buckets are a 15-minute average and lag; large differences are expected)") + print("\nDone") diff --git a/apps/predbat/enphase_livestream.proto b/apps/predbat/enphase_livestream.proto new file mode 100644 index 000000000..d269178d2 --- /dev/null +++ b/apps/predbat/enphase_livestream.proto @@ -0,0 +1,119 @@ +// Enphase Enlighten livestream "Data Channel" message. +// +// Schema published by Enphase at +// https://assets-enlighten.enphaseenergy.com/mobile/static/proto/DataMsg.proto and +// .../MeterSummaryData.proto, combined here into one file (their DataMsg imports +// HemsStreamMessage.proto but does not reference it, so that import is dropped). +// +// Regenerate enphase_livestream_pb2.py after editing: +// python -m grpc_tools.protoc -I apps/predbat --python_out=apps/predbat \ +// apps/predbat/enphase_livestream.proto +syntax = "proto3"; + +enum MeterSumGridState { + OPER_RELAY_UNKNOWN = 0; + OPER_RELAY_OPEN = 1; + OPER_RELAY_CLOSED = 2; + OPER_RELAY_OFFGRID_AC_GRID_PRESENT = 3; + OPER_RELAY_OFFGRID_READY_FOR_RESYNC_CMD = 4; + OPER_RELAY_WAITING_TO_INITIALIZE_ON_GRID = 5; + OPER_RELAY_GEN_OPEN = 6; + OPER_RELAY_GEN_CLOSED = 7; + OPER_RELAY_GEN_STARTUP = 8; + OPER_RELAY_GEN_SYNC_READY = 9; + OPER_RELAY_GEN_AC_STABLE = 10; + OPER_RELAY_GEN_AC_UNSTABLE = 11; +} + +enum BattMode { + BATT_MODE_FULL_BACKUP = 0; + BATT_MODE_SELF_CONS = 1; + BATT_MODE_SAVINGS = 2; +} + +enum DryContactId { + NC1 = 0; + NC2 = 1; + NO1 = 2; + NO2 = 3; +} + +enum DryContactRelayState { + DC_RELAY_STATE_INVALID = 0; + DC_RELAY_OFF = 1; + DC_RELAY_ON = 2; +} + +enum MeterType { + METER_TYPE_NONE = 0; + METER_TYPE_PV = 1; + METER_TYPE_STORAGE = 2; +} + +// One measured channel. agg_p_mw is real power in milliwatts (divide by 1000 for W); +// agg_s_mva is apparent power in milli-VA. +message MeterChannel { + int64 agg_p_mw = 1; + int64 agg_s_mva = 2; + repeated int64 agg_p_ph_mw = 3; + repeated int64 agg_s_ph_mva = 4; + optional string device_sn = 5; +} + +message AggMeterChannel { + int64 agg_p_mw = 1; + int64 agg_s_mva = 2; + repeated int64 agg_p_ph_mw = 3; + repeated int64 agg_s_ph_mva = 4; + optional MeterType type = 5; + repeated MeterChannel channels = 6; +} + +message MeterSummaryData { + MeterChannel pv = 1; + MeterChannel storage = 2; + MeterChannel grid = 3; + MeterChannel load = 4; + MeterSumGridState grid_relay = 5; + int32 soc = 6; + MeterChannel generator = 7; + MeterSumGridState gen_relay = 8; + uint32 phase_count = 9; + bool is_split_phase = 10; + repeated AggMeterChannel meter_channel = 14; +} + +message DryContactStatus { + DryContactId id = 1; + DryContactRelayState state = 2; +} + +message DryContactName { + DryContactId id = 1; + string load_name = 2; +} + +message LoadStatus { + string id = 1; + string relay_status = 2; + float power = 3; +} + +message PowerMatchStatus { + bool status = 1; + uint32 totalPCUCount = 2; + uint32 runningPCUCount = 3; + bool isSupported = 4; +} + +message DataMsg { + int32 protocol_ver = 1; + uint64 timestamp = 2; + MeterSummaryData meters = 3; + BattMode batt_mode = 4; + int32 backup_soc = 5; + repeated DryContactStatus dry_contact_relay_status = 6; + repeated DryContactName dry_contact_relay_name = 7; + repeated LoadStatus load_status = 8; + PowerMatchStatus power_match_status = 9; +} diff --git a/apps/predbat/enphase_livestream_pb2.py b/apps/predbat/enphase_livestream_pb2.py new file mode 100644 index 000000000..833fbdea2 --- /dev/null +++ b/apps/predbat/enphase_livestream_pb2.py @@ -0,0 +1,60 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE +# source: enphase_livestream.proto +# Protobuf Python Version: 7.35.1 +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 7, + 35, + 1, + '', + 'enphase_livestream.proto' +) +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x18\x65nphase_livestream.proto\"\x84\x01\n\x0cMeterChannel\x12\x10\n\x08\x61gg_p_mw\x18\x01 \x01(\x03\x12\x11\n\tagg_s_mva\x18\x02 \x01(\x03\x12\x13\n\x0b\x61gg_p_ph_mw\x18\x03 \x03(\x03\x12\x14\n\x0c\x61gg_s_ph_mva\x18\x04 \x03(\x03\x12\x16\n\tdevice_sn\x18\x05 \x01(\tH\x00\x88\x01\x01\x42\x0c\n\n_device_sn\"\xaa\x01\n\x0f\x41ggMeterChannel\x12\x10\n\x08\x61gg_p_mw\x18\x01 \x01(\x03\x12\x11\n\tagg_s_mva\x18\x02 \x01(\x03\x12\x13\n\x0b\x61gg_p_ph_mw\x18\x03 \x03(\x03\x12\x14\n\x0c\x61gg_s_ph_mva\x18\x04 \x03(\x03\x12\x1d\n\x04type\x18\x05 \x01(\x0e\x32\n.MeterTypeH\x00\x88\x01\x01\x12\x1f\n\x08\x63hannels\x18\x06 \x03(\x0b\x32\r.MeterChannelB\x07\n\x05_type\"\xdb\x02\n\x10MeterSummaryData\x12\x19\n\x02pv\x18\x01 \x01(\x0b\x32\r.MeterChannel\x12\x1e\n\x07storage\x18\x02 \x01(\x0b\x32\r.MeterChannel\x12\x1b\n\x04grid\x18\x03 \x01(\x0b\x32\r.MeterChannel\x12\x1b\n\x04load\x18\x04 \x01(\x0b\x32\r.MeterChannel\x12&\n\ngrid_relay\x18\x05 \x01(\x0e\x32\x12.MeterSumGridState\x12\x0b\n\x03soc\x18\x06 \x01(\x05\x12 \n\tgenerator\x18\x07 \x01(\x0b\x32\r.MeterChannel\x12%\n\tgen_relay\x18\x08 \x01(\x0e\x32\x12.MeterSumGridState\x12\x13\n\x0bphase_count\x18\t \x01(\r\x12\x16\n\x0eis_split_phase\x18\n \x01(\x08\x12\'\n\rmeter_channel\x18\x0e \x03(\x0b\x32\x10.AggMeterChannel\"S\n\x10\x44ryContactStatus\x12\x19\n\x02id\x18\x01 \x01(\x0e\x32\r.DryContactId\x12$\n\x05state\x18\x02 \x01(\x0e\x32\x15.DryContactRelayState\">\n\x0e\x44ryContactName\x12\x19\n\x02id\x18\x01 \x01(\x0e\x32\r.DryContactId\x12\x11\n\tload_name\x18\x02 \x01(\t\"=\n\nLoadStatus\x12\n\n\x02id\x18\x01 \x01(\t\x12\x14\n\x0crelay_status\x18\x02 \x01(\t\x12\r\n\x05power\x18\x03 \x01(\x02\"g\n\x10PowerMatchStatus\x12\x0e\n\x06status\x18\x01 \x01(\x08\x12\x15\n\rtotalPCUCount\x18\x02 \x01(\r\x12\x17\n\x0frunningPCUCount\x18\x03 \x01(\r\x12\x13\n\x0bisSupported\x18\x04 \x01(\x08\"\xbe\x02\n\x07\x44\x61taMsg\x12\x14\n\x0cprotocol_ver\x18\x01 \x01(\x05\x12\x11\n\ttimestamp\x18\x02 \x01(\x04\x12!\n\x06meters\x18\x03 \x01(\x0b\x32\x11.MeterSummaryData\x12\x1c\n\tbatt_mode\x18\x04 \x01(\x0e\x32\t.BattMode\x12\x12\n\nbackup_soc\x18\x05 \x01(\x05\x12\x33\n\x18\x64ry_contact_relay_status\x18\x06 \x03(\x0b\x32\x11.DryContactStatus\x12/\n\x16\x64ry_contact_relay_name\x18\x07 \x03(\x0b\x32\x0f.DryContactName\x12 \n\x0bload_status\x18\x08 \x03(\x0b\x32\x0b.LoadStatus\x12-\n\x12power_match_status\x18\t \x01(\x0b\x32\x11.PowerMatchStatus*\x87\x03\n\x11MeterSumGridState\x12\x16\n\x12OPER_RELAY_UNKNOWN\x10\x00\x12\x13\n\x0fOPER_RELAY_OPEN\x10\x01\x12\x15\n\x11OPER_RELAY_CLOSED\x10\x02\x12&\n\"OPER_RELAY_OFFGRID_AC_GRID_PRESENT\x10\x03\x12+\n\'OPER_RELAY_OFFGRID_READY_FOR_RESYNC_CMD\x10\x04\x12,\n(OPER_RELAY_WAITING_TO_INITIALIZE_ON_GRID\x10\x05\x12\x17\n\x13OPER_RELAY_GEN_OPEN\x10\x06\x12\x19\n\x15OPER_RELAY_GEN_CLOSED\x10\x07\x12\x1a\n\x16OPER_RELAY_GEN_STARTUP\x10\x08\x12\x1d\n\x19OPER_RELAY_GEN_SYNC_READY\x10\t\x12\x1c\n\x18OPER_RELAY_GEN_AC_STABLE\x10\n\x12\x1e\n\x1aOPER_RELAY_GEN_AC_UNSTABLE\x10\x0b*U\n\x08\x42\x61ttMode\x12\x19\n\x15\x42\x41TT_MODE_FULL_BACKUP\x10\x00\x12\x17\n\x13\x42\x41TT_MODE_SELF_CONS\x10\x01\x12\x15\n\x11\x42\x41TT_MODE_SAVINGS\x10\x02*2\n\x0c\x44ryContactId\x12\x07\n\x03NC1\x10\x00\x12\x07\n\x03NC2\x10\x01\x12\x07\n\x03NO1\x10\x02\x12\x07\n\x03NO2\x10\x03*U\n\x14\x44ryContactRelayState\x12\x1a\n\x16\x44\x43_RELAY_STATE_INVALID\x10\x00\x12\x10\n\x0c\x44\x43_RELAY_OFF\x10\x01\x12\x0f\n\x0b\x44\x43_RELAY_ON\x10\x02*K\n\tMeterType\x12\x13\n\x0fMETER_TYPE_NONE\x10\x00\x12\x11\n\rMETER_TYPE_PV\x10\x01\x12\x16\n\x12METER_TYPE_STORAGE\x10\x02\x62\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'enphase_livestream_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + DESCRIPTOR._loaded_options = None + _globals['_METERSUMGRIDSTATE']._serialized_start=1325 + _globals['_METERSUMGRIDSTATE']._serialized_end=1716 + _globals['_BATTMODE']._serialized_start=1718 + _globals['_BATTMODE']._serialized_end=1803 + _globals['_DRYCONTACTID']._serialized_start=1805 + _globals['_DRYCONTACTID']._serialized_end=1855 + _globals['_DRYCONTACTRELAYSTATE']._serialized_start=1857 + _globals['_DRYCONTACTRELAYSTATE']._serialized_end=1942 + _globals['_METERTYPE']._serialized_start=1944 + _globals['_METERTYPE']._serialized_end=2019 + _globals['_METERCHANNEL']._serialized_start=29 + _globals['_METERCHANNEL']._serialized_end=161 + _globals['_AGGMETERCHANNEL']._serialized_start=164 + _globals['_AGGMETERCHANNEL']._serialized_end=334 + _globals['_METERSUMMARYDATA']._serialized_start=337 + _globals['_METERSUMMARYDATA']._serialized_end=684 + _globals['_DRYCONTACTSTATUS']._serialized_start=686 + _globals['_DRYCONTACTSTATUS']._serialized_end=769 + _globals['_DRYCONTACTNAME']._serialized_start=771 + _globals['_DRYCONTACTNAME']._serialized_end=833 + _globals['_LOADSTATUS']._serialized_start=835 + _globals['_LOADSTATUS']._serialized_end=896 + _globals['_POWERMATCHSTATUS']._serialized_start=898 + _globals['_POWERMATCHSTATUS']._serialized_end=1001 + _globals['_DATAMSG']._serialized_start=1004 + _globals['_DATAMSG']._serialized_end=1322 +# @@protoc_insertion_point(module_scope) diff --git a/apps/predbat/predbat.py b/apps/predbat/predbat.py index 76a977639..5b3ee53c2 100644 --- a/apps/predbat/predbat.py +++ b/apps/predbat/predbat.py @@ -35,7 +35,7 @@ import pytz import asyncio -THIS_VERSION = "v8.47.6" +THIS_VERSION = "v8.47.7" from download import predbat_update_move, predbat_update_download, check_install, DEFAULT_PREDBAT_REPOSITORY from const import MINUTE_WATT diff --git a/apps/predbat/tests/test_enphase_api.py b/apps/predbat/tests/test_enphase_api.py index 144b3a607..02acd891a 100644 --- a/apps/predbat/tests/test_enphase_api.py +++ b/apps/predbat/tests/test_enphase_api.py @@ -601,6 +601,130 @@ def test_today_channel_kwh(): assert today_channel_kwh({"totals": {"charge": 9000, "solar_battery": 1000}}, "charge") == 9.0 +# A real DataMsg captured from the Enlighten livestream topic (site exporting on a sunny afternoon): +# pv 4632.8 W, storage 32.0 W, grid -2686.1 W, load 1978.7 W, soc 100%. +LIVESTREAM_FRAME = base64.b64decode( + "CAEQwIQ9GogBChYI7+GaAhDv4ZoCGgTv4ZoCIgTv4ZoCEhIIgPoBEPP6ARoDgPoBIgPz+gEaLgjZhtz+//////8BELLiwf7//////wEaCtmG3P7//////wEiCrLiwf7//////wEiEgjI4ngQlL9eGgPI4ngiA5S/XigCMGQ6BhoBACIBAEAGSAFaBghkEKCcASABKAUyAhACMgQIARACMgQIAhABMgQIAxAB" +) + + +def test_decode_livestream_message(): + """A livestream DataMsg decodes to watts per channel plus SOC. + + agg_p_mw is milliwatts. Signs follow Predbat's convention: grid negative when exporting, + battery positive when discharging. + """ + from enphase import decode_livestream_message + + reading = decode_livestream_message(LIVESTREAM_FRAME) + assert reading == {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100} + + +def test_decode_livestream_message_balances(): + """The decoded channels satisfy the energy balance load = pv + grid + battery.""" + from enphase import decode_livestream_message + + r = decode_livestream_message(LIVESTREAM_FRAME) + assert abs((r["pv"] + r["grid"] + r["battery"]) - r["load"]) < 0.05 + + +def test_decode_livestream_message_rejects_rubbish(): + """A payload that is not a DataMsg returns None rather than raising.""" + from enphase import decode_livestream_message + + assert decode_livestream_message(b"\xff\xff\xff\xff not protobuf") is None + + +def test_livestream_username_carries_the_authorizer_credentials(): + """The MQTT username is the query-string blob AWS IoT's custom authorizer expects. + + The WebSocket URL carries no query parameters - the browser cannot set custom headers on a + WebSocket - so the authorizer name, token and signature all travel in the CONNECT username. + """ + from enphase import livestream_username + + boot = { + "aws_authorizer": "aws-lambda-authoriser-prod", + "aws_token_key": "enph_token", + "aws_token_value": "tok123", + "aws_digest": "sig+with/reserved=chars", + } + user = livestream_username(boot, "5667604") + assert user.startswith("?") + assert "x-amz-customauthorizer-name=aws-lambda-authoriser-prod" in user + assert "enph_token=tok123" in user + assert "site-id=5667604" in user + # The digest is base64 and must be percent-encoded, not passed raw + from urllib.parse import quote_plus + + assert f"x-amz-customauthorizer-signature={quote_plus(boot['aws_digest'])}" in user + assert boot["aws_digest"] not in user + + +def test_gateway_serial_read_from_today(): + """The livestream bootstrap needs the gateway serial, which /today already carries.""" + from enphase import gateway_serial + + assert gateway_serial({"serial": "122530006866"}) == "122530006866" + assert gateway_serial({}) is None + + +def _publish_with_buckets(api): + """Publish a site whose /today buckets give pv 1000 W, grid 400 W, battery -800 W.""" + start = 1783724400 + + def bucket(value): + """Return a 96-slot Wh array with the bucket read by publish_data set to value.""" + out = [0] * 96 + out[80] = value + return out + + api.today["12345"] = { + "totals": {}, + "arrays": {"production": bucket(250), "import": bucket(100), "export": bucket(0), "charge": bucket(200), "discharge": bucket(0)}, + "start_time": start, + "interval_length": 900, + } + import enphase as enphase_module + + original = enphase_module.datetime + + class _Fixed(original): + @classmethod + def now(cls, tz=None): + """Return a time inside the interval after the bucket that publish_data reads.""" + return original.fromtimestamp(start + int(81.5 * 900), tz) + + enphase_module.datetime = _Fixed + try: + run_async(api.publish_data("12345")) + finally: + enphase_module.datetime = original + base = "sensor.predbat_enphase_12345" + return {name: api.dashboard_items[f"{base}_{name}_power"]["state"] for name in ("pv", "grid", "battery", "load")} + + +def test_publish_prefers_the_measured_livestream_reading(): + """When a livestream reading is available the power sensors use it, not the energy buckets. + + The livestream channels are metered, instantaneous and include a real house load, so they beat + 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} + published = _publish_with_buckets(api) + assert published == {"pv": 4632.8, "grid": -2686.1, "battery": 32.0, "load": 1978.7} + + +def test_publish_falls_back_to_buckets_without_a_livestream_reading(): + """With no livestream reading the bucket-derived values are still published.""" + api = MockEnphaseAPI() + published = _publish_with_buckets(api) + assert published["pv"] == 1000.0 + assert published["grid"] == 400.0 + assert published["battery"] == -800.0 + + def test_interval_power(): """interval_power converts a settled 15-minute Wh bucket into watts. @@ -956,6 +1080,7 @@ def _bucket(value): "start_time": start, "interval_length": 900, } + # No livestream reading here, so the power sensors fall back to the /today buckets. # Freeze "now" so interval_power selects settled bucket 80 (current 82, just-closed 81). import enphase as enphase_module @@ -1880,6 +2005,13 @@ def run_enphase_api_tests(my_predbat): test_load_power_is_derived_from_the_other_channels() test_load_power_never_goes_negative() test_power_sensors_ignore_the_unsettled_bucket() + test_decode_livestream_message() + test_decode_livestream_message_balances() + test_decode_livestream_message_rejects_rubbish() + test_livestream_username_carries_the_authorizer_credentials() + test_gateway_serial_read_from_today() + test_publish_prefers_the_measured_livestream_reading() + test_publish_falls_back_to_buckets_without_a_livestream_reading() test_get_schedules_parses_families() test_automatic_config() test_automatic_config_no_dtg_raises() diff --git a/docs/components.md b/docs/components.md index bf4762024..986188d25 100644 --- a/docs/components.md +++ b/docs/components.md @@ -585,8 +585,9 @@ Connects Predbat to the Enphase Enlighten cloud for monitoring and battery contr - Accounts with multi-factor authentication (MFA) enabled are **not supported** - disable MFA on the Enphase account before use - 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 are all derived from the same 15-minute energy bucket of the cloud's intra-day data, so they agree with each other and a power-flow display balances. The cloud keeps back-filling a bucket for several minutes after it closes, so Predbat reads a bucket that has settled - which means these sensors lag real time by roughly 15 to 30 minutes. Only the energy (`*_today`) sensors are real-time-ish -- 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. `load_today` is unaffected and remains accurate +- 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 +- 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) - Repeated login failures back off automatically to protect the Enphase account from lockout: a 5 minute cooldown after each rejection, rising to a 24 hour suspension after 3 consecutive rejections From b47d23c1147ce5a53ee3085ca5e647d3f59366a2 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci-lite[bot]" <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com> Date: Mon, 3 Aug 2026 13:41:44 +0000 Subject: [PATCH 2/7] [pre-commit.ci lite] apply automatic fixes --- apps/predbat/enphase_livestream_pb2.py | 72 ++++++++++++-------------- 1 file changed, 33 insertions(+), 39 deletions(-) diff --git a/apps/predbat/enphase_livestream_pb2.py b/apps/predbat/enphase_livestream_pb2.py index 833fbdea2..4ecede7fb 100644 --- a/apps/predbat/enphase_livestream_pb2.py +++ b/apps/predbat/enphase_livestream_pb2.py @@ -9,52 +9,46 @@ from google.protobuf import runtime_version as _runtime_version from google.protobuf import symbol_database as _symbol_database from google.protobuf.internal import builder as _builder -_runtime_version.ValidateProtobufRuntimeVersion( - _runtime_version.Domain.PUBLIC, - 7, - 35, - 1, - '', - 'enphase_livestream.proto' -) + +_runtime_version.ValidateProtobufRuntimeVersion(_runtime_version.Domain.PUBLIC, 7, 35, 1, "", "enphase_livestream.proto") # @@protoc_insertion_point(imports) _sym_db = _symbol_database.Default() - - -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x18\x65nphase_livestream.proto\"\x84\x01\n\x0cMeterChannel\x12\x10\n\x08\x61gg_p_mw\x18\x01 \x01(\x03\x12\x11\n\tagg_s_mva\x18\x02 \x01(\x03\x12\x13\n\x0b\x61gg_p_ph_mw\x18\x03 \x03(\x03\x12\x14\n\x0c\x61gg_s_ph_mva\x18\x04 \x03(\x03\x12\x16\n\tdevice_sn\x18\x05 \x01(\tH\x00\x88\x01\x01\x42\x0c\n\n_device_sn\"\xaa\x01\n\x0f\x41ggMeterChannel\x12\x10\n\x08\x61gg_p_mw\x18\x01 \x01(\x03\x12\x11\n\tagg_s_mva\x18\x02 \x01(\x03\x12\x13\n\x0b\x61gg_p_ph_mw\x18\x03 \x03(\x03\x12\x14\n\x0c\x61gg_s_ph_mva\x18\x04 \x03(\x03\x12\x1d\n\x04type\x18\x05 \x01(\x0e\x32\n.MeterTypeH\x00\x88\x01\x01\x12\x1f\n\x08\x63hannels\x18\x06 \x03(\x0b\x32\r.MeterChannelB\x07\n\x05_type\"\xdb\x02\n\x10MeterSummaryData\x12\x19\n\x02pv\x18\x01 \x01(\x0b\x32\r.MeterChannel\x12\x1e\n\x07storage\x18\x02 \x01(\x0b\x32\r.MeterChannel\x12\x1b\n\x04grid\x18\x03 \x01(\x0b\x32\r.MeterChannel\x12\x1b\n\x04load\x18\x04 \x01(\x0b\x32\r.MeterChannel\x12&\n\ngrid_relay\x18\x05 \x01(\x0e\x32\x12.MeterSumGridState\x12\x0b\n\x03soc\x18\x06 \x01(\x05\x12 \n\tgenerator\x18\x07 \x01(\x0b\x32\r.MeterChannel\x12%\n\tgen_relay\x18\x08 \x01(\x0e\x32\x12.MeterSumGridState\x12\x13\n\x0bphase_count\x18\t \x01(\r\x12\x16\n\x0eis_split_phase\x18\n \x01(\x08\x12\'\n\rmeter_channel\x18\x0e \x03(\x0b\x32\x10.AggMeterChannel\"S\n\x10\x44ryContactStatus\x12\x19\n\x02id\x18\x01 \x01(\x0e\x32\r.DryContactId\x12$\n\x05state\x18\x02 \x01(\x0e\x32\x15.DryContactRelayState\">\n\x0e\x44ryContactName\x12\x19\n\x02id\x18\x01 \x01(\x0e\x32\r.DryContactId\x12\x11\n\tload_name\x18\x02 \x01(\t\"=\n\nLoadStatus\x12\n\n\x02id\x18\x01 \x01(\t\x12\x14\n\x0crelay_status\x18\x02 \x01(\t\x12\r\n\x05power\x18\x03 \x01(\x02\"g\n\x10PowerMatchStatus\x12\x0e\n\x06status\x18\x01 \x01(\x08\x12\x15\n\rtotalPCUCount\x18\x02 \x01(\r\x12\x17\n\x0frunningPCUCount\x18\x03 \x01(\r\x12\x13\n\x0bisSupported\x18\x04 \x01(\x08\"\xbe\x02\n\x07\x44\x61taMsg\x12\x14\n\x0cprotocol_ver\x18\x01 \x01(\x05\x12\x11\n\ttimestamp\x18\x02 \x01(\x04\x12!\n\x06meters\x18\x03 \x01(\x0b\x32\x11.MeterSummaryData\x12\x1c\n\tbatt_mode\x18\x04 \x01(\x0e\x32\t.BattMode\x12\x12\n\nbackup_soc\x18\x05 \x01(\x05\x12\x33\n\x18\x64ry_contact_relay_status\x18\x06 \x03(\x0b\x32\x11.DryContactStatus\x12/\n\x16\x64ry_contact_relay_name\x18\x07 \x03(\x0b\x32\x0f.DryContactName\x12 \n\x0bload_status\x18\x08 \x03(\x0b\x32\x0b.LoadStatus\x12-\n\x12power_match_status\x18\t \x01(\x0b\x32\x11.PowerMatchStatus*\x87\x03\n\x11MeterSumGridState\x12\x16\n\x12OPER_RELAY_UNKNOWN\x10\x00\x12\x13\n\x0fOPER_RELAY_OPEN\x10\x01\x12\x15\n\x11OPER_RELAY_CLOSED\x10\x02\x12&\n\"OPER_RELAY_OFFGRID_AC_GRID_PRESENT\x10\x03\x12+\n\'OPER_RELAY_OFFGRID_READY_FOR_RESYNC_CMD\x10\x04\x12,\n(OPER_RELAY_WAITING_TO_INITIALIZE_ON_GRID\x10\x05\x12\x17\n\x13OPER_RELAY_GEN_OPEN\x10\x06\x12\x19\n\x15OPER_RELAY_GEN_CLOSED\x10\x07\x12\x1a\n\x16OPER_RELAY_GEN_STARTUP\x10\x08\x12\x1d\n\x19OPER_RELAY_GEN_SYNC_READY\x10\t\x12\x1c\n\x18OPER_RELAY_GEN_AC_STABLE\x10\n\x12\x1e\n\x1aOPER_RELAY_GEN_AC_UNSTABLE\x10\x0b*U\n\x08\x42\x61ttMode\x12\x19\n\x15\x42\x41TT_MODE_FULL_BACKUP\x10\x00\x12\x17\n\x13\x42\x41TT_MODE_SELF_CONS\x10\x01\x12\x15\n\x11\x42\x41TT_MODE_SAVINGS\x10\x02*2\n\x0c\x44ryContactId\x12\x07\n\x03NC1\x10\x00\x12\x07\n\x03NC2\x10\x01\x12\x07\n\x03NO1\x10\x02\x12\x07\n\x03NO2\x10\x03*U\n\x14\x44ryContactRelayState\x12\x1a\n\x16\x44\x43_RELAY_STATE_INVALID\x10\x00\x12\x10\n\x0c\x44\x43_RELAY_OFF\x10\x01\x12\x0f\n\x0b\x44\x43_RELAY_ON\x10\x02*K\n\tMeterType\x12\x13\n\x0fMETER_TYPE_NONE\x10\x00\x12\x11\n\rMETER_TYPE_PV\x10\x01\x12\x16\n\x12METER_TYPE_STORAGE\x10\x02\x62\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n\x18\x65nphase_livestream.proto"\x84\x01\n\x0cMeterChannel\x12\x10\n\x08\x61gg_p_mw\x18\x01 \x01(\x03\x12\x11\n\tagg_s_mva\x18\x02 \x01(\x03\x12\x13\n\x0b\x61gg_p_ph_mw\x18\x03 \x03(\x03\x12\x14\n\x0c\x61gg_s_ph_mva\x18\x04 \x03(\x03\x12\x16\n\tdevice_sn\x18\x05 \x01(\tH\x00\x88\x01\x01\x42\x0c\n\n_device_sn"\xaa\x01\n\x0f\x41ggMeterChannel\x12\x10\n\x08\x61gg_p_mw\x18\x01 \x01(\x03\x12\x11\n\tagg_s_mva\x18\x02 \x01(\x03\x12\x13\n\x0b\x61gg_p_ph_mw\x18\x03 \x03(\x03\x12\x14\n\x0c\x61gg_s_ph_mva\x18\x04 \x03(\x03\x12\x1d\n\x04type\x18\x05 \x01(\x0e\x32\n.MeterTypeH\x00\x88\x01\x01\x12\x1f\n\x08\x63hannels\x18\x06 \x03(\x0b\x32\r.MeterChannelB\x07\n\x05_type"\xdb\x02\n\x10MeterSummaryData\x12\x19\n\x02pv\x18\x01 \x01(\x0b\x32\r.MeterChannel\x12\x1e\n\x07storage\x18\x02 \x01(\x0b\x32\r.MeterChannel\x12\x1b\n\x04grid\x18\x03 \x01(\x0b\x32\r.MeterChannel\x12\x1b\n\x04load\x18\x04 \x01(\x0b\x32\r.MeterChannel\x12&\n\ngrid_relay\x18\x05 \x01(\x0e\x32\x12.MeterSumGridState\x12\x0b\n\x03soc\x18\x06 \x01(\x05\x12 \n\tgenerator\x18\x07 \x01(\x0b\x32\r.MeterChannel\x12%\n\tgen_relay\x18\x08 \x01(\x0e\x32\x12.MeterSumGridState\x12\x13\n\x0bphase_count\x18\t \x01(\r\x12\x16\n\x0eis_split_phase\x18\n \x01(\x08\x12\'\n\rmeter_channel\x18\x0e \x03(\x0b\x32\x10.AggMeterChannel"S\n\x10\x44ryContactStatus\x12\x19\n\x02id\x18\x01 \x01(\x0e\x32\r.DryContactId\x12$\n\x05state\x18\x02 \x01(\x0e\x32\x15.DryContactRelayState">\n\x0e\x44ryContactName\x12\x19\n\x02id\x18\x01 \x01(\x0e\x32\r.DryContactId\x12\x11\n\tload_name\x18\x02 \x01(\t"=\n\nLoadStatus\x12\n\n\x02id\x18\x01 \x01(\t\x12\x14\n\x0crelay_status\x18\x02 \x01(\t\x12\r\n\x05power\x18\x03 \x01(\x02"g\n\x10PowerMatchStatus\x12\x0e\n\x06status\x18\x01 \x01(\x08\x12\x15\n\rtotalPCUCount\x18\x02 \x01(\r\x12\x17\n\x0frunningPCUCount\x18\x03 \x01(\r\x12\x13\n\x0bisSupported\x18\x04 \x01(\x08"\xbe\x02\n\x07\x44\x61taMsg\x12\x14\n\x0cprotocol_ver\x18\x01 \x01(\x05\x12\x11\n\ttimestamp\x18\x02 \x01(\x04\x12!\n\x06meters\x18\x03 \x01(\x0b\x32\x11.MeterSummaryData\x12\x1c\n\tbatt_mode\x18\x04 \x01(\x0e\x32\t.BattMode\x12\x12\n\nbackup_soc\x18\x05 \x01(\x05\x12\x33\n\x18\x64ry_contact_relay_status\x18\x06 \x03(\x0b\x32\x11.DryContactStatus\x12/\n\x16\x64ry_contact_relay_name\x18\x07 \x03(\x0b\x32\x0f.DryContactName\x12 \n\x0bload_status\x18\x08 \x03(\x0b\x32\x0b.LoadStatus\x12-\n\x12power_match_status\x18\t \x01(\x0b\x32\x11.PowerMatchStatus*\x87\x03\n\x11MeterSumGridState\x12\x16\n\x12OPER_RELAY_UNKNOWN\x10\x00\x12\x13\n\x0fOPER_RELAY_OPEN\x10\x01\x12\x15\n\x11OPER_RELAY_CLOSED\x10\x02\x12&\n"OPER_RELAY_OFFGRID_AC_GRID_PRESENT\x10\x03\x12+\n\'OPER_RELAY_OFFGRID_READY_FOR_RESYNC_CMD\x10\x04\x12,\n(OPER_RELAY_WAITING_TO_INITIALIZE_ON_GRID\x10\x05\x12\x17\n\x13OPER_RELAY_GEN_OPEN\x10\x06\x12\x19\n\x15OPER_RELAY_GEN_CLOSED\x10\x07\x12\x1a\n\x16OPER_RELAY_GEN_STARTUP\x10\x08\x12\x1d\n\x19OPER_RELAY_GEN_SYNC_READY\x10\t\x12\x1c\n\x18OPER_RELAY_GEN_AC_STABLE\x10\n\x12\x1e\n\x1aOPER_RELAY_GEN_AC_UNSTABLE\x10\x0b*U\n\x08\x42\x61ttMode\x12\x19\n\x15\x42\x41TT_MODE_FULL_BACKUP\x10\x00\x12\x17\n\x13\x42\x41TT_MODE_SELF_CONS\x10\x01\x12\x15\n\x11\x42\x41TT_MODE_SAVINGS\x10\x02*2\n\x0c\x44ryContactId\x12\x07\n\x03NC1\x10\x00\x12\x07\n\x03NC2\x10\x01\x12\x07\n\x03NO1\x10\x02\x12\x07\n\x03NO2\x10\x03*U\n\x14\x44ryContactRelayState\x12\x1a\n\x16\x44\x43_RELAY_STATE_INVALID\x10\x00\x12\x10\n\x0c\x44\x43_RELAY_OFF\x10\x01\x12\x0f\n\x0b\x44\x43_RELAY_ON\x10\x02*K\n\tMeterType\x12\x13\n\x0fMETER_TYPE_NONE\x10\x00\x12\x11\n\rMETER_TYPE_PV\x10\x01\x12\x16\n\x12METER_TYPE_STORAGE\x10\x02\x62\x06proto3' +) _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) -_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'enphase_livestream_pb2', _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "enphase_livestream_pb2", _globals) if not _descriptor._USE_C_DESCRIPTORS: - DESCRIPTOR._loaded_options = None - _globals['_METERSUMGRIDSTATE']._serialized_start=1325 - _globals['_METERSUMGRIDSTATE']._serialized_end=1716 - _globals['_BATTMODE']._serialized_start=1718 - _globals['_BATTMODE']._serialized_end=1803 - _globals['_DRYCONTACTID']._serialized_start=1805 - _globals['_DRYCONTACTID']._serialized_end=1855 - _globals['_DRYCONTACTRELAYSTATE']._serialized_start=1857 - _globals['_DRYCONTACTRELAYSTATE']._serialized_end=1942 - _globals['_METERTYPE']._serialized_start=1944 - _globals['_METERTYPE']._serialized_end=2019 - _globals['_METERCHANNEL']._serialized_start=29 - _globals['_METERCHANNEL']._serialized_end=161 - _globals['_AGGMETERCHANNEL']._serialized_start=164 - _globals['_AGGMETERCHANNEL']._serialized_end=334 - _globals['_METERSUMMARYDATA']._serialized_start=337 - _globals['_METERSUMMARYDATA']._serialized_end=684 - _globals['_DRYCONTACTSTATUS']._serialized_start=686 - _globals['_DRYCONTACTSTATUS']._serialized_end=769 - _globals['_DRYCONTACTNAME']._serialized_start=771 - _globals['_DRYCONTACTNAME']._serialized_end=833 - _globals['_LOADSTATUS']._serialized_start=835 - _globals['_LOADSTATUS']._serialized_end=896 - _globals['_POWERMATCHSTATUS']._serialized_start=898 - _globals['_POWERMATCHSTATUS']._serialized_end=1001 - _globals['_DATAMSG']._serialized_start=1004 - _globals['_DATAMSG']._serialized_end=1322 + DESCRIPTOR._loaded_options = None + _globals["_METERSUMGRIDSTATE"]._serialized_start = 1325 + _globals["_METERSUMGRIDSTATE"]._serialized_end = 1716 + _globals["_BATTMODE"]._serialized_start = 1718 + _globals["_BATTMODE"]._serialized_end = 1803 + _globals["_DRYCONTACTID"]._serialized_start = 1805 + _globals["_DRYCONTACTID"]._serialized_end = 1855 + _globals["_DRYCONTACTRELAYSTATE"]._serialized_start = 1857 + _globals["_DRYCONTACTRELAYSTATE"]._serialized_end = 1942 + _globals["_METERTYPE"]._serialized_start = 1944 + _globals["_METERTYPE"]._serialized_end = 2019 + _globals["_METERCHANNEL"]._serialized_start = 29 + _globals["_METERCHANNEL"]._serialized_end = 161 + _globals["_AGGMETERCHANNEL"]._serialized_start = 164 + _globals["_AGGMETERCHANNEL"]._serialized_end = 334 + _globals["_METERSUMMARYDATA"]._serialized_start = 337 + _globals["_METERSUMMARYDATA"]._serialized_end = 684 + _globals["_DRYCONTACTSTATUS"]._serialized_start = 686 + _globals["_DRYCONTACTSTATUS"]._serialized_end = 769 + _globals["_DRYCONTACTNAME"]._serialized_start = 771 + _globals["_DRYCONTACTNAME"]._serialized_end = 833 + _globals["_LOADSTATUS"]._serialized_start = 835 + _globals["_LOADSTATUS"]._serialized_end = 896 + _globals["_POWERMATCHSTATUS"]._serialized_start = 898 + _globals["_POWERMATCHSTATUS"]._serialized_end = 1001 + _globals["_DATAMSG"]._serialized_start = 1004 + _globals["_DATAMSG"]._serialized_end = 1322 # @@protoc_insertion_point(module_scope) From a6d8db8d819bbb473e81b61e33504a9d81cbb789 Mon Sep 17 00:00:00 2001 From: Trefor Southwell Date: Mon, 3 Aug 2026 16:05:16 +0200 Subject: [PATCH 3/7] fix(enphase): keep livestream credentials out of the log The livestream bootstrap response carries aws_token_value and aws_digest, the live credentials for the account's AWS IoT stream. Debug API logging redacted only token/auth_token/access_token, so both were written out in full - and Predbat logs are routinely shared for debugging. The endpoint and topic are still logged so the call stays diagnosable. Co-Authored-By: Claude Opus 5 (1M context) --- apps/predbat/enphase.py | 4 +++- apps/predbat/tests/test_enphase_api.py | 26 ++++++++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/apps/predbat/enphase.py b/apps/predbat/enphase.py index f37ee3508..de98e9dfe 100644 --- a/apps/predbat/enphase.py +++ b/apps/predbat/enphase.py @@ -1625,7 +1625,9 @@ def _log_api_call(self, method, path, params, status, json_data, text): return if isinstance(json_data, dict): redacted = dict(json_data) - for key in ("token", "auth_token", "access_token"): + # aws_token_value/aws_digest are the livestream's AWS IoT credentials - short-lived, + # but Predbat logs get shared for debugging, so they must never be written out. + for key in ("token", "auth_token", "access_token", "aws_token_value", "aws_digest"): if key in redacted: redacted[key] = "***redacted***" preview = json.dumps(redacted, default=str) diff --git a/apps/predbat/tests/test_enphase_api.py b/apps/predbat/tests/test_enphase_api.py index 02acd891a..64c6eecd0 100644 --- a/apps/predbat/tests/test_enphase_api.py +++ b/apps/predbat/tests/test_enphase_api.py @@ -982,6 +982,31 @@ def test_log_api_call_redacts_token(): assert captured == [] +def test_log_api_call_redacts_livestream_credentials(): + """The livestream bootstrap's token and signature must never reach the log. + + Predbat logs are routinely shared for debugging, and this response carries live credentials + for the account's AWS IoT stream. + """ + api = MockEnphaseAPI() + captured = [] + api.log = lambda message: captured.append(message) + api.debug_api = True + api._log_api_call( + "GET", + "/pv/aws_sigv4/livestream.json", + {"serial_num": "122530006866"}, + 200, + {"aws_token_value": "token-must-not-be-logged", "aws_digest": "signature-must-not-be-logged", "aws_iot_endpoint": "iot.example.com", "live_stream_topic": "v1/live-stream/abc123"}, + "", + ) + assert "token-must-not-be-logged" not in captured[0] + assert "signature-must-not-be-logged" not in captured[0] + # Non-secret fields are still logged, so the call remains diagnosable + assert "iot.example.com" in captured[0] + assert "v1/live-stream/abc123" in captured[0] + + def test_login_dedupes_sites(): """Duplicate sites in the search response collapse to a single entry (no double-publish).""" api = MockEnphaseAPI() @@ -1977,6 +2002,7 @@ def run_enphase_api_tests(my_predbat): test_get_battery_status_handles_na() test_reads_handle_na_values() test_log_api_call_redacts_token() + test_log_api_call_redacts_livestream_credentials() test_login_dedupes_sites() test_run_single_site_publishes_once() test_run_no_battery_returns_false_without_raising() From 1089c4f3a29052116ee83c03214dc667c502d0a8 Mon Sep 17 00:00:00 2001 From: Trefor Southwell Date: Mon, 3 Aug 2026 16:14:00 +0200 Subject: [PATCH 4/7] fix(enphase): treat a pending schedule family as supported The cloud reports a schedule family as scheduleStatus "pending" while a change settles on the gateway - the normal state straight after any write Predbat makes. Only "active"/"enabled"/"supported"/"available" counted as supported, so a site with a perfectly good charge-from-grid family was judged incapable of it, automatic_config raised and run() returned False: Warn: Automatic configuration skipped - Charge-from-grid (CFG) scheduling not supported on this site, cannot configure seen on a site whose cfg family held an active schedule and whose profile reported scheduleSupported true for both cfg and dtg. "pending" now counts as supported, and so does any family that actually holds a schedule, whatever the status string says. "not_supported" still reports unsupported. Also aligns the livestream fallback test with the settled-bucket selection that landed on main in #4430: the helper froze time one bucket too early, and load now falls back to the energy-balance residual rather than being left empty. Co-Authored-By: Claude Opus 5 (1M context) --- apps/predbat/enphase.py | 10 +++++--- apps/predbat/tests/test_enphase_api.py | 35 +++++++++++++++++++++++--- 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/apps/predbat/enphase.py b/apps/predbat/enphase.py index de98e9dfe..d4de25b55 100644 --- a/apps/predbat/enphase.py +++ b/apps/predbat/enphase.py @@ -1354,10 +1354,14 @@ async def get_schedules(self, site_id): entry = details[0] if details else {} details = await self._prune_sibling_schedules(site_id, family_key, details, entry) # "supported" gates whether Predbat can use this schedule family. Real accounts report - # a per-family scheduleStatus ("active" seen so far); treat the usable statuses as - # supported, with a fallback to the (unverified) boolean flags. + # a per-family scheduleStatus; "active" and "pending" have both been seen, and + # "not_supported" is how a genuinely unavailable family reports. "pending" only means a + # schedule change is still settling on the gateway - which is the normal state straight + # after any write Predbat makes - so it must not be read as unsupported, or Predbat + # decides mid-run that the site cannot charge from grid and abandons configuration. + # A family that actually holds a schedule is supported whatever the status says. status_text = str(family_data.get("scheduleStatus", "")).strip().lower() - supported = status_text in ("active", "enabled", "supported", "available") or bool(family_data.get("scheduleSupported") or family_data.get("forceScheduleSupported")) + supported = status_text in ("active", "enabled", "supported", "available", "pending") or bool(details) or bool(family_data.get("scheduleSupported") or family_data.get("forceScheduleSupported")) parsed[family_key] = { "id": entry.get("scheduleId") or entry.get("id"), "startTime": entry.get("startTime"), diff --git a/apps/predbat/tests/test_enphase_api.py b/apps/predbat/tests/test_enphase_api.py index 64c6eecd0..4dedc56dd 100644 --- a/apps/predbat/tests/test_enphase_api.py +++ b/apps/predbat/tests/test_enphase_api.py @@ -692,8 +692,8 @@ def bucket(value): class _Fixed(original): @classmethod def now(cls, tz=None): - """Return a time inside the interval after the bucket that publish_data reads.""" - return original.fromtimestamp(start + int(81.5 * 900), tz) + """Return a time inside index 82, so the settled bucket publish_data reads is 80.""" + return original.fromtimestamp(start + int(82.5 * 900), tz) enphase_module.datetime = _Fixed try: @@ -717,12 +717,18 @@ def test_publish_prefers_the_measured_livestream_reading(): def test_publish_falls_back_to_buckets_without_a_livestream_reading(): - """With no livestream reading the bucket-derived values are still published.""" + """With no livestream reading all four sensors fall back to the settled bucket values. + + Load falls back to the energy-balance residual, so the four still agree and a power-flow + display still balances - a livestream failure degrades the sensors rather than blanking them. + """ api = MockEnphaseAPI() published = _publish_with_buckets(api) assert published["pv"] == 1000.0 assert published["grid"] == 400.0 assert published["battery"] == -800.0 + assert published["load"] == 600.0 # 1000 + 400 - 800 + assert published["load"] == published["pv"] + published["grid"] + published["battery"] def test_interval_power(): @@ -928,6 +934,28 @@ def test_get_schedules_supported_from_status(): assert api.dtg_supported("12345") is False # 'not_supported' status +def test_get_schedules_pending_family_is_still_supported(): + """A family whose scheduleStatus is 'pending' is supported - a write is in flight, that is all. + + The cloud reports a family as 'pending' while a schedule change settles on the gateway, which + happens right after any write Predbat makes. Treating that as unsupported made Predbat decide + the site could not do charge-from-grid at all and abandon automatic configuration, even with an + active schedule sitting in the family. + """ + api = MockEnphaseAPI() + detail = {"scheduleId": "c1", "startTime": "04:30", "endTime": "04:40", "limit": 5, "scheduleType": "CFG", "isDeleted": False, "isEnabled": True, "scheduleStatus": "active"} + payload = { + "type": "BATTERY_SCHEDULES_CONFIG", + "cfg": {"scheduleStatus": "pending", "count": 1, "details": [detail]}, + "dtg": {"scheduleStatus": "pending", "count": 1, "details": [dict(detail, scheduleId="d1", scheduleType="DTG")]}, + "rbd": {"scheduleStatus": "active", "count": 0}, + } + api.set_http_response("/service/batteryConfig/api/v1/battery/sites/12345/schedules", 200, payload) + run_async(api.get_schedules("12345")) + assert api.schedules["12345"]["cfg"]["supported"] is True + assert api.dtg_supported("12345") is True + + def test_inverter_def_enphase(): """EnphaseCloud INVERTER_DEF exists with the agreed capability flags.""" from config import INVERTER_DEF @@ -2043,6 +2071,7 @@ def run_enphase_api_tests(my_predbat): test_automatic_config_no_dtg_raises() test_automatic_config_no_charge_support_raises() test_get_schedules_supported_from_status() + test_get_schedules_pending_family_is_still_supported() test_inverter_def_enphase() test_run_first_polls_all_tiers() test_get_today() From 026aefd18ab1beb960caa2569214f27e6fbbc0b3 Mon Sep 17 00:00:00 2001 From: Trefor Southwell Date: Mon, 3 Aug 2026 16:42:34 +0200 Subject: [PATCH 5/7] fix(enphase): publish grid power positive when exporting Predbat's convention is grid positive on EXPORT - web.py's power flow reads `grid_power >= 10` as exporting, and sigenergy.py documents the same - but enphase.py published `import - export`, so every Enphase user's grid power has been inverted, showing import where the flow diagram expects export. The livestream's grid channel is negative while exporting too, so both the measured and the bucket-derived paths needed flipping. The residual load derivation follows from the sign change: in Predbat's signs (grid +export, battery +discharge) the balance is pv + battery - grid, not pv + grid + battery. Load values are unchanged by this; only the grid sensor's sign moves. Battery power is left alone: `discharge - charge` (positive on discharge) already matches the core convention in inverter.py, which detects charging as `power < -threshold`, and sigenergy's documented mapping. Note that web.py's power flow reads battery the opposite way (`>= 10` as charging), which looks like a display bug affecting every integration rather than something to correct here. Co-Authored-By: Claude Opus 5 (1M context) --- apps/predbat/enphase.py | 11 +++-- apps/predbat/tests/test_enphase_api.py | 56 ++++++++++++++++++++------ 2 files changed, 51 insertions(+), 16 deletions(-) diff --git a/apps/predbat/enphase.py b/apps/predbat/enphase.py index d4de25b55..a4d9869c5 100644 --- a/apps/predbat/enphase.py +++ b/apps/predbat/enphase.py @@ -580,7 +580,10 @@ async def publish_data(self, site_id): channel_watts = {channel: interval_power(arrays.get(channel, []), start_time, interval_length, now_ts) for channel in ("production", "import", "export", "charge", "discharge")} pv_power = channel_watts.get("production", 0.0) - grid_power = round(channel_watts.get("import", 0.0) - channel_watts.get("export", 0.0), 1) + # Predbat's convention is grid positive when EXPORTING and battery positive when + # DISCHARGING (see the power flow in web.py and the charge detection in inverter.py), so + # export leads the grid subtraction and discharge leads the battery one. + grid_power = round(channel_watts.get("export", 0.0) - channel_watts.get("import", 0.0), 1) battery_power = round(channel_watts.get("discharge", 0.0) - channel_watts.get("charge", 0.0), 1) # House load is the energy-balance residual of the other three, taken from the same settled # bucket so the four sensors agree and a power-flow display balances. This is exactly how the @@ -591,7 +594,8 @@ async def publish_data(self, site_id): # CT clamps and the battery telemetry. While the battery is cycling hard those terms dwarf # the house term and the residual can go unphysical, so it is clamped at zero - a negative # house load would render nonsensically. load_today remains the trustworthy energy figure. - load_power = max(0.0, round(pv_power + grid_power + battery_power, 1)) + # In Predbat's signs (grid +export, battery +discharge) the balance is pv + battery - grid. + load_power = max(0.0, round(pv_power + battery_power - grid_power, 1)) # Prefer the livestream when we have one: those four channels are separately metered and # instantaneous, where the buckets are a 15-minute average and load is only ever a residual. @@ -599,7 +603,8 @@ async def publish_data(self, site_id): live = self.live_power.get(site_id) or {} if live: pv_power = live.get("pv", pv_power) - grid_power = live.get("grid", grid_power) + # The livestream reports grid negative while exporting, the opposite of Predbat's sign. + grid_power = round(-live["grid"], 1) if "grid" in live else grid_power battery_power = live.get("battery", battery_power) load_power = live.get("load", load_power) diff --git a/apps/predbat/tests/test_enphase_api.py b/apps/predbat/tests/test_enphase_api.py index 4dedc56dd..03d0197c7 100644 --- a/apps/predbat/tests/test_enphase_api.py +++ b/apps/predbat/tests/test_enphase_api.py @@ -669,8 +669,8 @@ def test_gateway_serial_read_from_today(): assert gateway_serial({}) is None -def _publish_with_buckets(api): - """Publish a site whose /today buckets give pv 1000 W, grid 400 W, battery -800 W.""" +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 def bucket(value): @@ -681,7 +681,7 @@ def bucket(value): api.today["12345"] = { "totals": {}, - "arrays": {"production": bucket(250), "import": bucket(100), "export": bucket(0), "charge": bucket(200), "discharge": bucket(0)}, + "arrays": {"production": bucket(production), "import": bucket(imp), "export": bucket(exp), "charge": bucket(charge), "discharge": bucket(discharge)}, "start_time": start, "interval_length": 900, } @@ -713,7 +713,34 @@ def test_publish_prefers_the_measured_livestream_reading(): api = MockEnphaseAPI() api.live_power["12345"] = {"pv": 4632.8, "battery": 32.0, "grid": -2686.1, "load": 1978.7, "soc": 100} published = _publish_with_buckets(api) - assert published == {"pv": 4632.8, "grid": -2686.1, "battery": 32.0, "load": 1978.7} + assert published == {"pv": 4632.8, "grid": 2686.1, "battery": 32.0, "load": 1978.7} # grid flipped to +export + + +def test_grid_power_is_positive_when_exporting(): + """Grid power follows Predbat's convention: positive exporting, negative importing. + + web.py's power flow reads `grid_power >= 10` as exporting, and sigenergy documents the same. + The Enphase channels are the other way round (a livestream grid reading is negative while + 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} + published = _publish_with_buckets(api) + assert published["grid"] == 2686.1 # exporting 2.7 kW -> positive + + +def test_grid_power_from_buckets_is_positive_when_exporting(): + """The bucket fallback follows the same convention as the livestream.""" + api = MockEnphaseAPI() + published = _publish_with_buckets(api, production=250, imp=0, exp=100, charge=0, discharge=0) + assert published["grid"] == 400.0 # 100 Wh exported over a 15-min bucket + + +def test_published_power_satisfies_the_predbat_energy_balance(): + """With Predbat's signs the balance is load = pv + battery - grid, not pv + battery + grid.""" + api = MockEnphaseAPI() + published = _publish_with_buckets(api) + assert published["load"] == published["pv"] + published["battery"] - published["grid"] def test_publish_falls_back_to_buckets_without_a_livestream_reading(): @@ -725,10 +752,10 @@ def test_publish_falls_back_to_buckets_without_a_livestream_reading(): api = MockEnphaseAPI() published = _publish_with_buckets(api) assert published["pv"] == 1000.0 - assert published["grid"] == 400.0 - assert published["battery"] == -800.0 - assert published["load"] == 600.0 # 1000 + 400 - 800 - assert published["load"] == published["pv"] + published["grid"] + published["battery"] + assert published["grid"] == -400.0 # importing + assert published["battery"] == -800.0 # charging + assert published["load"] == 600.0 # 1000 - 800 + 400 + assert published["load"] == published["pv"] + published["battery"] - published["grid"] def test_interval_power(): @@ -801,7 +828,7 @@ def _published_power(api): def test_load_power_is_derived_from_the_other_channels(): - """Load is the energy-balance residual: pv + grid + battery. + """Load is the energy-balance residual: pv + battery - grid, in Predbat's signs. The cloud's own consumption channel is exactly this sum, and `get_latest_power` reports production rather than consumption, so publishing that as load made the load sensor track PV. @@ -810,9 +837,9 @@ def test_load_power_is_derived_from_the_other_channels(): api = _api_with_today_buckets(production=250, imp=100, exp=0, charge=0, discharge=25) run_async(api.publish_data("12345")) pv, grid, battery, load = _published_power(api) - assert (pv, grid, battery) == (1000.0, 400.0, 100.0) + assert (pv, grid, battery) == (1000.0, -400.0, 100.0) # grid negative: importing assert load == 1500.0 - assert load == pv + grid + battery # the power-flow card must balance + assert load == pv + battery - grid # the power-flow card must balance def test_load_power_never_goes_negative(): @@ -825,7 +852,7 @@ def test_load_power_never_goes_negative(): api = _api_with_today_buckets(production=0, imp=1461, exp=714, charge=1796, discharge=203) run_async(api.publish_data("12345")) pv, grid, battery, load = _published_power(api) - assert pv + grid + battery < 0 # the raw residual really is negative + assert pv + battery - grid < 0 # the raw residual really is negative assert load == 0.0 @@ -1160,7 +1187,7 @@ def now(cls, tz=None): assert items["sensor.predbat_enphase_12345_export_today"]["state"] == 0.4 assert items["sensor.predbat_enphase_12345_battery_reserve_min"]["state"] == 5 assert items["sensor.predbat_enphase_12345_pv_power"]["state"] == 4000.0 # 1000 Wh / 0.25h - assert items["sensor.predbat_enphase_12345_grid_power"]["state"] == 400.0 + assert items["sensor.predbat_enphase_12345_grid_power"]["state"] == -400.0 # importing assert items["sensor.predbat_enphase_12345_battery_power"]["state"] == -800.0 # charging assert items["sensor.predbat_enphase_12345_load_power"]["state"] == 3600.0 # derived residual @@ -2066,6 +2093,9 @@ def run_enphase_api_tests(my_predbat): test_gateway_serial_read_from_today() test_publish_prefers_the_measured_livestream_reading() test_publish_falls_back_to_buckets_without_a_livestream_reading() + test_grid_power_is_positive_when_exporting() + test_grid_power_from_buckets_is_positive_when_exporting() + test_published_power_satisfies_the_predbat_energy_balance() test_get_schedules_parses_families() test_automatic_config() test_automatic_config_no_dtg_raises() From 91ae498ecfa333d7b97f8eeecd8bb2e7effdeb0e Mon Sep 17 00:00:00 2001 From: Trefor Southwell Date: Mon, 3 Aug 2026 16:52:22 +0200 Subject: [PATCH 6/7] fix(enphase): never republish a stale livestream reading Livestream readings are instantaneous and carry no usable timestamp of their own (DataMsg.timestamp is a constant), so keeping one around past its moment presents an old measurement as current. Two ways that happened: - Caching. _load_cache restores each key's storage age into data_age, so a restart within ENPHASE_REFRESH_POWER left the refresh gate satisfied, get_live_power unrun and the restored reading published as live. live_power is now in-memory only. - A failed read left the previous reading in place, so a stream outage mid-run republished the last good measurement indefinitely, no restart required. A failure now drops it. Either way the sensors fall back to the bucket values, which lag but are genuinely current. Co-Authored-By: Claude Opus 5 (1M context) --- apps/predbat/enphase.py | 22 ++++++++++++++------ apps/predbat/tests/test_enphase_api.py | 28 ++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 6 deletions(-) diff --git a/apps/predbat/enphase.py b/apps/predbat/enphase.py index a4d9869c5..26d597365 100644 --- a/apps/predbat/enphase.py +++ b/apps/predbat/enphase.py @@ -70,7 +70,9 @@ ENPHASE_LIVESTREAM_TIMEOUT = 15 # seconds to wait for a livestream message before giving up LIVESTREAM_BOOTSTRAP = "/pv/aws_sigv4/livestream.json" # returns the AWS IoT endpoint, topic and authorizer credentials -ENPHASE_CACHE_KEYS = ["sites", "battery_status", "battery_settings", "profile", "schedules", "site_settings", "today", "latest_power", "live_power"] +# live_power is deliberately absent: livestream readings are instantaneous and carry no usable +# timestamp, so a restored one would be republished as if current. In-memory only. +ENPHASE_CACHE_KEYS = ["sites", "battery_status", "battery_settings", "profile", "schedules", "site_settings", "today", "latest_power"] ENPHASE_CACHE_VERSION = 2 # Battery profiles accepted by the profile endpoint @@ -1203,6 +1205,18 @@ async def get_live_power(self, site_id): re-authorising every `live_stream_duration` (900s). Returns the reading, or None on any failure, leaving the caller to fall back to the bucket-derived values. """ + reading = await self._fetch_live_power(site_id) + if reading: + 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) + return reading + + async def _fetch_live_power(self, site_id): + """Bootstrap the livestream and return one decoded reading, or None if unavailable.""" if not (HAS_AIOMQTT and HAS_LIVESTREAM_PROTOBUF): return None serial = gateway_serial(self.today.get(site_id, {})) @@ -1211,11 +1225,7 @@ async def get_live_power(self, site_id): boot = await self.request_json("GET", LIVESTREAM_BOOTSTRAP, params={"serial_num": serial}) if not boot or not boot.get("aws_iot_endpoint") or not boot.get("live_stream_topic"): return None - reading = await self._read_livestream(site_id, boot, serial) - if reading: - self.live_power[site_id] = reading - await self._save_cache("live_power", self.live_power) - return reading + return await self._read_livestream(site_id, boot, serial) async def _read_livestream(self, site_id, boot, serial): """Connect to AWS IoT, take the first livestream message for a site, then disconnect. diff --git a/apps/predbat/tests/test_enphase_api.py b/apps/predbat/tests/test_enphase_api.py index 03d0197c7..16f8552d3 100644 --- a/apps/predbat/tests/test_enphase_api.py +++ b/apps/predbat/tests/test_enphase_api.py @@ -716,6 +716,32 @@ def test_publish_prefers_the_measured_livestream_reading(): assert published == {"pv": 4632.8, "grid": 2686.1, "battery": 32.0, "load": 1978.7} # grid flipped to +export +def test_live_power_is_never_persisted(): + """Livestream readings stay in memory only. + + They are instantaneous and carry no usable timestamp of their own, so restoring one from the + cache after a restart would republish an old measurement as if it were current - and the + refresh gate can be satisfied by the restored age, so nothing would immediately correct it. + """ + from enphase import ENPHASE_CACHE_KEYS + + 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. + + Falling back to the (lagging but current) bucket values beats presenting a stale measurement + as though it were live. + """ + 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} + # 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 + + def test_grid_power_is_positive_when_exporting(): """Grid power follows Predbat's convention: positive exporting, negative importing. @@ -2093,6 +2119,8 @@ def run_enphase_api_tests(my_predbat): test_gateway_serial_read_from_today() 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_grid_power_is_positive_when_exporting() test_grid_power_from_buckets_is_positive_when_exporting() test_published_power_satisfies_the_predbat_energy_balance() From 41e59a37cd5f5770b8a7ee22a49f9a27d15250d0 Mon Sep 17 00:00:00 2001 From: Trefor Southwell Date: Mon, 3 Aug 2026 16:58:48 +0200 Subject: [PATCH 7/7] 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) --- apps/predbat/enphase.py | 15 ++++++--- apps/predbat/tests/test_enphase_api.py | 43 +++++++++++++++++++------- docs/components.md | 3 +- 3 files changed, 44 insertions(+), 17 deletions(-) diff --git a/apps/predbat/enphase.py b/apps/predbat/enphase.py index 26d597365..b0e97fbce 100644 --- a/apps/predbat/enphase.py +++ b/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): diff --git a/apps/predbat/tests/test_enphase_api.py b/apps/predbat/tests/test_enphase_api.py index 16f8552d3..dd42d3d54 100644 --- a/apps/predbat/tests/test_enphase_api.py +++ b/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() diff --git a/docs/components.md b/docs/components.md index 986188d25..9720f48d2 100644 --- a/docs/components.md +++ b/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)