diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index 29fc89542..af273424e 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -69,6 +69,7 @@ coro cprofile creds crosscharge +csp customisation Customise cvalue @@ -172,6 +173,7 @@ Ginlong GivEnergy givtcp gnueabihf +greppable gridcode gridconsumption gridconsumptionpower @@ -188,6 +190,7 @@ hasattr hass hassapi hassio +hazmat heatingactive heatingtemp heatpump @@ -326,7 +329,9 @@ perc percnt peterhaban photovoltaics +pinv pitem +pkcs1v15 pkwh plenticore postdata @@ -359,6 +364,7 @@ pytest pytz randn rarr +rawkey reactivepower recp redownload @@ -469,6 +475,7 @@ twinx tzfile tzpath unconfigured +unfetched unparseable unsmoothed unstaged diff --git a/apps/predbat/components.py b/apps/predbat/components.py index 7eb6405ab..a874cecf9 100644 --- a/apps/predbat/components.py +++ b/apps/predbat/components.py @@ -35,6 +35,7 @@ from db_manager import DatabaseManager from fox import FoxAPI from deye import DeyeAPI +from sunsynk import SunsynkAPI from enphase import EnphaseAPI from kraken import KrakenAPI from web_mcp import PredbatMCPServer @@ -483,6 +484,27 @@ "phase": 1, "can_restart": True, }, + "sunsynk": { + "class": SunsynkAPI, + "name": "Sunsynk Connect", + "event_filter": "predbat_sunsynk_", + "phase": 1, + "can_restart": True, + # key required so the component only starts for Sunsynk instances, not fleet-wide - + # Sunsynk uses a single injected-token credential (sunsynk_key), like teslemetry's + # "key", rather than deye/solis's multiple-auth-path required_or gate. + "args": { + "key": {"required": True, "config": "sunsynk_key"}, + "inverter_sn": {"required": False, "config": "sunsynk_inverter_sn"}, + "plant_id": {"required": False, "config": "sunsynk_plant_id"}, + "automatic": {"required": False, "default": False, "config": "sunsynk_automatic"}, + "automatic_ignore_pv": {"required": False, "default": False, "config": "sunsynk_automatic_ignore_pv"}, + "auth_method": {"required": False, "config": "sunsynk_auth_method"}, + "token_expires_at": {"required": False, "config": "sunsynk_token_expires_at"}, + "token_hash": {"required": False, "config": "sunsynk_token_hash"}, + "base_url": {"required": False, "default": "https://api.sunsynk.net", "config": "sunsynk_base_url"}, + }, + }, "load_ml": { "class": LoadMLComponent, "name": "ML Load Forecaster", diff --git a/apps/predbat/config.py b/apps/predbat/config.py index fdfa23730..6f8f3636a 100644 --- a/apps/predbat/config.py +++ b/apps/predbat/config.py @@ -2004,6 +2004,42 @@ "charge_discharge_with_rate": False, "target_soc_used_for_discharge": True, }, + # Sunsynk-branded hardware is Deye hardware under the skin, and the cloud API shares the + # same 6-slot-TOU control model, "HH:MM" time format, and "%" SOC units - so this entry is + # copied verbatim from "DeyeCloud" above (only "name" differs). Sunsynk's read path (device + # discovery + telemetry publish) ships now; the control-specific fields below + # (charge_time_format, support_charge_freeze/support_discharge_freeze, has_timed_pause, + # etc.) are inherited placeholders and are inert until control lands (Task 14+) - confirm + # each one against Sunsynk's real control API before it is relied upon for writes. + "SUNSYNK": { + "name": "SUNSYNK", + "has_rest_api": False, + "has_mqtt_api": False, + "output_charge_control": "power", + "charge_control_immediate": False, + "has_charge_enable_time": True, + "has_discharge_enable_time": True, + "has_target_soc": True, + "has_reserve_soc": True, + "has_timed_pause": False, + "charge_time_format": "HH:MM", + "charge_time_entity_is_option": True, + "soc_units": "%", + "num_load_entities": 1, + "has_ge_inverter_mode": False, + "has_ge_eco_toggle": False, + "has_fox_inverter_mode": False, + "time_button_press": True, + "clock_time_format": "%Y-%m-%d %H:%M:%S", + "write_and_poll_sleep": 2, + "has_time_window": False, + "support_charge_freeze": True, + "support_discharge_freeze": True, + "has_idle_time": False, + "can_span_midnight": False, + "charge_discharge_with_rate": False, + "target_soc_used_for_discharge": True, + }, "SolaxCloud": { "name": "SolaxCloud", "has_rest_api": False, @@ -2299,6 +2335,15 @@ "deye_inverter_sn": {"type": "string|string_list", "empty": False}, "deye_automatic": {"type": "boolean"}, "deye_automatic_ignore_pv": {"type": "boolean"}, + "sunsynk_key": {"type": "string", "empty": False}, + "sunsynk_inverter_sn": {"type": "string|string_list", "empty": False}, + "sunsynk_plant_id": {"type": "string", "empty": False}, + "sunsynk_automatic": {"type": "boolean"}, + "sunsynk_automatic_ignore_pv": {"type": "boolean"}, + "sunsynk_auth_method": {"type": "string", "empty": False}, + "sunsynk_token_expires_at": {"type": "string", "empty": False}, + "sunsynk_token_hash": {"type": "string", "empty": False}, + "sunsynk_base_url": {"type": "string", "empty": False}, "teslemetry_key": {"type": "string", "empty": False}, "teslemetry_site_id": {"type": "string|string_list"}, "teslemetry_base_url": {"type": "string", "empty": False}, diff --git a/apps/predbat/sunsynk.py b/apps/predbat/sunsynk.py new file mode 100644 index 000000000..f0a04050f --- /dev/null +++ b/apps/predbat/sunsynk.py @@ -0,0 +1,1065 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# Sunsynk Connect API Library +# ----------------------------------------------------------------------------- + +"""Sunsynk Connect Cloud API integration. + +Injected-token Bearer client for Sunsynk Connect inverters. Predbat.com's SaaS edge +functions own the Sunsynk RSA publicKey -> login -> access-token exchange and refresh +chain server-side; the ``sunsynk_key`` config value handed to this component IS the +resulting Bearer access token, so ``SunsynkAPI`` never performs RSA signing or login +itself here - it only issues plain Bearer-authenticated requests and flags +``needs_reauth`` when the injected token is rejected, for the SaaS side to notice and +refresh. Device discovery and read-only telemetry are implemented here (Task 9), along with +entity publishing via ``publish_data()`` (Task 10), the automatic-config set_arg wiring via +``automatic_config()`` (Task 11), and the ``run()`` loop plus ``COMPONENT_LIST``/config +registration (Task 12); control writes are added in a later task. + +The ONE place batpred performs its own Sunsynk login is the standalone CLI harness at the +bottom of this module (Task 13, after ``MockBase``), for HA add-on / self-managed users who +supply a Sunsynk Connect username+password directly rather than going through Predbat.com's +SaaS token injection. That standalone login helper (``_sunsynk_standalone_login()``) does the +publicKey -> RSA-encrypt-password -> ``/oauth/token/new`` exchange in Python, and is the only +code path in this module that imports ``cryptography`` - that import is guarded inside +``_sunsynk_rsa_encrypt_password()`` so importing this module as a SaaS component never +requires the dependency, only running the CLI harness's login path does. +""" + +import argparse +import asyncio +import base64 +import hashlib +import json +import os +import random +import time +from datetime import datetime, timedelta, timezone + +import aiohttp +import requests + +from component_base import ComponentBase +from oauth_mixin import OAuthMixin +from predbat_metrics import record_api_call + +SUNSYNK_DOMAIN = "https://api.sunsynk.net" +TIMEOUT = 60 +SUNSYNK_RETRIES = 10 + +# Defined locally (rather than imported from const.py) to avoid dependency issues, matching +# fox.py's TIME_FORMAT_HA approach - components should stay standalone-testable via MockBase. +SUNSYNK_DATE_FORMAT = "%Y-%m-%d" + +# Maximum age (minutes) of cached data before an API refresh is triggered, matching fox.py's +# FOX_REFRESH_* constants. STATIC and REALTIME are consumed by the read methods below (device +# list, and flow/battery/grid/input/output telemetry); SETTINGS is not yet used - device +# settings read/write land in a later task. +SUNSYNK_REFRESH_STATIC = 24 * 60 # Device/plant list rarely changes +SUNSYNK_REFRESH_SETTINGS = 60 # Device settings +SUNSYNK_REFRESH_REALTIME = 5 # Real time monitoring data + +# Storage cache keys for device data persisted between reboots, matching fox.py's +# FOX_CACHE_KEYS. "device_settings" is not yet used - device settings land in a later task. +SUNSYNK_CACHE_KEYS = ["device_list", "device_settings", "device_values"] + +# Sentinel default for _publish_battery_extras()'s `battery` parameter, distinct from a real +# `None` value. `None` means "already fetched this cycle and the fetch failed" (do NOT +# re-fetch - that would hit a failing endpoint twice per publish cycle); the sentinel means +# "not attempted yet" (safe to lazily fetch here). A plain `None` default can't tell these +# apart, since a failed get_battery(sn) call also returns None. +_UNFETCHED = object() + + +class SunsynkAPI(ComponentBase, OAuthMixin): + """Sunsynk Connect API client.""" + + def initialize(self, key, automatic, automatic_ignore_pv=False, inverter_sn=None, plant_id=None, auth_method=None, token_expires_at=None, token_hash=None, base_url=SUNSYNK_DOMAIN): + """Initialise the Sunsynk API component.""" + self.key = key + self.automatic = automatic + self.automatic_ignore_pv = automatic_ignore_pv + self.base_url = base_url + self.failures_total = 0 + # Set on a 401/403 so callers (and the SaaS oauth-refresh edge function, which owns + # the RSA/login chain) can notice the injected token was rejected; cleared again on + # the next successful request. This component never attempts a re-login itself. + self.needs_reauth = False + + # Scopes device discovery to a single plant (config key sunsynk_plant_id, wired by a + # later task). Multi-plant Sunsynk accounts are a v1 non-goal - see get_device_list(). + self.plant_id = plant_id + + # Cache state. device_settings lands in a later task (empty/unused here). + self.device_list = [] + self.device_values = {"flow": {}, "battery": {}, "grid": {}, "input": {}, "output": {}} + self.device_settings = {} + self.data_age = {} + + # Tracks the set of discovered inverter SNs across run() cycles, so run() can tell + # apart "device set changed since last cycle" (re-run automatic_config even when + # first=False - e.g. a new inverter was added to the account) from "same devices as + # last time" (skip automatic_config, first=False already covers the initial wiring). + self._known_sns = set() + + # Initialise OAuth bookkeeping (expiry tracking, provider_name etc.) for consistency + # with fox.py/deye.py, even though Sunsynk's injected-token model means refreshing is + # done entirely server-side rather than via check_and_refresh_oauth_token(). + self._init_oauth(auth_method=auth_method, key=key, token_expires_at=token_expires_at, provider_name="sunsynk") + # _init_oauth() only sets self.access_token when auth_method == "oauth" (None + # otherwise) - but 'key' IS the Bearer access token here regardless of auth_method, + # so it must be (re)applied after _init_oauth or get_headers() would send "Bearer + # None" whenever auth_method isn't "oauth". + self.access_token = key + # _init_oauth() also unconditionally resets token_hash to "" (see oauth_mixin.py) so + # the configured value must be applied after it too, exactly as fox.py/deye.py do - + # otherwise a configured hash is silently discarded and the Predbat.com SaaS dedup + # keyed on it breaks. + self.token_hash = token_hash or "" + + # Convert inverter_sn to list, same normalisation as fox.py/deye.py + if inverter_sn is None: + self.inverter_sn_filter = [] + elif isinstance(inverter_sn, str): + self.inverter_sn_filter = [inverter_sn] + else: + self.inverter_sn_filter = inverter_sn + + def get_headers(self): + """Return the plain Bearer authorisation headers used for every Sunsynk request. + + No request signing (unlike fox.py's MD5 signature) - Sunsynk reads use a plain + Bearer token, and that token is injected/refreshed server-side rather than derived + here. + """ + return {"Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json"} + + def is_alive(self): + """Check if the API is alive (component started and holding a usable token).""" + return self.api_started and bool(self.access_token) + + async def request_get(self, path, post=False, datain=None): + """ + Retry wrapper around request_get_func. + """ + retries = 0 + result = None + self.log("Sunsynk: API Requesting {} {} - data {}".format("POST" if post else "GET", path, datain)) + + while retries < SUNSYNK_RETRIES: + result, allow_retry = await self.request_get_func(path, post=post, datain=datain) + if result is not None: + return result + if not allow_retry: + break + retries += 1 + await asyncio.sleep(retries * random.random()) + self.log("Sunsynk: API Response failed after {} retries for {}".format(SUNSYNK_RETRIES, path)) + return result + + async def request_get_func(self, path, post=False, datain=None): + """ + Perform a single Sunsynk API request. + + Plain Bearer auth, no request signing. On 401/403 there is no re-login attempt - + the access token is injected and refreshed server-side by the SaaS edge function, so + a stale token here just sets self.needs_reauth for the caller to notice, rather than + triggering a login flow this component doesn't implement. + """ + headers = self.get_headers() + url = self.base_url + path + self.log("Sunsynk: API Request: path {} post {} datain {}".format(path, post, datain)) + + timeout = aiohttp.ClientTimeout(total=TIMEOUT) + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + if post: + async with session.post(url, headers=headers, json=datain) as response: + status_code = response.status + try: + data = await response.json() + except (aiohttp.ContentTypeError, json.JSONDecodeError): + self.log("Warn: Sunsynk: Failed to decode response from {} code {}".format(url, status_code)) + data = None + else: + async with session.get(url, headers=headers, params=datain) as response: + status_code = response.status + try: + data = await response.json() + except (aiohttp.ContentTypeError, json.JSONDecodeError): + self.log("Warn: Sunsynk: Failed to decode response from {} code {}".format(url, status_code)) + data = None + except (aiohttp.ClientError, asyncio.TimeoutError) as e: + self.log(f"Warn: Sunsynk: Exception during request to {url}: {e}") + self.failures_total += 1 + record_api_call("sunsynk", False, "connection_error") + return None, False + + if status_code in (401, 403): + # Injected-token model: no re-login here, just flag it for the caller/SaaS side. + self.needs_reauth = True + self.log("Warn: Sunsynk: Authentication error with status code {} from {}".format(status_code, url)) + self.failures_total += 1 + record_api_call("sunsynk", False, "auth_error") + return None, False + + if status_code in (200, 201): + if data is None: + data = {} + if isinstance(data, dict) and data.get("success") is False: + # Sunsynk's API family returns HTTP 200 with {"success": false} on failure + # (e.g. stale token, invalid request) rather than a non-2xx status - treat it + # the same as a hard failure so a stale-token 200 doesn't silently become an + # empty "data" payload that gets cached as if it were fresh. Mirrors deye.py's + # `if not data.get("success", True)` discipline. + self.failures_total += 1 + self.log("Warn: Sunsynk: API call to {} returned success=false (code {}): {}".format(url, data.get("code"), data.get("msg"))) + record_api_call("sunsynk", False, "api_success_false") + return None, False + self.needs_reauth = False + self.update_success_timestamp() + record_api_call("sunsynk") + return data, False + + self.failures_total += 1 + if status_code == 429: + # Rate limiting so wait up to 30 seconds and allow a retry + self.log("Info: Sunsynk: Rate limiting detected, waiting...") + record_api_call("sunsynk", False, "rate_limit") + await asyncio.sleep(random.random() * 30 + 1) + return None, True + record_api_call("sunsynk", False, "server_error") + return None, False + + async def get_device_list(self): + """ + Discover the inverter device list for the account, scoped to a single plant. + + Predbat.com's SaaS connect handler already rejects multi-plant Sunsynk accounts at + onboarding, so multi-plant support is a v1 non-goal here - this always resolves to + ONE plant. If ``self.plant_id`` is configured, it is used directly; otherwise the + account's plants are listed and the sole plant used (a warning is logged and the + first plant used defensively if more than one comes back). + + Real inverters response shape (GET /api/v1/plant/{id}/inverters): + {"code": 0, "success": true, "data": {"pageSize": 50, "total": 2, "infos": [ + {"id": 260047, "sn": "2504040106", "model": "SUN-50KW-SG01HP3-EU-BM4", "status": 1}, + {"id": 260046, "sn": "2504040164", "model": "SUN-50KW-SG01HP3-EU-BM4", "status": 1} + ]}} + + Returns a list of ``{"sn": ..., "model": ..., "plant_id": ...}`` dicts, filtered by + ``self.inverter_sn_filter`` when set. Returns None on API/discovery failure - any + existing cached list is left untouched so callers can keep using it. + """ + if self.device_list and not self._needs_refresh("device_list", SUNSYNK_REFRESH_STATIC): + return self.device_list + + plant_id = self.plant_id + if not plant_id: + plant_id = await self._discover_plant_id() + if plant_id is None: + return None + + query = {"page": 1, "limit": 50, "status": -1, "type": -2} + result = await self.request_get("/api/v1/plant/{}/inverters".format(plant_id), datain=query) + if result is None: + return None + + infos = (result.get("data") or {}).get("infos", []) or [] + devices = [{"sn": info.get("sn"), "model": info.get("model"), "plant_id": plant_id} for info in infos] + if self.inverter_sn_filter: + devices = [device for device in devices if device.get("sn") in self.inverter_sn_filter] + + self.device_list = devices + await self._save_cache("device_list", devices) + return devices + + async def _discover_plant_id(self): + """ + Resolve the account's single plant id by listing plants. + + Real response shape (GET /api/v1/plants): + {"code": 0, "success": true, "data": {"infos": [ + {"id": 505825, "name": "Virtual power station", "status": 1} + ]}} + + Returns None on API failure or when the account has no plants. Logs a warning and + uses the first plant when more than one comes back - the SaaS connect handler already + rejects multi-plant accounts at onboarding, so this is a defensive fallback only. + """ + result = await self.request_get("/api/v1/plants", datain={"page": 1, "limit": 100}) + if result is None: + return None + + infos = (result.get("data") or {}).get("infos", []) or [] + if not infos: + self.log("Warn: Sunsynk: No plants found for this account") + return None + if len(infos) > 1: + self.log("Warn: Sunsynk: {} plants found for this account - multi-plant is not supported, using the first".format(len(infos))) + return infos[0].get("id") + + async def get_plant_flow(self, plant_id): + """ + Get the live power-flow snapshot for a plant (PV/battery/grid/load). + + Real response shape (GET /api/v1/plant/energy/{id}/flow): + {"code": 0, "success": true, "data": { + "pvPower": 0, "battPower": 400, "gridOrMeterPower": 3179, "loadOrEpsPower": 3164, + "soc": 100, "batTo": true, "toBat": false, "gridTo": true, "toGrid": false, + "toLoad": true, "pvTo": false + }} + + Returns the parsed 'data' dict, or None on API failure. + """ + date_str = self.midnight_utc.strftime(SUNSYNK_DATE_FORMAT) + return await self._get_realtime("flow", plant_id, "/api/v1/plant/energy/{}/flow".format(plant_id), datain={"date": date_str}) + + async def get_battery(self, sn): + """ + Get real-time battery telemetry for an inverter. + + Real response shape (GET /api/v1/inverter/battery/{sn}/realtime): + {"code": 0, "success": true, "data": { + "power": 170, "capacity": "100.0", "soc": "100.0", "temp": "17.0", "voltage": "639.8" + }} + + Returns the parsed 'data' dict, or None on API failure. + """ + return await self._get_realtime("battery", sn, "/api/v1/inverter/battery/{}/realtime".format(sn), datain={"sn": sn, "lan": "en"}) + + async def get_grid(self, sn): + """ + Get real-time grid telemetry for an inverter. + + Returns the parsed 'data' dict, or None on API failure. + """ + return await self._get_realtime("grid", sn, "/api/v1/inverter/grid/{}/realtime".format(sn), datain={"sn": sn, "lan": "en"}) + + async def get_input(self, sn): + """ + Get real-time PV/input telemetry for an inverter. + + Returns the parsed 'data' dict, or None on API failure. + """ + return await self._get_realtime("input", sn, "/api/v1/inverter/{}/realtime/input".format(sn)) + + async def get_output(self, sn): + """ + Get real-time output telemetry for an inverter. + + Returns the parsed 'data' dict, or None on API failure. + """ + return await self._get_realtime("output", sn, "/api/v1/inverter/{}/realtime/output".format(sn)) + + async def _get_realtime(self, bucket, cache_key, path, datain=None): + """ + Shared cache-and-fetch logic for the realtime telemetry endpoints (flow/battery/grid/ + input/output). + + Freshness is tracked PER bucket+cache_key (e.g. "device_values:battery:SN1"), NOT via + one shared "device_values" clock - a single shared clock would mean fetching any + bucket/sn resets the timestamp for every other bucket/sn, so a genuinely-stale entry + for a DIFFERENT key would be served from cache without ever being fetched again. Returns + the cached value for ``cache_key`` within ``device_values[bucket]`` when it is already + populated and still within SUNSYNK_REFRESH_REALTIME for THIS bucket+cache_key pair; + otherwise polls the API, updates the in-memory bucket and persists the whole + device_values cache. Returns None on API failure, leaving any existing cached value + untouched. + """ + store = self.device_values[bucket] + age_key = self._realtime_age_key(bucket, cache_key) + if cache_key in store and not self._needs_refresh(age_key, SUNSYNK_REFRESH_REALTIME): + return store[cache_key] + + result = await self.request_get(path, datain=datain) + if result is None: + return None + + data = result.get("data") or {} + store[cache_key] = data + self.data_age[age_key] = datetime.now(timezone.utc) + # Persists the whole in-memory device_values struct (all buckets/sns together) so a + # reboot restores everything at once; _save_cache() also stamps its own "device_values" + # bookkeeping key, which is unrelated to (and must not be confused with) the per-entry + # age_key freshness tracked above. + await self._save_cache("device_values", self.device_values) + return data + + @staticmethod + def _realtime_age_key(bucket, cache_key): + """Return the per-entry data_age key for a realtime telemetry bucket+cache_key pair.""" + return "device_values:{}:{}".format(bucket, cache_key) + + def _data_age_minutes(self, key): + """ + Return the age in minutes of the in-memory data for a cache key, or None if unknown. + """ + timestamp = self.data_age.get(key, None) + if timestamp is None: + return None + return (datetime.now(timezone.utc) - timestamp).total_seconds() / 60.0 + + def _needs_refresh(self, key, max_age_minutes): + """ + Return True if the data for a cache key is missing or older than max_age_minutes. + """ + age = self._data_age_minutes(key) + return age is None or age >= max_age_minutes + + async def _save_cache(self, key, data): + """ + Save data to storage under the sunsynk module and record its update time in memory. + """ + now = datetime.now(timezone.utc) + self.data_age[key] = now + if self.storage: + # Expire after a day so stale data doesn't linger in the cache forever + await self.storage.save("sunsynk", key, data, format="json", expiry=now + timedelta(days=1)) + + async def _load_cache(self, key): + """ + Load cached data for a key from storage, recording its age. Returns None if absent. + """ + if not self.storage: + return None + data = await self.storage.load("sunsynk", key) + if data is None: + return None + age = await self.storage.age("sunsynk", key) + if age is None: + return None + self.data_age[key] = datetime.now(timezone.utc) - timedelta(minutes=age) + return data + + async def load_cached_data(self): + """ + Restore cached device data (device_list, device_values) from storage on startup. + + Mirrors fox.py's load_cached_data(): reads back each category of data _save_cache() + previously persisted, recording its age in self.data_age via _load_cache() so the + age-based refresh logic (_needs_refresh()) can decide whether a fresh API poll is + still needed. Without this, a Predbat restart silently discarded the persisted cache + and cold-re-polled the Sunsynk Connect API on the very next cycle - no reboot-storm + protection. device_settings is a listed cache key (SUNSYNK_CACHE_KEYS) but is not yet + written by _save_cache() anywhere (device settings read/write land in a later task), + so it is deliberately not restored here either - restoring it now would be a no-op. + """ + if not self.storage: + return + + device_list = await self._load_cache("device_list") + if device_list is not None: + self.device_list = device_list + + device_values = await self._load_cache("device_values") + if device_values is not None: + self.device_values = device_values + + self.log("Sunsynk: Restored cached data from storage for keys {}".format(sorted(self.data_age.keys()))) + + async def publish_data(self): + """ + Publish Sunsynk telemetry as Predbat entities via dashboard_item(), fox.py-style. + + The plant-level flow endpoint (get_plant_flow) is PLANT-AGGREGATED - in a + multi-inverter site it already sums every inverter's contribution. Publishing its + values under EACH inverter's SN would double-count a multi-inverter site (Task 10 + brief, Codex Critical 3), so this method branches on the device count: + + - Single-inverter plant: plant flow IS that one inverter's data, so it is safe to + derive battery_power/grid_power/battery_soc/pv_power/load_power directly from + flow's VERIFIED direction-boolean sign mapping (batTo/toGrid - Risk 5, confirmed + against live data at the plant level). + - Multi-inverter plant: per-SN entities instead come from the per-SN battery/grid/ + input/output endpoints (see _publish_multi_inverter_sn()), including load_power + (sourced from the per-SN output endpoint's AC output, NOT plant flow's + loadOrEpsPower, which is plant-aggregate and would double-count here too) - and a + single plant-level summary (sensor.{prefix}_sunsynk_site_*) is published from flow + separately, since that one is plant-aggregate by design and isn't attributed to + any single SN. + + battery_temperature and soc_max (from battery.capacity) always come from the per-SN + battery/{sn}/realtime endpoint in both branches, since flow carries neither. + """ + devices = self.device_list + if not devices: + self.log("Warn: Sunsynk: publish_data() called with no discovered devices - nothing to publish") + return + + plant_id = devices[0].get("plant_id") + flow = await self.get_plant_flow(plant_id) if plant_id else None + + entity_name_sensor = f"sensor.{self.prefix}_sunsynk" + + if len(devices) == 1: + sn = devices[0].get("sn") + if sn: + await self._publish_single_inverter(entity_name_sensor, sn, flow) + else: + self.log("Warn: Sunsynk: skipping device with missing SN: {}".format(devices[0])) + else: + for device in devices: + sn = device.get("sn") + if sn: + await self._publish_multi_inverter_sn(entity_name_sensor, sn) + else: + self.log("Warn: Sunsynk: skipping device with missing SN: {}".format(device)) + if flow is not None: + self._publish_site_summary(entity_name_sensor, flow) + else: + self.log("Warn: Sunsynk: site - no plant flow data available, skipping site summary") + + async def _publish_single_inverter(self, entity_name_sensor, sn, flow): + """ + Publish one SN's entities for a SINGLE-inverter plant. + + battery_power/grid_power/battery_soc/pv_power/load_power are derived from the plant + flow endpoint using the VERIFIED sign mapping - safe here because the plant's flow + total IS this one inverter's data (no double-counting risk with only one inverter). + """ + if flow is not None: + battery_power, grid_power = self._flow_signed_powers(flow) + + self._publish_entity(entity_name_sensor, sn, "battery_power", battery_power, "W", "power", "mdi:battery-charging") + self._publish_entity(entity_name_sensor, sn, "grid_power", grid_power, "W", "power", "mdi:transmission-tower") + self._publish_entity(entity_name_sensor, sn, "battery_soc", flow.get("soc"), "%", "battery", "mdi:battery-50") + self._publish_entity(entity_name_sensor, sn, "pv_power", flow.get("pvPower"), "W", "power", "mdi:solar-power") + self._publish_entity(entity_name_sensor, sn, "load_power", flow.get("loadOrEpsPower"), "W", "power", "mdi:home-lightning-bolt") + else: + self.log("Warn: Sunsynk: {} - no plant flow data available, skipping flow-derived entities".format(sn)) + + await self._publish_battery_extras(entity_name_sensor, sn) + + async def _publish_multi_inverter_sn(self, entity_name_sensor, sn): + """ + Publish one SN's entities for a MULTI-inverter plant, from the PER-SN battery/grid/ + input/output endpoints rather than plant flow (see publish_data() docstring for why). + + # TODO(risk5-perSN): the per-SN battery.power / grid power sign convention is + # UNCONFIRMED. Risk 5 was only verified at the plant-flow level (the batTo/toGrid + # direction booleans) - the demo plant used to develop this was near-idle, so these + # per-SN fields' sign couldn't be cross-checked against a known charge/discharge or + # import/export state. Confirm against a live, non-idle multi-inverter plant before + # relying on these signs for control decisions. + + # TODO(risk6-perSN-load): per-SN load source unconfirmed. There is no per-SN "load" + # endpoint - loadOrEpsPower on plant flow is plant-aggregate and would double-count + # here (see publish_data() docstring), so load_power is instead derived from the + # per-SN inverter/{sn}/realtime/output endpoint's AC output (pac/pInv), on the + # assumption that an inverter's AC output IS its load contribution. That assumption, + # and the endpoint's shape, are both unconfirmed against a live multi-inverter plant. + """ + self.log("Warn: Sunsynk: {} - per-SN battery/grid power sign is UNCONFIRMED (only plant-level flow signs are verified) - treat with caution until confirmed against live multi-inverter data".format(sn)) + + battery = await self.get_battery(sn) + if battery is not None: + self._publish_entity(entity_name_sensor, sn, "battery_power", self._to_float(battery.get("power")), "W", "power", "mdi:battery-charging") + self._publish_entity(entity_name_sensor, sn, "battery_soc", self._to_float(battery.get("soc")), "%", "battery", "mdi:battery-50") + + await self._publish_battery_extras(entity_name_sensor, sn, battery) + + grid = await self.get_grid(sn) + if grid is not None: + self._publish_entity(entity_name_sensor, sn, "grid_power", self._sum_phase_power(grid, ("vip", "meters", "phases"), ("power", "pac", "gridPower")), "W", "power", "mdi:transmission-tower") + + pv_input = await self.get_input(sn) + if pv_input is not None: + self._publish_entity(entity_name_sensor, sn, "pv_power", self._sum_phase_power(pv_input, ("pv", "pvStrings", "strings"), ("power", "pac", "pvPower")), "W", "power", "mdi:solar-power") + + # TODO(risk6-perSN-load): per-SN load source unconfirmed - see docstring above. + self.log("Warn: Sunsynk: {} - per-SN load_power source (inverter AC output) is UNCONFIRMED (not one of the plant-flow-verified quantities) - treat with caution until confirmed against live multi-inverter data".format(sn)) + output = await self.get_output(sn) + if output is not None: + self._publish_entity(entity_name_sensor, sn, "load_power", self._sum_phase_power(output, ("vip", "phases"), ("power", "pac", "pInv")), "W", "power", "mdi:home-lightning-bolt") + + async def _publish_battery_extras(self, entity_name_sensor, sn, battery=_UNFETCHED): + """ + Publish battery_temperature and soc_max (from battery.capacity) for one SN. + + Always sourced from the per-SN battery/{sn}/realtime endpoint - plant flow carries + neither field. Used by both the single- and multi-inverter publish paths; the + multi-inverter path passes its already-fetched battery dict in (which may legitimately + be None if that fetch already failed) to avoid a duplicate request, the single-inverter + path fetches it here since it has no other need for it. + + ``battery`` defaults to the module-level ``_UNFETCHED`` sentinel rather than ``None``, + so a genuinely-unfetched battery (lazy-fetch here) can be told apart from an + already-fetched-and-failed one (``None``, passed explicitly by the multi-inverter + path) - a plain ``None`` default can't distinguish the two, which meant a battery + endpoint that had already failed this cycle was silently re-fetched a second time. + """ + if battery is _UNFETCHED: + battery = await self.get_battery(sn) + if battery is None: + return + self._publish_entity(entity_name_sensor, sn, "battery_temperature", self._to_float(battery.get("temp")), "°C", "temperature", "mdi:thermometer") + self._publish_entity(entity_name_sensor, sn, "soc_max", self._to_float(battery.get("capacity")), "Ah", None, "mdi:battery-high") + + def _publish_site_summary(self, entity_name_sensor, flow): + """ + Publish a single plant-level summary (sensor.{prefix}_sunsynk_site_*) for a MULTI- + inverter plant, from the plant flow endpoint using the VERIFIED sign mapping. + + This is the one place flow's plant-aggregate values are safe to publish directly in + a multi-inverter plant - it is attributed to "site", not to any single inverter's SN, + so there is no double-counting risk (unlike publishing it per-SN would be). + """ + battery_power, grid_power = self._flow_signed_powers(flow) + + self._publish_entity(entity_name_sensor, "site", "battery_power", battery_power, "W", "power", "mdi:battery-charging") + self._publish_entity(entity_name_sensor, "site", "grid_power", grid_power, "W", "power", "mdi:transmission-tower") + self._publish_entity(entity_name_sensor, "site", "battery_soc", flow.get("soc"), "%", "battery", "mdi:battery-50") + self._publish_entity(entity_name_sensor, "site", "pv_power", flow.get("pvPower"), "W", "power", "mdi:solar-power") + self._publish_entity(entity_name_sensor, "site", "load_power", flow.get("loadOrEpsPower"), "W", "power", "mdi:home-lightning-bolt") + + def _publish_entity(self, entity_name_sensor, sn, leaf, state, unit, device_class, icon): + """ + Publish one sensor.{prefix}_sunsynk_{sn}_{leaf} entity via dashboard_item(), using + fox.py/solis.py-style attributes (friendly_name/unit_of_measurement/device_class/ + state_class/icon). + """ + entity_id = "{}_{}_{}".format(entity_name_sensor, sn.lower(), leaf) + attributes = { + "friendly_name": "Sunsynk {} {}".format(sn, leaf.replace("_", " ").title()), + "unit_of_measurement": unit, + "icon": icon, + } + if device_class: + attributes["device_class"] = device_class + attributes["state_class"] = "measurement" + self.dashboard_item(entity_id, state=state, attributes=attributes, app="sunsynk") + + @staticmethod + def _flow_signed_powers(flow): + """ + Apply the VERIFIED plant-flow sign mapping to battery_power/grid_power. + + battery_power = battPower if batTo else -battPower (discharge +ve, charge -ve); + grid_power = gridOrMeterPower if toGrid else -gridOrMeterPower (predbat convention: + export +ve, import -ve). This is Risk 5's verified direction-boolean mapping - the + single source of truth for it, since a sign error here is the highest-stakes mistake + in this module. Used by both _publish_single_inverter() (flow IS that one inverter's + data) and _publish_site_summary() (flow attributed to "site", not any inverter's SN). + + Returns a (battery_power, grid_power) tuple. + """ + batt_power_raw = flow.get("battPower", 0) or 0 + battery_power = batt_power_raw if flow.get("batTo") else -batt_power_raw + grid_power_raw = flow.get("gridOrMeterPower", 0) or 0 + grid_power = grid_power_raw if flow.get("toGrid") else -grid_power_raw + return battery_power, grid_power + + @staticmethod + def _to_float(value): + """Best-effort float conversion, returning None (rather than raising) on bad input.""" + try: + return float(value) + except (TypeError, ValueError): + return None + + @staticmethod + def _sum_phase_power(data, list_keys, flat_keys): + """ + Sum a 'power' field across a per-phase/per-string breakdown, falling back to the + first present flat field when no breakdown list is found. + + # TODO(risk5-perSN): shape AND sign are UNCONFIRMED for the per-SN grid/input + # endpoints (see _publish_multi_inverter_sn's TODO) - this is defensive parsing + # only, not verified against real API responses. + """ + for key in list_keys: + entries = data.get(key) + if isinstance(entries, list) and entries: + total = 0.0 + counted = False + for entry in entries: + if isinstance(entry, dict) and "power" in entry: + power = SunsynkAPI._to_float(entry.get("power")) + if power is not None: + total += power + counted = True + if counted: + return total + for flat_key in flat_keys: + if flat_key in data: + value = SunsynkAPI._to_float(data.get(flat_key)) + if value is not None: + return value + return None + + async def automatic_config(self): + """ + Automatically configure Predbat's base args to wire up the discovered Sunsynk + inverters, fox.py/deye.py/solis.py-style. + + Sensor entity ids MUST exactly match what publish_data() (Task 10) actually publishes + - sensor.{prefix}_sunsynk_{sn}_{leaf}, SN lower-cased - a mismatch here means Predbat + reads an entity publish_data() never writes, which is a silent broken-wiring bug (not + a crash). The charge/discharge slot control entities (select/number/switch.{prefix}_ + sunsynk_{sn}_*) don't exist yet - device control lands in a later task - this just + names them here so they are wired up the moment they appear, mirroring how fox.py/ + deye.py/solis.py all name control entities ahead of the write path landing. + + battery_power/grid_power carry the VERIFIED sign (predbat convention: discharge/ + export +ve - see _flow_signed_powers()) ONLY for a single-inverter plant and for the + multi-inverter site-level summary - both are sourced from plant flow's batTo/toGrid + direction booleans, the one part of Task 10's sign mapping that was actually verified + against live data (Risk 5). So - unlike solis.py's battery_power_invert or fox.py's + grid_power_invert - no invert flag is needed for THOSE two cases. For a MULTI-inverter + plant, though, the per-SN battery_power/grid_power this method wires up instead come + from the per-SN battery/grid endpoints (_publish_multi_inverter_sn()), whose sign is + UNCONFIRMED (Task 10's open risk5/risk6 - see that method's TODOs); this + automatic_config wiring inherits that same open risk for per-SN entities, since it + only names whatever publish_data() actually publishes under those ids without + re-verifying their sign - do not treat per-SN battery_power/grid_power as verified. + + Callers must gate this on self.automatic themselves, mirroring fox.py/deye.py/ + solis.py's ``if first and self.automatic: await self.automatic_config()`` in run() - + this method does not check it itself. run() (Task 12) performs this gating, and also + re-invokes automatic_config() on any cycle where the discovered device set changes + (not just the first run), so a newly-added inverter gets wired up without a restart. + + Emits a single structured Info: log line with the complete arg map (spec section 3.1) + so the automatic-config outcome is Loki-greppable without reconstructing it from many + individual set_arg log lines. + """ + devices_raw = await self.get_device_list() + if not devices_raw: + self.log("Warn: SunsynkAPI automatic_config found no inverters") + return + + devices = [] + for device in devices_raw: + sn = device.get("sn") + if not sn: + self.log("Warn: SunsynkAPI automatic_config: skipping device with missing SN: {}".format(device)) + continue + devices.append(sn.lower()) + + if not devices: + self.log("Warn: SunsynkAPI automatic_config found no inverters") + return + + arg_map = {"inverter_type": ["SUNSYNK" for _ in devices], "num_inverters": len(devices)} + + # Telemetry sensors - MUST match publish_data()'s sensor.{prefix}_sunsynk_{sn}_{leaf} + # ids exactly (Task 10). + arg_map["soc_percent"] = [f"sensor.{self.prefix}_sunsynk_{sn}_battery_soc" for sn in devices] + arg_map["battery_power"] = [f"sensor.{self.prefix}_sunsynk_{sn}_battery_power" for sn in devices] + arg_map["grid_power"] = [f"sensor.{self.prefix}_sunsynk_{sn}_grid_power" for sn in devices] + arg_map["load_power"] = [f"sensor.{self.prefix}_sunsynk_{sn}_load_power" for sn in devices] + if not self.automatic_ignore_pv: + arg_map["pv_power"] = [f"sensor.{self.prefix}_sunsynk_{sn}_pv_power" for sn in devices] + arg_map["battery_temperature"] = [f"sensor.{self.prefix}_sunsynk_{sn}_battery_temperature" for sn in devices] + arg_map["soc_max"] = [f"sensor.{self.prefix}_sunsynk_{sn}_soc_max" for sn in devices] + + # Reserve / minimum SOC control - one entity serves both keys, mirroring solis.py's + # dual use of its over_discharge_soc control for both "reserve" and "battery_min_soc" + # (there is no separate read-only min-soc sensor). Not published yet - lands with + # device control (Task 15). + arg_map["battery_min_soc"] = [f"number.{self.prefix}_sunsynk_{sn}_battery_min_soc" for sn in devices] + arg_map["reserve"] = [f"number.{self.prefix}_sunsynk_{sn}_battery_min_soc" for sn in devices] + + # Charge/discharge controls - slot 1 of the 6-slot TOU model carries Predbat's primary + # control, mirroring solis.py's charge_slot1_*/discharge_slot1_* leaf naming exactly. + arg_map["charge_start_time"] = [f"select.{self.prefix}_sunsynk_{sn}_charge_slot1_start_time" for sn in devices] + arg_map["charge_end_time"] = [f"select.{self.prefix}_sunsynk_{sn}_charge_slot1_end_time" for sn in devices] + arg_map["charge_limit"] = [f"number.{self.prefix}_sunsynk_{sn}_charge_slot1_soc" for sn in devices] + arg_map["charge_rate"] = [f"number.{self.prefix}_sunsynk_{sn}_charge_slot1_power" for sn in devices] + arg_map["scheduled_charge_enable"] = [f"switch.{self.prefix}_sunsynk_{sn}_charge_slot1_enable" for sn in devices] + + arg_map["discharge_start_time"] = [f"select.{self.prefix}_sunsynk_{sn}_discharge_slot1_start_time" for sn in devices] + arg_map["discharge_end_time"] = [f"select.{self.prefix}_sunsynk_{sn}_discharge_slot1_end_time" for sn in devices] + arg_map["discharge_target_soc"] = [f"number.{self.prefix}_sunsynk_{sn}_discharge_slot1_soc" for sn in devices] + arg_map["discharge_rate"] = [f"number.{self.prefix}_sunsynk_{sn}_discharge_slot1_power" for sn in devices] + arg_map["scheduled_discharge_enable"] = [f"switch.{self.prefix}_sunsynk_{sn}_discharge_slot1_enable" for sn in devices] + + for key, value in arg_map.items(): + self.set_arg(key, value) + + self.log(f"Info: SunsynkAPI automatic_config: {json.dumps(arg_map)}") + + async def run(self, seconds, first): + """ + Main run loop, fox.py/deye.py-style: refresh the discovered device list, publish + telemetry, and (re)wire automatic_config when appropriate. + + Device discovery is polled via get_device_list() every cycle rather than gated here + on an explicit interval check, because get_device_list() already caches in-memory + (self.device_list, keyed off SUNSYNK_REFRESH_STATIC) and only hits the API once that + cache has actually gone stale - so calling it unconditionally here is a cheap no-op + on most cycles, and it also means automatic_config()'s own get_device_list() call + later in the SAME cycle is served from the cache this call just (re)populated, rather + than firing a second API request for the same data. + + Mirrors fox.py's run() in checking self.device_list (not get_device_list()'s return + value) for the "no devices" failure path - get_device_list() returns None on API + failure but leaves any existing cached self.device_list untouched, so a transient API + error does not wrongly report "no devices" when a cached list from a previous cycle is + still available to publish from. + + automatic_config() is gated on self.automatic (it does not check it itself - see its + own docstring) and is invoked on the first run OR whenever the set of discovered + inverter SNs has changed since the last cycle (e.g. an inverter was added to or removed + from the Sunsynk account), so newly-discovered inverters get wired up without requiring + a Predbat restart. + + On the first cycle, cached device data is restored from storage (load_cached_data()) + BEFORE the first get_device_list() call, fox.py-style, so a quick restart resumes from + the persisted cache rather than cold-re-polling the Sunsynk API immediately. + """ + if first: + await self.load_cached_data() + + result = await self.get_device_list() + if result: + self.log("Sunsynk: API Found {} devices".format(len(result))) + + if not self.device_list: + self.log("Error: SunsynkAPI: No devices found, unable to publish or configure") + return False + + current_sns = {device.get("sn") for device in self.device_list} + device_set_changed = current_sns != self._known_sns + self._known_sns = current_sns + + # publish_data()/automatic_config() are each wrapped so one bad cycle (e.g. a transient + # parsing error on a malformed API response) doesn't abort the whole run() and take the + # rest of the component down with it - mirrors deye.py's per-poll error-wrapping + # discipline (deye.py run()). + try: + await self.publish_data() + except Exception as e: + self.log(f"Warn: SunsynkAPI publish_data failed: {e}") + + if self.automatic and (first or device_set_changed): + try: + await self.automatic_config() + except Exception as e: + self.log(f"Warn: SunsynkAPI automatic_config failed: {e}") + + return True + + +class MockBase: # pragma: no cover + """Mock base class for standalone testing""" + + def __init__(self): + """Initialise minimal base state (timezone, prefix, args, entities).""" + self.local_tz = datetime.now().astimezone().tzinfo + self.now_utc = datetime.now(self.local_tz) + self.prefix = "predbat" + self.args = {} + self.midnight_utc = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + self.minutes_now = self.now_utc.hour * 60 + self.now_utc.minute + self.entities = {} + + def get_state_wrapper(self, entity_id, default=None, attribute=None, refresh=False, required_unit=None, raw=None): + """Return a stored entity's state (or the full record when raw=True).""" + if raw: + return self.entities.get(entity_id, {}) + else: + return self.entities.get(entity_id, {}).get("state", default) + + def set_state_wrapper(self, entity_id, state, attributes=None, app=None): + """Store an entity's state and attributes.""" + self.entities[entity_id] = {"state": state, "attributes": attributes or {}} + + def log(self, message): + """Print a timestamped log message.""" + print(f"[{datetime.now().strftime('%H:%M:%S')}] {message}") + + def dashboard_item(self, entity_id, state=None, attributes=None, app=None): + """Print and store a dashboard entity update.""" + print(f"ENTITY: {entity_id} = {state}") + if attributes: + if "options" in attributes: + attributes["options"] = "..." + print(f" Attributes: {json.dumps(attributes, indent=2)}") + self.set_state_wrapper(entity_id, state, attributes) + + def get_arg(self, key, default=None): + """Return a config argument (always the default - no real config backing).""" + return default + + def set_arg(self, key, value): + """Print a config argument write for debugging, and record it in self.args - matching + the real base's set_arg() (userinterface.py), which does `self.args[arg] = value` - + so automatic_config() can be asserted against api.base.args in tests.""" + state = None + if isinstance(value, str) and "." in value: + state = self.get_state_wrapper(value, default=None) + elif isinstance(value, list): + state = "n/a []" + for v in value: + if isinstance(v, str) and "." in v: + state = self.get_state_wrapper(v, default=None) + break + else: + state = "n/a" + print(f"Set arg {key} = {value} (state={state})") + self.args[key] = value + + +def _sunsynk_nonce(): # pragma: no cover + """Return a millisecond-precision timestamp nonce, as Sunsynk's login flow requires.""" + return int(time.time() * 1000) + + +def _sunsynk_md5_hex(value): + """Return the hex MD5 digest of a string. + + Sunsynk's login flow uses MD5 request signing for API-compatibility with the official + Sunsynk Connect web client - it is not a security control. + """ + return hashlib.md5(value.encode()).hexdigest() + + +def _sunsynk_rsa_encrypt_password(raw_key, password): + """RSA-PKCS1v15 encrypt a password with Sunsynk's publicKey response body, base64-encoded. + + This is the ONE place in this module that needs the ``cryptography`` package, and the + import is guarded here (not at module level) so importing sunsynk.py as a SaaS component + - which only ever does plain Bearer requests with a server-injected token, never its own + login - does not require the dependency. Only the standalone CLI harness's login path + (HA add-on / self-managed users) calls this. Raises RuntimeError with an actionable + message if ``cryptography`` isn't installed. + """ + try: + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric import padding + except ImportError as exc: + raise RuntimeError("standalone Sunsynk auth needs `pip install cryptography`") from exc + + pem = "-----BEGIN PUBLIC KEY-----\n{}\n-----END PUBLIC KEY-----".format(raw_key).encode() + public_key = serialization.load_pem_public_key(pem) + encrypted = public_key.encrypt(password.encode(), padding.PKCS1v15()) + return base64.b64encode(encrypted).decode() + + +def _sunsynk_build_login_payload(raw_key, username, password, nonce=None): + """Build the ``/oauth/token/new`` JSON payload for standalone Sunsynk login. + + Split out from ``_sunsynk_standalone_login()`` so the payload/sign construction can be + unit tested against a fixed ``raw_key``/``nonce`` without a live network call - see + ``tests/test_sunsynk.py``. ``client_id``/``source`` are Sunsynk's public constants (not + per-developer credentials), matching ``scratchpad/sunsynk_auth_spike.py``'s verified flow. + """ + if nonce is None: + nonce = _sunsynk_nonce() # pragma: no cover + return { + "username": username, + "password": _sunsynk_rsa_encrypt_password(raw_key, password), + "grant_type": "password", + "client_id": "csp-web", + "source": "sunsynk", + "nonce": nonce, + "sign": _sunsynk_md5_hex("nonce={}&source=sunsynk{}".format(nonce, raw_key[:10])), + } + + +def _sunsynk_standalone_login(username, password, base_url=SUNSYNK_DOMAIN): # pragma: no cover + """ + Perform the standalone Sunsynk RSA login (publicKey -> RSA-encrypt password -> + ``/oauth/token/new``), for HA add-on / self-managed users who supply a Sunsynk Connect + username+password directly - see the module docstring for why this is the ONE place + batpred does its own login (SaaS mode never does; the edge function owns that chain + server-side and injects a plain Bearer token instead). Mirrors + ``scratchpad/sunsynk_auth_spike.py``'s verified flow exactly. + + Synchronous (plain ``requests``, already a core Predbat dependency) rather than + ``aiohttp`` - this is a one-off CLI-harness login step outside the async component's + event loop, not part of ``SunsynkAPI`` itself. + + Returns the access token string. Raises RuntimeError on any failure (missing + ``cryptography``, HTTP error, or an unsuccessful/malformed API response) so the CLI + harness can print a clear error instead of a confusing traceback. + """ + key_nonce = _sunsynk_nonce() + key_sign = _sunsynk_md5_hex("nonce={}&source=sunsynkPOWER_VIEW".format(key_nonce)) + key_response = requests.get( + "{}/anonymous/publicKey".format(base_url), + params={"nonce": key_nonce, "source": "sunsynk", "sign": key_sign}, + headers={"Content-Type": "application/json"}, + timeout=TIMEOUT, + ) + key_response.raise_for_status() + key_body = key_response.json() + if not key_body.get("success") or not key_body.get("data"): + raise RuntimeError("Sunsynk publicKey request failed: {}".format(key_body)) + raw_key = key_body["data"] + + payload = _sunsynk_build_login_payload(raw_key, username, password) + login_response = requests.post("{}/oauth/token/new".format(base_url), headers={"Content-Type": "application/json"}, json=payload, timeout=TIMEOUT) + login_body = login_response.json() + if not (login_response.status_code == 200 and login_body.get("success")): + raise RuntimeError("Sunsynk login failed (HTTP {}): {}".format(login_response.status_code, login_body)) + + access_token = (login_body.get("data") or {}).get("access_token") + if not access_token: + raise RuntimeError("Sunsynk login response missing access_token: {}".format(login_body)) + return access_token + + +async def _test_sunsynk_api(args): # pragma: no cover + """ + Run one read-only Sunsynk cycle for the CLI harness: log in (standalone RSA, unless + ``--token`` was given directly), discover devices, and - with ``--plants`` - publish and + dump entities via the MockBase, fox.py/deye.py harness-style. + """ + mock_base = MockBase() + + if args.token: + access_token = args.token + else: + print("Logging in to Sunsynk Connect as {}...".format(args.user)) + access_token = _sunsynk_standalone_login(args.user, args.password, base_url=args.base_url) + print("Login OK - access token acquired") + + api = SunsynkAPI(mock_base, key=access_token, automatic=True, plant_id=args.plant_id) + + devices = await api.get_device_list() + if not devices: + print("No devices found") + return + print("Found {} device(s):".format(len(devices))) + for device in devices: + print(" - {}".format(json.dumps(device))) + + if args.plants: + print("Publishing entities...") + await api.publish_data() + print("Published entities:") + for entity_id, record in mock_base.entities.items(): + print(" {} = {}".format(entity_id, record.get("state"))) + + +def main(): # pragma: no cover + """ + Command-line entry point for the standalone Sunsynk CLI harness (HA add-on / self-managed + users): logs in with a username+password via the standalone RSA flow (or reuses an + existing ``--token``), then lists discovered devices and - with ``--plants`` - publishes + and dumps entities. + """ + parser = argparse.ArgumentParser(description="Test the Sunsynk Connect API (read-only, standalone RSA login)") + parser.add_argument("--user", default=os.environ.get("SUNSYNK_USER"), help="Sunsynk Connect account username/email") + parser.add_argument("--pass", "--password", dest="password", default=os.environ.get("SUNSYNK_PASS"), help="Sunsynk Connect account password") + parser.add_argument("--token", default=None, help="Skip login and use an existing Bearer access token directly") + parser.add_argument("--plant-id", type=int, default=None, help="Sunsynk plant id (auto-discovered if omitted)") + parser.add_argument("--base-url", default=SUNSYNK_DOMAIN, help="Sunsynk Connect API base URL") + parser.add_argument("--plants", action="store_true", help="Also publish and dump entities (default: device list only)") + + args = parser.parse_args() + + if not args.token and not (args.user and args.password): + parser.error("provide either --token or both --user and --pass/--password") + + asyncio.run(_test_sunsynk_api(args)) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/apps/predbat/tests/test_sunsynk.py b/apps/predbat/tests/test_sunsynk.py new file mode 100644 index 000000000..40ec7cfa9 --- /dev/null +++ b/apps/predbat/tests/test_sunsynk.py @@ -0,0 +1,1315 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# Test Sunsynk API skeleton + Bearer request layer +# ----------------------------------------------------------------------------- + +"""Tests for the SunsynkAPI component (Tasks 8-13). + +Task 8 covers the injected-token Bearer request layer. Task 9 adds device discovery +(single-plant scoped) and read-only telemetry (flow/battery/grid/input/output). Task 10 adds +publish_data() - entity publishing with the VERIFIED sign mapping. Task 11 adds +automatic_config() - the set_arg map wiring Predbat to the entities publish_data() emits. +Task 12 adds run() plus COMPONENT_LIST/config.py registration. Task 13 adds the standalone +CLI harness's RSA login helpers (publicKey -> RSA-encrypt password -> token/new) - tested here +via the payload/sign construction and the guarded `cryptography` import, NOT a live call (the +harness itself is exercised manually against a real Sunsynk account). Control writes land in a +later task and are not tested here. +""" + +import base64 +import hashlib +import json +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, patch + +import sunsynk +from sunsynk import SunsynkAPI, MockBase, SUNSYNK_REFRESH_REALTIME +from tests.test_infra import run_async, create_aiohttp_mock_response, create_aiohttp_mock_session + +# Real captured demo responses (Task 9 brief) - used verbatim as test fixtures so the parsing +# logic is checked against the actual Sunsynk Connect API contract, not an invented shape. +PLANTS_RESPONSE = {"code": 0, "success": True, "data": {"infos": [{"id": 505825, "name": "Virtual power station", "status": 1}]}} + +INVERTERS_RESPONSE = { + "code": 0, + "success": True, + "data": { + "pageSize": 50, + "total": 2, + "infos": [ + {"id": 260047, "sn": "2504040106", "model": "SUN-50KW-SG01HP3-EU-BM4", "status": 1}, + {"id": 260046, "sn": "2504040164", "model": "SUN-50KW-SG01HP3-EU-BM4", "status": 1}, + ], + }, +} + +FLOW_RESPONSE = { + "code": 0, + "success": True, + "data": { + "pvPower": 0, + "battPower": 400, + "gridOrMeterPower": 3179, + "loadOrEpsPower": 3164, + "soc": 100, + "batTo": True, + "toBat": False, + "gridTo": True, + "toGrid": False, + "toLoad": True, + "pvTo": False, + }, +} + +BATTERY_RESPONSE = {"code": 0, "success": True, "data": {"power": 170, "capacity": "100.0", "soc": "100.0", "temp": "17.0", "voltage": "639.8"}} + + +class FakeStorage: + """In-memory fake of the Storage component, for _save_cache/_load_cache/load_cached_data + tests - mirrors test_sigenergy.py's FakeStorage.""" + + def __init__(self): + """Initialise an empty in-memory store keyed by (module, filename).""" + self.data = {} + + async def save(self, module, filename, data, format="yaml", expiry=None): + """Store data under (module, filename), mirroring StorageBase.save's signature and return value.""" + self.data[(module, filename)] = data + return True + + async def load(self, module, filename): + """Return previously saved data for (module, filename), or None if absent.""" + return self.data.get((module, filename)) + + async def age(self, module, filename): + """Return 0 (fresh) if data exists for (module, filename), else None.""" + return 0 if (module, filename) in self.data else None + + +class _FakeComponents: + """Minimal stand-in for Components, resolving get_component("storage") to a fixed FakeStorage - + lets a plain SunsynkAPI(MockBase(), ...) pick up a working storage via ComponentBase.storage's + real `self.base.components.get_component("storage")` lookup, without a bespoke Mock subclass.""" + + def __init__(self, storage): + """Bind the FakeStorage instance this fake registry will hand back.""" + self._storage = storage + + def get_component(self, name): + """Return the bound storage component for 'storage', None for anything else.""" + return self._storage if name == "storage" else None + + +def _sunsynk_api_with_storage(storage=None): + """Build a SunsynkAPI wired to a FakeStorage (shared if given) via base.components, for + _save_cache/_load_cache/load_cached_data tests. Returns (api, storage).""" + storage = storage if storage is not None else FakeStorage() + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.base.components = _FakeComponents(storage) + return api, storage + + +def test_get_headers_uses_bearer_token(): + """get_headers() returns a plain Bearer + JSON content-type header, no signing.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + h = api.get_headers() + assert h["Authorization"] == "Bearer tok123" + assert h["Content-Type"] == "application/json" + + +def test_initialize_stores_key_as_access_token_regardless_of_auth_method(): + """'key' is always the Bearer access token, even when auth_method isn't 'oauth'. + + _init_oauth() only populates self.access_token when auth_method == "oauth" (None + otherwise), but Sunsynk's injected-token model means 'key' IS the access token no + matter what auth_method is configured - so initialize() must re-apply it afterwards. + """ + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, auth_method=None) + assert api.access_token == "tok123" + + +def test_initialize_reapplies_token_hash_after_oauth_init(): + """A configured token_hash must survive _init_oauth(), which resets it to "". + + This is the same trap fox.py/deye.py document: _init_oauth() unconditionally sets + self.token_hash = "" as part of its own bookkeeping, so the configured value has to be + re-applied AFTER that call or the Predbat.com SaaS dedup keyed on the hash breaks. + """ + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, auth_method="oauth", token_hash="abc123hash") + assert api.token_hash == "abc123hash" + + +def test_initialize_normalizes_inverter_sn(): + """inverter_sn is list-normalized the same way fox.py/deye.py do.""" + api_none = SunsynkAPI(MockBase(), key="tok123", automatic=True, inverter_sn=None) + assert api_none.inverter_sn_filter == [] + + api_str = SunsynkAPI(MockBase(), key="tok123", automatic=True, inverter_sn="SN1") + assert api_str.inverter_sn_filter == ["SN1"] + + api_list = SunsynkAPI(MockBase(), key="tok123", automatic=True, inverter_sn=["SN1", "SN2"]) + assert api_list.inverter_sn_filter == ["SN1", "SN2"] + + +def test_is_alive_requires_started_and_token(): + """is_alive() is False until the component has started, even with a token set.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + assert api.is_alive() is False + api.api_started = True + assert api.is_alive() is True + + +def test_request_get_func_success_returns_parsed_json(): + """A 200 response returns the parsed JSON body and clears needs_reauth.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.needs_reauth = True # Simulate a prior auth failure that has since cleared + + mock_response = create_aiohttp_mock_response(status=200, json_data={"code": 0, "data": {"sn": "SN1"}}) + mock_session = create_aiohttp_mock_session(mock_response) + + with patch("aiohttp.ClientSession", return_value=mock_session): + data = run_async(api.request_get("/api/v1/plants")) + + assert data == {"code": 0, "data": {"sn": "SN1"}} + assert api.needs_reauth is False + + +def test_request_get_func_401_sets_needs_reauth_and_does_not_retry(): + """A 401 flags needs_reauth and does NOT attempt a re-login (token is injected).""" + api = SunsynkAPI(MockBase(), key="stale-token", automatic=True) + + mock_response = create_aiohttp_mock_response(status=401, json_data={}) + mock_session = create_aiohttp_mock_session(mock_response) + + with patch("aiohttp.ClientSession", return_value=mock_session): + data, allow_retry = run_async(api.request_get_func("/api/v1/plants")) + + assert data is None + assert allow_retry is False + assert api.needs_reauth is True + assert api.failures_total == 1 + + +def test_request_get_func_403_sets_needs_reauth(): + """A 403 is treated the same as a 401 - injected token rejected, flag needs_reauth.""" + api = SunsynkAPI(MockBase(), key="stale-token", automatic=True) + + mock_response = create_aiohttp_mock_response(status=403, json_data={}) + mock_session = create_aiohttp_mock_session(mock_response) + + with patch("aiohttp.ClientSession", return_value=mock_session): + data, allow_retry = run_async(api.request_get_func("/api/v1/plants")) + + assert data is None + assert allow_retry is False + assert api.needs_reauth is True + + +def test_request_get_func_200_with_success_false_is_treated_as_failure(): + """Sunsynk's API family returns HTTP 200 with {"success": false} on failure (e.g. stale + token, invalid request) rather than a non-2xx status. This must not be unwrapped into an + empty/None 'data' payload as if the call succeeded - mirrors deye.py's + `if not data.get("success", True)` discipline.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + + mock_response = create_aiohttp_mock_response(status=200, json_data={"code": 1, "success": False, "msg": "token invalid", "data": None}) + mock_session = create_aiohttp_mock_session(mock_response) + + with patch("aiohttp.ClientSession", return_value=mock_session): + data, allow_retry = run_async(api.request_get_func("/api/v1/plants")) + + assert data is None + assert allow_retry is False + assert api.failures_total == 1 + + +def test_get_battery_treats_200_success_false_as_failure_and_does_not_cache(): + """End-to-end: a 200 response with {"success": false} must surface as None all the way up + through request_get()/get_battery(), and must NOT populate the battery cache with an + empty dict as if it were valid, fresh data.""" + api = SunsynkAPI(MockBase(), key="stale-token", automatic=True) + + mock_response = create_aiohttp_mock_response(status=200, json_data={"code": 1, "success": False, "msg": "token invalid", "data": None}) + mock_session = create_aiohttp_mock_session(mock_response) + + with patch("aiohttp.ClientSession", return_value=mock_session): + result = run_async(api.get_battery("SN1")) + + assert result is None + assert "SN1" not in api.device_values["battery"] + assert "device_values:battery:SN1" not in api.data_age + + +# ----------------------------------------------------------------------------- +# Task 9: device discovery + telemetry reads +# ----------------------------------------------------------------------------- + + +def test_get_device_list_uses_configured_plant_id_and_returns_devices(): + """When plant_id is configured, get_device_list() skips discovery entirely.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + api.request_get = AsyncMock(return_value=INVERTERS_RESPONSE) + + devices = run_async(api.get_device_list()) + + assert api.request_get.await_count == 1 + assert {d["sn"] for d in devices} == {"2504040106", "2504040164"} + assert all(d["plant_id"] == 505825 for d in devices) + assert all(d["model"] == "SUN-50KW-SG01HP3-EU-BM4" for d in devices) + + +def test_get_device_list_discovers_single_plant_when_plant_id_not_configured(): + """Without a configured plant_id, GET /api/v1/plants is polled first for the sole plant.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(side_effect=[PLANTS_RESPONSE, INVERTERS_RESPONSE]) + + devices = run_async(api.get_device_list()) + + assert api.request_get.await_count == 2 + assert api.request_get.await_args_list[0].args[0] == "/api/v1/plants" + assert api.request_get.await_args_list[1].args[0] == "/api/v1/plant/505825/inverters" + assert {d["sn"] for d in devices} == {"2504040106", "2504040164"} + assert all(d["plant_id"] == 505825 for d in devices) + + +def test_get_device_list_uses_first_plant_when_multiple_found(): + """Multi-plant accounts are defensive-only (SaaS handler rejects them at onboarding) - + the first plant returned is used.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + multi_plant_response = {"code": 0, "success": True, "data": {"infos": [{"id": 111, "name": "Plant A", "status": 1}, {"id": 222, "name": "Plant B", "status": 1}]}} + empty_inverters = {"code": 0, "success": True, "data": {"infos": []}} + api.request_get = AsyncMock(side_effect=[multi_plant_response, empty_inverters]) + + run_async(api.get_device_list()) + + assert api.request_get.await_args_list[1].args[0] == "/api/v1/plant/111/inverters" + + +def test_get_device_list_filters_by_inverter_sn_filter(): + """inverter_sn_filter, when set, restricts the discovered devices.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825, inverter_sn="2504040106") + api.request_get = AsyncMock(return_value=INVERTERS_RESPONSE) + + devices = run_async(api.get_device_list()) + + assert [d["sn"] for d in devices] == ["2504040106"] + + +def test_get_device_list_returns_none_on_api_failure(): + """A failed inverters poll returns None and leaves any cached device_list untouched.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + api.request_get = AsyncMock(return_value=None) + + result = run_async(api.get_device_list()) + + assert result is None + assert api.device_list == [] + + +def test_get_device_list_returns_none_when_no_plants_found(): + """When plant discovery finds no plants, get_device_list() fails closed with None.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + no_plants = {"code": 0, "success": True, "data": {"infos": []}} + api.request_get = AsyncMock(return_value=no_plants) + + result = run_async(api.get_device_list()) + + assert result is None + assert api.request_get.await_count == 1 # never got to the inverters call + + +def test_get_device_list_caches_within_static_refresh_window(): + """A second call within SUNSYNK_REFRESH_STATIC re-uses the cached list, no new request.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + api.request_get = AsyncMock(return_value=INVERTERS_RESPONSE) + + first = run_async(api.get_device_list()) + second = run_async(api.get_device_list()) + + assert api.request_get.await_count == 1 + assert second == first + + +def test_get_plant_flow_parses_flow_shape(): + """get_plant_flow() returns the parsed 'data' dict with the power-flow fields.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value=FLOW_RESPONSE) + + data = run_async(api.get_plant_flow(505825)) + + assert data["battPower"] == 400 + assert data["batTo"] is True + assert data["gridOrMeterPower"] == 3179 + assert data["gridTo"] is True + assert data["soc"] == 100 + + call = api.request_get.await_args + assert call.args[0] == "/api/v1/plant/energy/505825/flow" + assert "date" in call.kwargs["datain"] + + +def test_get_battery_parses_battery_shape(): + """get_battery() returns the parsed 'data' dict and requests the documented path/params.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value=BATTERY_RESPONSE) + + data = run_async(api.get_battery("2504040106")) + + assert data["power"] == 170 + assert data["soc"] == "100.0" + assert data["capacity"] == "100.0" + + call = api.request_get.await_args + assert call.args[0] == "/api/v1/inverter/battery/2504040106/realtime" + assert call.kwargs["datain"] == {"sn": "2504040106", "lan": "en"} + + +def test_get_grid_requests_expected_path(): + """get_grid() hits /inverter/grid/{sn}/realtime with sn+lan query params.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value={"code": 0, "success": True, "data": {"gridPower": 123}}) + + data = run_async(api.get_grid("SN1")) + + assert data["gridPower"] == 123 + call = api.request_get.await_args + assert call.args[0] == "/api/v1/inverter/grid/SN1/realtime" + assert call.kwargs["datain"] == {"sn": "SN1", "lan": "en"} + + +def test_get_input_requests_expected_path(): + """get_input() hits /inverter/{sn}/realtime/input.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value={"code": 0, "success": True, "data": {"pvPower": 5}}) + + data = run_async(api.get_input("SN1")) + + assert data["pvPower"] == 5 + assert api.request_get.await_args.args[0] == "/api/v1/inverter/SN1/realtime/input" + + +def test_get_output_requests_expected_path(): + """get_output() hits /inverter/{sn}/realtime/output.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value={"code": 0, "success": True, "data": {"outputPower": 7}}) + + data = run_async(api.get_output("SN1")) + + assert data["outputPower"] == 7 + assert api.request_get.await_args.args[0] == "/api/v1/inverter/SN1/realtime/output" + + +def test_get_battery_returns_none_on_api_failure(): + """A failed realtime poll returns None rather than an empty/partial dict.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value=None) + + assert run_async(api.get_battery("SN1")) is None + + +def test_realtime_reads_cache_within_refresh_window(): + """A second call for the same sn within SUNSYNK_REFRESH_REALTIME re-uses the cache.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value=BATTERY_RESPONSE) + + run_async(api.get_battery("SN1")) + run_async(api.get_battery("SN1")) + + assert api.request_get.await_count == 1 + + +def test_realtime_reads_are_cached_independently_per_sn(): + """Caching a value for one sn must not short-circuit the fetch for a different sn.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value=BATTERY_RESPONSE) + + run_async(api.get_battery("SN1")) + run_async(api.get_battery("SN2")) + + assert api.request_get.await_count == 2 + + +def test_realtime_cache_freshness_is_per_bucket_and_sn_not_a_shared_clock(): + """Reviewer scenario: all 5 telemetry buckets x all SNs must NOT share one clock. + + Fetch battery(SN1) and grid(SN1) at t0 (2 calls). Advance every data_age entry past + SUNSYNK_REFRESH_REALTIME. Refetch grid(SN1) (3rd call) - this must only refresh grid(SN1)'s + own age, not reset a shared "device_values" timestamp. Then fetch battery(SN1) again: it + must ALSO be recognised as stale and refetch (4th call), not be served stale from cache + just because a different bucket/sn was refreshed in between. + """ + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.request_get = AsyncMock(return_value=BATTERY_RESPONSE) + + run_async(api.get_battery("SN1")) # call 1 + run_async(api.get_grid("SN1")) # call 2 + assert api.request_get.await_count == 2 + + # Clock-injection: age every tracked cache entry past the refresh window, same pattern + # used elsewhere in this suite (test_fox_api.py, test_sigenergy.py, test_enphase_api.py). + stale = datetime.now(timezone.utc) - timedelta(minutes=SUNSYNK_REFRESH_REALTIME + 1) + for key in list(api.data_age.keys()): + api.data_age[key] = stale + + run_async(api.get_grid("SN1")) # call 3 - refetches; must only touch grid(SN1)'s own age + assert api.request_get.await_count == 3 + + run_async(api.get_battery("SN1")) # call 4 - must ALSO refetch, not served stale + assert api.request_get.await_count == 4 + + +# ----------------------------------------------------------------------------- +# Task 10: publish_data() - entity publishing with the VERIFIED sign mapping +# ----------------------------------------------------------------------------- + + +def test_sign_mapping_single_inverter_discharge_and_import(): + """Single-inverter plant, battery DISCHARGING (batTo=true) and grid IMPORTING + (toGrid=false, gridTo=true): battery_power stays positive (discharge +ve), grid_power + flips negative (predbat convention: grid +ve = export, so import must be negative).""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [{"sn": "2504040106", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}] + api.get_plant_flow = AsyncMock(return_value=dict(FLOW_RESPONSE["data"])) + api.get_battery = AsyncMock(return_value=dict(BATTERY_RESPONSE["data"])) + + run_async(api.publish_data()) + + assert api.base.entities["sensor.predbat_sunsynk_2504040106_battery_power"]["state"] == 400 + assert api.base.entities["sensor.predbat_sunsynk_2504040106_grid_power"]["state"] == -3179 + + +def test_sign_mapping_single_inverter_charge_and_export(): + """Single-inverter plant, battery CHARGING (batTo=false, toBat=true) and grid EXPORTING + (toGrid=true): battery_power flips negative (charge -ve), grid_power stays positive + (export +ve).""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [{"sn": "2504040106", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}] + flow = { + "pvPower": 5000, + "battPower": 500, + "batTo": False, + "toBat": True, + "gridOrMeterPower": 1000, + "toGrid": True, + "gridTo": False, + "soc": 80, + "loadOrEpsPower": 2000, + "toLoad": True, + "pvTo": True, + } + api.get_plant_flow = AsyncMock(return_value=flow) + api.get_battery = AsyncMock(return_value=dict(BATTERY_RESPONSE["data"])) + + run_async(api.publish_data()) + + assert api.base.entities["sensor.predbat_sunsynk_2504040106_battery_power"]["state"] == -500 + assert api.base.entities["sensor.predbat_sunsynk_2504040106_grid_power"]["state"] == 1000 + + +def test_publish_data_single_inverter_also_publishes_soc_pv_load_and_battery_extras(): + """Single-inverter plant also publishes battery_soc/pv_power/load_power straight from + flow, plus battery_temperature/soc_max sourced from the per-SN battery endpoint (flow + carries neither field).""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [{"sn": "2504040106", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}] + api.get_plant_flow = AsyncMock(return_value=dict(FLOW_RESPONSE["data"])) + api.get_battery = AsyncMock(return_value=dict(BATTERY_RESPONSE["data"])) + + run_async(api.publish_data()) + + entities = api.base.entities + assert entities["sensor.predbat_sunsynk_2504040106_battery_soc"]["state"] == 100 + assert entities["sensor.predbat_sunsynk_2504040106_pv_power"]["state"] == 0 + assert entities["sensor.predbat_sunsynk_2504040106_load_power"]["state"] == 3164 + assert entities["sensor.predbat_sunsynk_2504040106_battery_temperature"]["state"] == 17.0 + assert entities["sensor.predbat_sunsynk_2504040106_soc_max"]["state"] == 100.0 + + +def test_publish_data_multi_inverter_avoids_double_count_and_warns_on_per_sn_signs(): + """Multi-inverter plant: per-SN entities must come from the per-SN battery/grid/input + endpoints, NOT from plant flow (flow is plant-aggregated and would double-count a + multi-inverter site - Codex Critical 3). Plant flow is fetched exactly once, used only + for the site-level summary. A Warn: log must flag that per-SN signs are unconfirmed.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [ + {"sn": "SNA", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + {"sn": "SNB", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + ] + api.get_plant_flow = AsyncMock(return_value=dict(FLOW_RESPONSE["data"])) + + battery_by_sn = { + "SNA": {"power": 170, "capacity": "100.0", "soc": "55.0", "temp": "17.0", "voltage": "639.8"}, + "SNB": {"power": -220, "capacity": "150.0", "soc": "60.0", "temp": "19.5", "voltage": "640.1"}, + } + grid_by_sn = { + "SNA": {"vip": [{"power": 100}, {"power": 200}, {"power": 300}]}, + "SNB": {"power": 50}, + } + input_by_sn = { + "SNA": {"pv": [{"power": 1000}, {"power": 500}]}, + "SNB": {"pvPower": 750}, + } + api.get_battery = AsyncMock(side_effect=lambda sn: battery_by_sn[sn]) + api.get_grid = AsyncMock(side_effect=lambda sn: grid_by_sn[sn]) + api.get_input = AsyncMock(side_effect=lambda sn: input_by_sn[sn]) + # Not under test here - the multi-inverter path also fetches per-SN output for load_power + # (Fix 1); mocked purely to avoid a real network call from the unmocked get_output(). + api.get_output = AsyncMock(return_value={"pInv": 0}) + + logs = [] + api.log = logs.append + + run_async(api.publish_data()) + + entities = api.base.entities + + # Per-SN entities come from the per-SN endpoints, not plant flow (no double-count). + assert entities["sensor.predbat_sunsynk_sna_battery_power"]["state"] == 170 + assert entities["sensor.predbat_sunsynk_snb_battery_power"]["state"] == -220 + assert entities["sensor.predbat_sunsynk_sna_grid_power"]["state"] == 600 # summed vip phases + assert entities["sensor.predbat_sunsynk_snb_grid_power"]["state"] == 50 # flat fallback + assert entities["sensor.predbat_sunsynk_sna_pv_power"]["state"] == 1500 # summed pv strings + assert entities["sensor.predbat_sunsynk_snb_pv_power"]["state"] == 750 # flat fallback + assert entities["sensor.predbat_sunsynk_sna_soc_max"]["state"] == 100.0 + assert entities["sensor.predbat_sunsynk_snb_soc_max"]["state"] == 150.0 + + # Plant flow is used exactly once, for the site-level summary only - never per-SN. + assert api.get_plant_flow.await_count == 1 + assert entities["sensor.predbat_sunsynk_site_battery_power"]["state"] == 400 # batTo=true + assert entities["sensor.predbat_sunsynk_site_grid_power"]["state"] == -3179 # toGrid=false + + # Per-SN sign-uncertainty must be surfaced, not silently assumed correct. + assert any("UNCONFIRMED" in message for message in logs) + + +def test_publish_data_multi_inverter_publishes_load_power_per_sn(): + """Multi-inverter plant: per-SN load_power must ALSO be published (Fix 1) - solis.py + wires load_power as a per-device entity (solis.py:1273), so without a per-SN + sensor.{prefix}_sunsynk_{sn}_load_power, a future automatic_config wiring step would have + nothing to map for multi-inverter customers. Sourced from the per-SN output endpoint + (inverter AC output), NOT plant flow's loadOrEpsPower (which is plant-aggregate and would + double-count here, same as battery/grid/pv).""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [ + {"sn": "SNA", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + {"sn": "SNB", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + ] + api.get_plant_flow = AsyncMock(return_value=dict(FLOW_RESPONSE["data"])) + api.get_battery = AsyncMock(return_value=dict(BATTERY_RESPONSE["data"])) + api.get_grid = AsyncMock(return_value={"power": 100}) + api.get_input = AsyncMock(return_value={"pvPower": 200}) + + output_by_sn = { + "SNA": {"vip": [{"power": 300}, {"power": 400}]}, # summed per-phase breakdown + "SNB": {"pac": 650}, # flat fallback + } + api.get_output = AsyncMock(side_effect=lambda sn: output_by_sn[sn]) + + logs = [] + api.log = logs.append + + run_async(api.publish_data()) + + entities = api.base.entities + assert entities["sensor.predbat_sunsynk_sna_load_power"]["state"] == 700 + assert entities["sensor.predbat_sunsynk_snb_load_power"]["state"] == 650 + + # Per-SN load source uncertainty must be surfaced too, same discipline as the sign caveat. + assert any("load_power source" in message and "UNCONFIRMED" in message for message in logs) + + +def test_multi_inverter_sn_does_not_refetch_battery_that_already_failed(): + """Fix 3: _publish_battery_extras() must NOT re-fetch a battery that the multi-inverter + path already fetched and failed (None) this cycle - a plain `None` default couldn't tell + 'not fetched yet' apart from 'already fetched and failed', so a failing battery endpoint + was being hit twice per publish cycle.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.get_battery = AsyncMock(return_value=None) + api.get_grid = AsyncMock(return_value=None) + api.get_input = AsyncMock(return_value=None) + api.get_output = AsyncMock(return_value=None) + + run_async(api._publish_multi_inverter_sn("sensor.predbat_sunsynk", "SNA")) + + assert api.get_battery.await_count == 1 + + +def test_publish_data_single_inverter_skips_device_with_missing_sn_and_warns(): + """Final-review Fix 2: the single-inverter branch must guard a missing SN the same way the + multi-inverter branch already does. Before this fix, `devices[0].get("sn")` was passed + straight into `_publish_single_inverter` -> `_publish_entity`, which does `sn.lower()` with + no None check - an AttributeError if a discovered single device lacks "sn".""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [{"sn": None, "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}] + api.get_plant_flow = AsyncMock(return_value=dict(FLOW_RESPONSE["data"])) + logs = [] + api.log = logs.append + + run_async(api.publish_data()) # must not raise AttributeError on sn.lower() + + assert any("skipping device with missing SN" in message for message in logs) + assert api.base.entities == {} + + +def test_publish_data_multi_inverter_skips_device_with_missing_sn_and_warns(): + """Fix 4: a device with no SN must be skipped (not crash on sn.lower()) AND must log a + Warn:, rather than silently disappearing from published entities.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [ + {"sn": None, "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + {"sn": "SNB", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + ] + api.get_plant_flow = AsyncMock(return_value=None) + api.get_battery = AsyncMock(return_value=None) + api.get_grid = AsyncMock(return_value=None) + api.get_input = AsyncMock(return_value=None) + api.get_output = AsyncMock(return_value=None) + logs = [] + api.log = logs.append + + run_async(api.publish_data()) + + assert any("skipping device with missing SN" in message for message in logs) + assert api.get_battery.await_count == 1 # only attempted for SNB, never for the missing-sn device + + +def test_publish_data_multi_inverter_warns_when_flow_fetch_fails(): + """Fix 5: parity with the single-inverter branch - when plant flow fails, the multi- + inverter branch must also log a Warn: (not silently skip the site summary).""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [ + {"sn": "SNA", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + {"sn": "SNB", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + ] + api.get_plant_flow = AsyncMock(return_value=None) + api.get_battery = AsyncMock(return_value=None) + api.get_grid = AsyncMock(return_value=None) + api.get_input = AsyncMock(return_value=None) + api.get_output = AsyncMock(return_value=None) + logs = [] + api.log = logs.append + + run_async(api.publish_data()) + + assert any("no plant flow data available, skipping site summary" in message for message in logs) + assert "sensor.predbat_sunsynk_site_battery_power" not in api.base.entities + + +def test_publish_data_with_no_devices_logs_warning_and_publishes_nothing(): + """publish_data() with an empty device_list must not raise, and must publish nothing.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.device_list = [] + logs = [] + api.log = logs.append + + run_async(api.publish_data()) + + assert api.base.entities == {} + assert any("no discovered devices" in message for message in logs) + + +# ----------------------------------------------------------------------------- +# Task 11: automatic_config() - set_arg map wiring Predbat to the Task 10 entities +# ----------------------------------------------------------------------------- + + +def test_automatic_config_maps_two_inverters(): + """automatic_config() discovers devices via get_device_list() and emits a set_arg map + whose per-device lists are ordered per the discovered device order, pointing at the + EXACT entity ids publish_data() (Task 10) publishes - sensor.{prefix}_sunsynk_{sn}_{leaf} + with SN lower-cased.""" + api = SunsynkAPI(MockBase(), key="t", automatic=True, plant_id=505825) + api.request_get = AsyncMock(return_value=INVERTERS_RESPONSE) + + run_async(api.automatic_config()) + + assert api.base.args["inverter_type"] == ["SUNSYNK", "SUNSYNK"] + assert api.base.args["num_inverters"] == 2 + assert api.base.args["soc_percent"] == [ + "sensor.predbat_sunsynk_2504040106_battery_soc", + "sensor.predbat_sunsynk_2504040164_battery_soc", + ] + assert api.base.args["battery_power"] == [ + "sensor.predbat_sunsynk_2504040106_battery_power", + "sensor.predbat_sunsynk_2504040164_battery_power", + ] + assert api.base.args["grid_power"] == [ + "sensor.predbat_sunsynk_2504040106_grid_power", + "sensor.predbat_sunsynk_2504040164_grid_power", + ] + assert api.base.args["load_power"] == [ + "sensor.predbat_sunsynk_2504040106_load_power", + "sensor.predbat_sunsynk_2504040164_load_power", + ] + assert api.base.args["pv_power"] == [ + "sensor.predbat_sunsynk_2504040106_pv_power", + "sensor.predbat_sunsynk_2504040164_pv_power", + ] + assert api.base.args["battery_temperature"] == [ + "sensor.predbat_sunsynk_2504040106_battery_temperature", + "sensor.predbat_sunsynk_2504040164_battery_temperature", + ] + assert api.base.args["soc_max"] == [ + "sensor.predbat_sunsynk_2504040106_soc_max", + "sensor.predbat_sunsynk_2504040164_soc_max", + ] + # A charge control entity - not published yet (Task 15), but must be named consistently. + assert api.base.args["charge_start_time"] == [ + "select.predbat_sunsynk_2504040106_charge_slot1_start_time", + "select.predbat_sunsynk_2504040164_charge_slot1_start_time", + ] + assert api.base.args["scheduled_discharge_enable"] == [ + "switch.predbat_sunsynk_2504040106_discharge_slot1_enable", + "switch.predbat_sunsynk_2504040164_discharge_slot1_enable", + ] + + +def test_automatic_config_respects_automatic_ignore_pv(): + """When automatic_ignore_pv is set, pv_power must NOT be wired (mirrors fox.py/deye.py), + leaving any manually configured PV sensor in apps.yaml untouched.""" + api = SunsynkAPI(MockBase(), key="t", automatic=True, plant_id=505825, automatic_ignore_pv=True) + api.request_get = AsyncMock(return_value=INVERTERS_RESPONSE) + + run_async(api.automatic_config()) + + assert "pv_power" not in api.base.args + + +def test_automatic_config_logs_one_structured_json_line_with_the_full_map(): + """The complete arg map must be logged as ONE structured, Loki-greppable line (spec + section 3.1, non-optional) - not scattered across many individual set_arg log lines.""" + api = SunsynkAPI(MockBase(), key="t", automatic=True, plant_id=505825) + api.request_get = AsyncMock(return_value=INVERTERS_RESPONSE) + logs = [] + api.log = logs.append + + run_async(api.automatic_config()) + + matches = [message for message in logs if message.startswith("Info: SunsynkAPI automatic_config: ")] + assert len(matches) == 1 + + logged_map = json.loads(matches[0][len("Info: SunsynkAPI automatic_config: ") :]) + assert logged_map["inverter_type"] == ["SUNSYNK", "SUNSYNK"] + assert logged_map["num_inverters"] == 2 + assert logged_map["soc_percent"] == api.base.args["soc_percent"] + assert logged_map["charge_start_time"] == api.base.args["charge_start_time"] + + +def test_automatic_config_no_inverters_warns_and_sets_nothing(): + """No discovered devices: log a Warn: and set no args at all, rather than emitting an + empty-list config that would silently disable Predbat's inverter control.""" + api = SunsynkAPI(MockBase(), key="t", automatic=True, plant_id=505825) + api.request_get = AsyncMock(return_value={"code": 0, "success": True, "data": {"infos": []}}) + logs = [] + api.log = logs.append + + run_async(api.automatic_config()) + + assert any("SunsynkAPI automatic_config found no inverters" in message for message in logs) + assert api.base.args == {} + + +def test_automatic_config_skips_device_with_missing_sn_and_warns(): + """A device with no SN must be skipped (not crash on sn.lower()) and must log a Warn:, + mirroring publish_data()'s Fix 4 discipline (Task 10).""" + api = SunsynkAPI(MockBase(), key="t", automatic=True) + api.get_device_list = AsyncMock( + return_value=[ + {"sn": None, "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + {"sn": "2504040164", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}, + ] + ) + logs = [] + api.log = logs.append + + run_async(api.automatic_config()) + + assert any("skipping device with missing SN" in message for message in logs) + assert api.base.args["num_inverters"] == 1 + assert api.base.args["soc_percent"] == ["sensor.predbat_sunsynk_2504040164_battery_soc"] + + +# ----------------------------------------------------------------------------- +# Task 12: run() loop + COMPONENT_LIST/config.py registration +# ----------------------------------------------------------------------------- + + +def test_sunsynk_registered_in_inverter_def(): + """SUNSYNK must be registered in config.py's INVERTER_DEF (final-review Fix 1). + + automatic_config() sets inverter_type="SUNSYNK" (Task 11) but never sets a custom + "inverter" arg, so inverter.py's GE-copy fallback (inverter.py:193-195, only triggered when + "inverter" is in self.base.args) never runs for it - without a registered entry, + Inverter.__init__() falls straight to `raise ValueError("Inverter type SUNSYNK not + defined")` (inverter.py:210), crashing Predbat's inverter setup for every + sunsynk_automatic customer. Mirrors deye.py's "DeyeCloud" registration, added to guard the + exact same gap. + + Imports ``predbat`` first - see test_component_registered()'s docstring below for why + (components.py -> web.py -> predbat.py -> components.py circular import).""" + import predbat # noqa: F401 - import order matters, see test_component_registered() docstring + + from config import INVERTER_DEF + + assert "SUNSYNK" in INVERTER_DEF + idef = INVERTER_DEF["SUNSYNK"] + # These are exactly the keys inverter.py's __init__ reads immediately after inverter_type + # is resolved (inverter.py:208, :217-218) - the minimum needed for setup to not crash. + assert idef["name"] == "SUNSYNK" + assert idef["has_rest_api"] is False + assert idef["has_mqtt_api"] is False + + +def test_component_registered(): + """sunsynk is registered in COMPONENT_LIST, wired to SunsynkAPI with the expected + event_filter, matching how fox/deye/solis register themselves. + + Imports ``predbat`` before ``components`` - components.py -> web.py -> predbat.py -> + components.py is a genuine circular import, and entering it via a bare + `from components import ...` (before anything has imported predbat) fails because + Components/COMPONENT_LIST aren't defined yet in the partially-initialised components + module when predbat.py re-enters it. unit_test.py's real suite avoids this by importing + predbat first (`from predbat import PredBat`) before any test module runs; mirroring + that order here keeps this test robust when run standalone via pytest too.""" + import predbat # noqa: F401 - import order matters, see docstring above + + from components import COMPONENT_LIST + + assert "sunsynk" in COMPONENT_LIST + entry = COMPONENT_LIST["sunsynk"] + assert entry["class"] is SunsynkAPI + assert entry["event_filter"] == "predbat_sunsynk_" + assert entry["phase"] == 1 + assert entry["args"]["key"]["config"] == "sunsynk_key" + assert entry["args"]["plant_id"]["config"] == "sunsynk_plant_id" + assert entry["args"]["automatic"]["config"] == "sunsynk_automatic" + assert entry["args"]["automatic"]["default"] is False + + +def test_component_registered_key_required_so_it_does_not_start_fleet_wide(): + """The key arg must be required: True (mirroring teslemetry's single-token gate), not + required: False with a required_or fallback (deye/solis's multi-auth-path gate). Sunsynk + has only one auth path (an injected sunsynk_key), so without this every individual arg is + optional and Components.initialize()'s have_all_args check would be unconditionally True, + starting SunsynkAPI for every instance regardless of whether it's a Sunsynk customer.""" + import predbat # noqa: F401 - import order matters, see docstring on test_component_registered + + from components import COMPONENT_LIST + + assert COMPONENT_LIST["sunsynk"]["args"]["key"]["required"] is True + + +def test_component_registry_config_schema_has_sunsynk_keys(): + """APPS_SCHEMA declares the sunsynk_* keys, mirroring fox_*/solis_*/deye_*.""" + from config import APPS_SCHEMA + + assert APPS_SCHEMA["sunsynk_key"] == {"type": "string", "empty": False} + assert APPS_SCHEMA["sunsynk_automatic"] == {"type": "boolean"} + assert APPS_SCHEMA["sunsynk_plant_id"] == {"type": "string", "empty": False} + + +def test_run_first_publishes_and_runs_automatic_config_when_automatic(): + """First run: get_device_list() populates devices, publish_data() runs, and + automatic_config() fires because first=True and automatic=True.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + api.get_device_list = AsyncMock(return_value=[{"sn": "SN1", "model": "M", "plant_id": 505825}]) + api.device_list = [{"sn": "SN1", "model": "M", "plant_id": 505825}] + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock() + + result = run_async(api.run(5, True)) + + assert result is True + api.publish_data.assert_awaited_once() + api.automatic_config.assert_awaited_once() + assert api._known_sns == {"SN1"} + + +def test_run_not_automatic_skips_automatic_config(): + """When self.automatic is False, automatic_config() must never be called, even on first.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=False, plant_id=505825) + api.get_device_list = AsyncMock(return_value=[{"sn": "SN1", "model": "M", "plant_id": 505825}]) + api.device_list = [{"sn": "SN1", "model": "M", "plant_id": 505825}] + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock() + + result = run_async(api.run(5, True)) + + assert result is True + api.publish_data.assert_awaited_once() + api.automatic_config.assert_not_awaited() + + +def test_run_no_devices_returns_false_and_skips_publish(): + """No discovered devices (empty self.device_list): run() logs an Error: and returns False + without calling publish_data() or automatic_config().""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + api.get_device_list = AsyncMock(return_value=None) + api.device_list = [] + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock() + logs = [] + api.log = logs.append + + result = run_async(api.run(5, True)) + + assert result is False + api.publish_data.assert_not_awaited() + api.automatic_config.assert_not_awaited() + assert any("No devices found" in message for message in logs) + + +def test_run_second_cycle_skips_automatic_config_when_device_set_unchanged(): + """On a later cycle (first=False) with the same discovered SNs as last time, + automatic_config() must NOT be re-invoked.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + devices = [{"sn": "SN1", "model": "M", "plant_id": 505825}] + api.get_device_list = AsyncMock(return_value=devices) + api.device_list = devices + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock() + + run_async(api.run(5, True)) # first cycle - seeds _known_sns, fires automatic_config + api.automatic_config.reset_mock() + + result = run_async(api.run(5, False)) # second cycle - same device set + + assert result is True + api.automatic_config.assert_not_awaited() + + +def test_run_reinvokes_automatic_config_when_device_set_changes(): + """When the discovered device set changes between cycles (e.g. an inverter is added), + automatic_config() fires again even though first=False.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + devices_before = [{"sn": "SN1", "model": "M", "plant_id": 505825}] + devices_after = [{"sn": "SN1", "model": "M", "plant_id": 505825}, {"sn": "SN2", "model": "M", "plant_id": 505825}] + api.get_device_list = AsyncMock(return_value=devices_before) + api.device_list = devices_before + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock() + + run_async(api.run(5, True)) + api.automatic_config.reset_mock() + + api.get_device_list = AsyncMock(return_value=devices_after) + api.device_list = devices_after + result = run_async(api.run(5, False)) + + assert result is True + api.automatic_config.assert_awaited_once() + assert api._known_sns == {"SN1", "SN2"} + + +def test_run_continues_when_publish_data_raises(): + """Final-review Fix 3: an exception from publish_data() must not abort run() - matches + deye.py's per-poll error-wrapping discipline so one bad cycle doesn't kill automatic_config + (or the component as a whole). Before this fix, publish_data() was awaited with no + try/except in run(), so any exception it raised propagated straight out of run().""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + api.get_device_list = AsyncMock(return_value=[{"sn": "SN1", "model": "M", "plant_id": 505825}]) + api.device_list = [{"sn": "SN1", "model": "M", "plant_id": 505825}] + api.publish_data = AsyncMock(side_effect=RuntimeError("boom")) + api.automatic_config = AsyncMock() + logs = [] + api.log = logs.append + + result = run_async(api.run(5, True)) # must not raise + + assert result is True + api.automatic_config.assert_awaited_once() # still runs despite the publish_data failure + assert any("publish_data failed" in message for message in logs) + + +def test_run_continues_when_automatic_config_raises(): + """Final-review Fix 3: mirrors the publish_data() case - an exception from + automatic_config() must also not abort run() or its return value.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + api.get_device_list = AsyncMock(return_value=[{"sn": "SN1", "model": "M", "plant_id": 505825}]) + api.device_list = [{"sn": "SN1", "model": "M", "plant_id": 505825}] + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock(side_effect=RuntimeError("boom")) + logs = [] + api.log = logs.append + + result = run_async(api.run(5, True)) # must not raise + + assert result is True + api.publish_data.assert_awaited_once() + assert any("automatic_config failed" in message for message in logs) + + +def test_run_uses_cached_device_list_on_transient_api_failure(): + """Mirrors fox.py's run(): if get_device_list() returns None (transient API failure) but a + previously-cached self.device_list is still populated, run() must still publish from the + cached list rather than reporting "no devices".""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + api.device_list = [{"sn": "SN1", "model": "M", "plant_id": 505825}] # pre-populated cache + api.get_device_list = AsyncMock(return_value=None) # this cycle's poll failed + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock() + + result = run_async(api.run(5, True)) + + assert result is True + api.publish_data.assert_awaited_once() + + +def test_sunsynk_load_cached_data_no_storage_leaves_defaults(): + """load_cached_data() is a no-op (keeps empty defaults) when there is no storage + component wired up - matches _save_cache()/_load_cache()'s own no-storage no-op + discipline (both already return early without a storage component).""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True) + assert api.storage is None + + run_async(api.load_cached_data()) + + assert api.device_list == [] + assert api.device_values == {"flow": {}, "battery": {}, "grid": {}, "input": {}, "output": {}} + + +def test_sunsynk_save_then_load_cache_restores_state(): + """Fix 2: load_cached_data() restores device_list/device_values (and their data_age) from + a previously-saved cache on a FRESH instance, proving a Predbat restart no longer throws + away the persisted cache and cold-re-polls the Sunsynk Connect API immediately - three + reviewers flagged that _load_cache() existed but was never called anywhere.""" + storage = FakeStorage() + + api1, _ = _sunsynk_api_with_storage(storage) + api1.device_list = [{"sn": "2504040106", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}] + api1.device_values = {"flow": {}, "battery": {"2504040106": {"soc": "80.0"}}, "grid": {}, "input": {}, "output": {}} + run_async(api1._save_cache("device_list", api1.device_list)) + run_async(api1._save_cache("device_values", api1.device_values)) + + # A brand new instance, as a restart would create - starts with the usual empty defaults. + api2, _ = _sunsynk_api_with_storage(storage) + assert api2.device_list == [] + assert api2.data_age == {} + + run_async(api2.load_cached_data()) + + assert api2.device_list == [{"sn": "2504040106", "model": "SUN-50KW-SG01HP3-EU-BM4", "plant_id": 505825}] + assert api2.device_values["battery"]["2504040106"]["soc"] == "80.0" + assert "device_list" in api2.data_age + assert "device_values" in api2.data_age + + +def test_run_first_calls_load_cached_data_before_get_device_list(): + """Fix 2: run() must restore cached device data BEFORE the first get_device_list() poll on + a cold start (first=True), mirroring fox.py's run(). Without this ordering the cache restore + is either never wired in at all, or wired in too late to avoid an immediate re-poll.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + calls = [] + + async def fake_load_cached_data(): + """Record that load_cached_data() ran, in call order.""" + calls.append("load_cached_data") + + async def fake_get_device_list(): + """Record that get_device_list() ran, in call order, and return one discovered device.""" + calls.append("get_device_list") + return [{"sn": "SN1", "model": "M", "plant_id": 505825}] + + api.load_cached_data = fake_load_cached_data + api.get_device_list = fake_get_device_list + api.device_list = [{"sn": "SN1", "model": "M", "plant_id": 505825}] + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock() + + result = run_async(api.run(5, True)) + + assert result is True + assert calls == ["load_cached_data", "get_device_list"] + + +def test_run_second_cycle_does_not_call_load_cached_data(): + """Fix 2: run() must only restore cached data on the FIRST cycle (first=True), not on + every subsequent cycle - a restart-only concern, not a per-cycle one.""" + api = SunsynkAPI(MockBase(), key="tok123", automatic=True, plant_id=505825) + devices = [{"sn": "SN1", "model": "M", "plant_id": 505825}] + api.get_device_list = AsyncMock(return_value=devices) + api.device_list = devices + api.publish_data = AsyncMock() + api.automatic_config = AsyncMock() + api.load_cached_data = AsyncMock() + + run_async(api.run(5, True)) + api.load_cached_data.assert_awaited_once() + api.load_cached_data.reset_mock() + + result = run_async(api.run(5, False)) + + assert result is True + api.load_cached_data.assert_not_awaited() + + +# --- Task 13: standalone CLI harness RSA login helpers --------------------------------------- +# +# These test the payload/sign CONSTRUCTION only (no live network call - the harness itself is +# exercised manually against a real Sunsynk account, per the task brief). They also prove the +# `cryptography` import is correctly guarded, so importing sunsynk.py as a SaaS component never +# requires it. + + +def test_sunsynk_rsa_encrypt_password_round_trips_with_matching_private_key(): + """_sunsynk_rsa_encrypt_password() must produce a base64 PKCS1v15 ciphertext that decrypts + back to the original plaintext with the matching private key - proves the standalone login + helper's RSA step is wired correctly (right padding scheme, right PEM wrapping of Sunsynk's + raw publicKey body) without ever calling the live Sunsynk API.""" + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric import padding, rsa + + private_key = rsa.generate_private_key(public_exponent=65537, key_size=1024) + public_pem = private_key.public_key().public_bytes(encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo).decode() + # Sunsynk's publicKey response body is the raw base64 key WITHOUT PEM header/footer - + # strip them here, mirroring what _sunsynk_rsa_encrypt_password() itself re-wraps. + raw_key = "".join(public_pem.strip().splitlines()[1:-1]) + + encrypted_b64 = sunsynk._sunsynk_rsa_encrypt_password(raw_key, "hunter2") + decrypted = private_key.decrypt(base64.b64decode(encrypted_b64), padding.PKCS1v15()) + + assert decrypted == b"hunter2" + + +def test_sunsynk_build_login_payload_shape_and_sign(): + """_sunsynk_build_login_payload() must build the exact /oauth/token/new payload shape and + MD5 sign the verified auth spike (scratchpad/sunsynk_auth_spike.py) uses, for a fixed + nonce/raw_key - this is the payload the standalone CLI harness's login step sends.""" + with patch("sunsynk._sunsynk_rsa_encrypt_password", return_value="ENCRYPTED") as mock_rsa: + payload = sunsynk._sunsynk_build_login_payload("RAWKEY1234567890", "user@example.com", "pw", nonce=1700000000000) + + mock_rsa.assert_called_once_with("RAWKEY1234567890", "pw") + assert payload == { + "username": "user@example.com", + "password": "ENCRYPTED", + "grant_type": "password", + "client_id": "csp-web", + "source": "sunsynk", + "nonce": 1700000000000, + "sign": hashlib.md5("nonce=1700000000000&source=sunsynkRAWKEY1234".encode()).hexdigest(), + } + + +def test_sunsynk_rsa_encrypt_password_missing_cryptography_raises_clear_error(): + """When `cryptography` isn't installed, the standalone login helper must raise a clear, + actionable RuntimeError (not a bare ImportError/traceback) - the SaaS/component import + path is unaffected either way, since it never calls this function.""" + import builtins + + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): + """Raise ImportError for `cryptography` (and its submodules), else defer to the real import.""" + if name == "cryptography" or name.startswith("cryptography."): + raise ImportError("No module named 'cryptography'") + return real_import(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=fake_import): + try: + sunsynk._sunsynk_rsa_encrypt_password("RAWKEY", "pw") + except RuntimeError as error: + assert "pip install cryptography" in str(error) + else: + raise AssertionError("expected RuntimeError for missing cryptography") + + +def test_sunsynk_module_has_no_top_level_cryptography_import(): + """Static guard: `cryptography` must never be imported at module scope in sunsynk.py - + only inside _sunsynk_rsa_encrypt_password()'s guarded try/except (Task 13 brief) - so + importing sunsynk.py as a SaaS component never requires the dependency. Checked via AST + rather than an in-process reload (which would mutate the shared sunsynk module for every + other test in this pytest session) - a real "does the guard work end-to-end" run is done + separately, outside pytest, against a clean subprocess.""" + import ast + import inspect + + tree = ast.parse(inspect.getsource(sunsynk)) + for node in tree.body: # tree.body is module-level statements only, not nested in defs + names = [] + if isinstance(node, ast.Import): + names = [alias.name for alias in node.names] + elif isinstance(node, ast.ImportFrom): + names = [node.module or ""] + assert not any(name == "cryptography" or name.startswith("cryptography.") for name in names), "cryptography must not be imported at module level" + + +def run_sunsynk_tests(my_predbat): + """Run all Sunsynk Connect API component tests (Tasks 8-13, plus cache-restore Fix 2).""" + failed = False + for name, fn in [ + ("get_headers_uses_bearer_token", test_get_headers_uses_bearer_token), + ("initialize_stores_key_as_access_token_regardless_of_auth_method", test_initialize_stores_key_as_access_token_regardless_of_auth_method), + ("initialize_reapplies_token_hash_after_oauth_init", test_initialize_reapplies_token_hash_after_oauth_init), + ("initialize_normalizes_inverter_sn", test_initialize_normalizes_inverter_sn), + ("is_alive_requires_started_and_token", test_is_alive_requires_started_and_token), + ("request_get_func_success_returns_parsed_json", test_request_get_func_success_returns_parsed_json), + ("request_get_func_401_sets_needs_reauth_and_does_not_retry", test_request_get_func_401_sets_needs_reauth_and_does_not_retry), + ("request_get_func_403_sets_needs_reauth", test_request_get_func_403_sets_needs_reauth), + ("request_get_func_200_with_success_false_is_treated_as_failure", test_request_get_func_200_with_success_false_is_treated_as_failure), + ("get_battery_treats_200_success_false_as_failure_and_does_not_cache", test_get_battery_treats_200_success_false_as_failure_and_does_not_cache), + ("get_device_list_uses_configured_plant_id_and_returns_devices", test_get_device_list_uses_configured_plant_id_and_returns_devices), + ("get_device_list_discovers_single_plant_when_plant_id_not_configured", test_get_device_list_discovers_single_plant_when_plant_id_not_configured), + ("get_device_list_uses_first_plant_when_multiple_found", test_get_device_list_uses_first_plant_when_multiple_found), + ("get_device_list_filters_by_inverter_sn_filter", test_get_device_list_filters_by_inverter_sn_filter), + ("get_device_list_returns_none_on_api_failure", test_get_device_list_returns_none_on_api_failure), + ("get_device_list_returns_none_when_no_plants_found", test_get_device_list_returns_none_when_no_plants_found), + ("get_device_list_caches_within_static_refresh_window", test_get_device_list_caches_within_static_refresh_window), + ("get_plant_flow_parses_flow_shape", test_get_plant_flow_parses_flow_shape), + ("get_battery_parses_battery_shape", test_get_battery_parses_battery_shape), + ("get_grid_requests_expected_path", test_get_grid_requests_expected_path), + ("get_input_requests_expected_path", test_get_input_requests_expected_path), + ("get_output_requests_expected_path", test_get_output_requests_expected_path), + ("get_battery_returns_none_on_api_failure", test_get_battery_returns_none_on_api_failure), + ("realtime_reads_cache_within_refresh_window", test_realtime_reads_cache_within_refresh_window), + ("realtime_reads_are_cached_independently_per_sn", test_realtime_reads_are_cached_independently_per_sn), + ("realtime_cache_freshness_is_per_bucket_and_sn_not_a_shared_clock", test_realtime_cache_freshness_is_per_bucket_and_sn_not_a_shared_clock), + ("sign_mapping_single_inverter_discharge_and_import", test_sign_mapping_single_inverter_discharge_and_import), + ("sign_mapping_single_inverter_charge_and_export", test_sign_mapping_single_inverter_charge_and_export), + ("publish_data_single_inverter_also_publishes_soc_pv_load_and_battery_extras", test_publish_data_single_inverter_also_publishes_soc_pv_load_and_battery_extras), + ("publish_data_multi_inverter_avoids_double_count_and_warns_on_per_sn_signs", test_publish_data_multi_inverter_avoids_double_count_and_warns_on_per_sn_signs), + ("publish_data_multi_inverter_publishes_load_power_per_sn", test_publish_data_multi_inverter_publishes_load_power_per_sn), + ("multi_inverter_sn_does_not_refetch_battery_that_already_failed", test_multi_inverter_sn_does_not_refetch_battery_that_already_failed), + ("publish_data_single_inverter_skips_device_with_missing_sn_and_warns", test_publish_data_single_inverter_skips_device_with_missing_sn_and_warns), + ("publish_data_multi_inverter_skips_device_with_missing_sn_and_warns", test_publish_data_multi_inverter_skips_device_with_missing_sn_and_warns), + ("publish_data_multi_inverter_warns_when_flow_fetch_fails", test_publish_data_multi_inverter_warns_when_flow_fetch_fails), + ("publish_data_with_no_devices_logs_warning_and_publishes_nothing", test_publish_data_with_no_devices_logs_warning_and_publishes_nothing), + ("automatic_config_maps_two_inverters", test_automatic_config_maps_two_inverters), + ("automatic_config_respects_automatic_ignore_pv", test_automatic_config_respects_automatic_ignore_pv), + ("automatic_config_logs_one_structured_json_line_with_the_full_map", test_automatic_config_logs_one_structured_json_line_with_the_full_map), + ("automatic_config_no_inverters_warns_and_sets_nothing", test_automatic_config_no_inverters_warns_and_sets_nothing), + ("automatic_config_skips_device_with_missing_sn_and_warns", test_automatic_config_skips_device_with_missing_sn_and_warns), + ("sunsynk_registered_in_inverter_def", test_sunsynk_registered_in_inverter_def), + ("component_registered", test_component_registered), + ("component_registered_key_required_so_it_does_not_start_fleet_wide", test_component_registered_key_required_so_it_does_not_start_fleet_wide), + ("component_registry_config_schema_has_sunsynk_keys", test_component_registry_config_schema_has_sunsynk_keys), + ("run_first_publishes_and_runs_automatic_config_when_automatic", test_run_first_publishes_and_runs_automatic_config_when_automatic), + ("run_not_automatic_skips_automatic_config", test_run_not_automatic_skips_automatic_config), + ("run_no_devices_returns_false_and_skips_publish", test_run_no_devices_returns_false_and_skips_publish), + ("run_second_cycle_skips_automatic_config_when_device_set_unchanged", test_run_second_cycle_skips_automatic_config_when_device_set_unchanged), + ("run_reinvokes_automatic_config_when_device_set_changes", test_run_reinvokes_automatic_config_when_device_set_changes), + ("run_continues_when_publish_data_raises", test_run_continues_when_publish_data_raises), + ("run_continues_when_automatic_config_raises", test_run_continues_when_automatic_config_raises), + ("run_uses_cached_device_list_on_transient_api_failure", test_run_uses_cached_device_list_on_transient_api_failure), + ("sunsynk_load_cached_data_no_storage_leaves_defaults", test_sunsynk_load_cached_data_no_storage_leaves_defaults), + ("sunsynk_save_then_load_cache_restores_state", test_sunsynk_save_then_load_cache_restores_state), + ("run_first_calls_load_cached_data_before_get_device_list", test_run_first_calls_load_cached_data_before_get_device_list), + ("run_second_cycle_does_not_call_load_cached_data", test_run_second_cycle_does_not_call_load_cached_data), + ("sunsynk_rsa_encrypt_password_round_trips_with_matching_private_key", test_sunsynk_rsa_encrypt_password_round_trips_with_matching_private_key), + ("sunsynk_build_login_payload_shape_and_sign", test_sunsynk_build_login_payload_shape_and_sign), + ("sunsynk_rsa_encrypt_password_missing_cryptography_raises_clear_error", test_sunsynk_rsa_encrypt_password_missing_cryptography_raises_clear_error), + ("sunsynk_module_has_no_top_level_cryptography_import", test_sunsynk_module_has_no_top_level_cryptography_import), + ]: + try: + if fn(): + print(f" FAILED: sunsynk.{name}") + failed = True + except Exception as e: + print(f" EXCEPTION in sunsynk.{name}: {e}") + import traceback + + traceback.print_exc() + failed = True + return failed diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index 64e68b3d9..3522f89f7 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -114,6 +114,7 @@ from tests.test_deye_oauth import run_deye_oauth_tests from tests.test_deye_control import run_deye_control_tests from tests.test_deye_publish import run_deye_publish_tests +from tests.test_sunsynk import run_sunsynk_tests from tests.test_enphase_api import run_enphase_api_tests from tests.test_solcast import run_solcast_tests from tests.test_open_meteo import run_open_meteo_tests @@ -300,6 +301,7 @@ def main(): ("deye_oauth", run_deye_oauth_tests, "DEYE auth tests", False), ("deye_control", run_deye_control_tests, "DEYE control-logic tests", False), ("deye_publish", run_deye_publish_tests, "DEYE publish/config tests", False), + ("sunsynk", run_sunsynk_tests, "Sunsynk cloud inverter tests", False), ("enphase_api", run_enphase_api_tests, "Enphase API tests", False), ("solcast", run_solcast_tests, "Solcast API tests", False), ("open_meteo", run_open_meteo_tests, "Open-Meteo solar forecast provider tests", False), diff --git a/requirements.txt b/requirements.txt index 814314f4f..9a279cc88 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,7 @@ aiofiles aiohttp aiomqtt coverage +cryptography matplotlib numpy==2.3.5 pre-commit