"""API client for Solar of Things (solar.siseli.com). Authentication strategy ───────────────────────────────────────────────────────────────────────────── The Siseli portal uses a dual-token system. Login requires the IOT Open Platform signing scheme discovered via JS bundle analysis (umi.js): Endpoint : POST https://test.solar.siseli.com/apis/login/account Signed : yes — IOT-Open-AppID, IOT-Open-Nonce, IOT-Open-Body-Hash, IOT-Open-Sign headers Signing algorithm (reverse-engineered from portal umi.js): 1. Build a dict of signing headers: {"IOT-Open-AppID": appId, "IOT-Open-Body-Hash": sha256(body_bytes).lower(), "IOT-Open-Nonce": random_32_char_hex} 2. Sort keys alphabetically, join as k1=v1&k2=v2 (no URL-encoding). 3. base64-encode the resulting string. 4. HMAC-SHA256(b64_str, decrypted_app_secret) [bytes result] 5. MD5(hmac_bytes).hexdigest() → IOT-Open-Sign value App secret decryption (qe() in umi.js): key = MD5(appId).lower()[:16] treated as ASCII bytes (16 bytes = AES-128) iv = MD5(appId).lower()[16:] treated as ASCII bytes (16 bytes) AES-128-CBC-ZeroPadding decrypt of base64(encrypted_secret) After successful login the server returns an accessToken (used as IOT-Token header for data requests) and a refreshToken. This class supports three auth modes, tried in priority order: 1. User-ID + password (recommended) • Call login() at startup → stores both tokens in memory. • _ensure_token_valid() checks expiry before every API call and proactively refreshes (TOKEN_REFRESH_LEAD_SECONDS = 5 min before expiry, mirroring the portal JS behaviour). • If refresh fails, raises TokenExpiredError so the HA integration can trigger a re-auth flow. 2. Token-pair (accessToken + refreshToken) without password • User pastes both tokens from DevTools. • Same proactive-refresh logic; re-auth needed when refreshToken expires. 3. Legacy IOT-token only (backwards compatibility) • No refresh possible; raises TokenExpiredError on 401 so HA can prompt the user to re-enter a fresh token. Usage in Home Assistant ───────────────────────────────────────────────────────────────────────────── api = SolarOfThingsAPI( user_id="myaccount", password="secret", # or omit and pass iot_token= time_zone="Asia/Manila", on_token_refreshed=_save_tokens_callback, ) await hass.async_add_executor_job(api.login) data = await hass.async_add_executor_job(api.fetch_latest_data, device_id) """ from __future__ import annotations import base64 import hashlib import hmac as _hmac import logging import os import threading from datetime import datetime, timezone, timedelta from typing import Any, Callable import requests try: from Crypto.Cipher import AES as _AES _CRYPTO_AVAILABLE = True except ImportError: _CRYPTO_AVAILABLE = False try: from zoneinfo import ZoneInfo except Exception: # pragma: no cover ZoneInfo = None # type: ignore[assignment] from .const import ( API_BASE_URL, API_AUTH_BASE_URL, API_LOGIN, API_REFRESH_TOKEN as API_REFRESH_TOKEN_ENDPOINT, API_TIME_SERIES, API_MONTHLY_SUMMARY, API_DEVICE_LIST, API_SETTINGS_GET, API_SETTINGS_SET, API_ENERGY_FLOW, ENERGY_FLOW_RULES, REALTIME_PROBE_KEYS, IOT_APP_ID, IOT_APP_SECRET_ENC, TOKEN_REFRESH_LEAD_SECONDS, ) _LOGGER = logging.getLogger(__name__) _DEFAULT_TZ = "Asia/Manila" # ────────────────────────────────────────────────────────────────────────────── # Custom exceptions # ────────────────────────────────────────────────────────────────────────────── class TokenExpiredError(Exception): """Raised when the access token has expired and cannot be refreshed. The HA integration should catch this and call config_entry.async_start_reauth() so the user can re-enter credentials. """ class AuthenticationError(Exception): """Raised when login credentials are rejected by the server.""" class EnergyFlowRuleNotConfiguredError(RuntimeError): """Raised when the portal returns code 70132 ("Energy flow rule not exists"). This means the device's *account* has no energy-flow rule configured on the portal side — a per-device setup step on solar.siseli.com, not something this integration can fix by mapping a different field. Reported in issue #21: without this, the fallback failing this way looked identical to "no data returned", so a device in this state got no sensors and no warning either. """ # ────────────────────────────────────────────────────────────────────────────── # Signing helpers (reverse-engineered from portal umi.js) # ────────────────────────────────────────────────────────────────────────────── def _decrypt_app_secret(app_id: str, encrypted_b64: str) -> str: """AES-128-CBC decrypt the embedded app secret. Key derivation mirrors the portal qe() function: key = MD5(app_id).lower()[:16] as ASCII bytes iv = MD5(app_id).lower()[16:] as ASCII bytes The ciphertext is the base64-decoded encrypted_b64 value. """ if not _CRYPTO_AVAILABLE: raise RuntimeError( "pycryptodome is not installed. " "Add 'pycryptodome' to the integration requirements." ) md5_hex = hashlib.md5(app_id.encode("utf-8")).hexdigest() key = md5_hex[:16].encode("ascii") # 16 bytes — AES-128 iv = md5_hex[16:].encode("ascii") # 16 bytes — CBC IV ciphertext = base64.b64decode(encrypted_b64) cipher = _AES.new(key, _AES.MODE_CBC, iv) plaintext = cipher.decrypt(ciphertext).rstrip(b"\x00") return plaintext.decode("utf-8") def _compute_iot_sign(app_id: str, nonce: str, body_hash: str, secret: str) -> str: """Compute the IOT-Open-Sign header value. Algorithm (Ye() in portal umi.js): 1. Sort signing headers alphabetically by key. 2. qs.stringify → "IOT-Open-AppID=X&IOT-Open-Body-Hash=Y&IOT-Open-Nonce=Z" 3. Base64-encode the qs string. 4. HMAC-SHA256(b64_qs, secret) → raw bytes. 5. MD5(hmac_bytes).hexdigest() → sign value. """ sign_headers = { "IOT-Open-AppID": app_id, "IOT-Open-Body-Hash": body_hash, "IOT-Open-Nonce": nonce, } qs_str = "&".join(f"{k}={sign_headers[k]}" for k in sorted(sign_headers.keys())) b64_qs = base64.b64encode(qs_str.encode("utf-8")).decode("ascii") hmac_bytes = _hmac.new(secret.encode("utf-8"), b64_qs.encode("utf-8"), hashlib.sha256).digest() return hashlib.md5(hmac_bytes).hexdigest() def _make_signed_headers(body_bytes: bytes, extra: dict[str, str] | None = None) -> dict[str, str]: """Build the complete set of IOT Open Platform signed request headers. Returns headers suitable for POST to API_AUTH_BASE_URL endpoints. """ secret = _decrypt_app_secret(IOT_APP_ID, IOT_APP_SECRET_ENC) nonce = os.urandom(16).hex() # 32-char hex nonce body_hash = hashlib.sha256(body_bytes).hexdigest() sign = _compute_iot_sign(IOT_APP_ID, nonce, body_hash, secret) headers: dict[str, str] = { "Accept": "application/json", "Content-Type": "application/json; charset=utf-8", "Origin": "https://solar.siseli.com", "Referer": "https://solar.siseli.com/", "IOT-Open-AppID": IOT_APP_ID, "IOT-Open-Nonce": nonce, "IOT-Open-Body-Hash": body_hash, "IOT-Open-Sign": sign, } if extra: headers.update(extra) return headers # ────────────────────────────────────────────────────────────────────────────── # Helper: parse Siseli ISO expiry strings safely # ────────────────────────────────────────────────────────────────────────────── def _parse_expiry(value: str | None) -> datetime | None: """Return an aware UTC datetime from an ISO-8601 string, or None.""" if not value: return None try: # Python 3.7+ fromisoformat doesn't handle trailing 'Z' cleaned = value.replace("Z", "+00:00") dt = datetime.fromisoformat(cleaned) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) except Exception: return None # ────────────────────────────────────────────────────────────────────────────── # Energy-flow fallback mapping # ────────────────────────────────────────────────────────────────────────────── def _coerce_number(value: Any) -> float | None: """Best-effort conversion of an API field value to a float. The energy-flow payload has only been observed second-hand (issue #7), so accept every shape this portal is known to use elsewhere: a bare number, a numeric string, a latest-wins list (as the time-series endpoint returns), or a {"value": ...} wrapper. Anything unrecognised yields None so the caller leaves the sensor untouched instead of publishing a garbage reading. """ if value is None or isinstance(value, bool): return None if isinstance(value, (int, float)): return float(value) if isinstance(value, str): text = value.strip() if not text: return None try: return float(text) except ValueError: return None if isinstance(value, list): for item in reversed(value): number = _coerce_number(item) if number is not None: return number return None if isinstance(value, dict): for key in ("value", "val", "latest"): if key in value: return _coerce_number(value[key]) return None def map_energy_flow_fields(fields: Any) -> dict[str, float]: """Translate an energy-flow ``fields`` mapping into canonical sensor keys. Pure function: no network access and no instance state, so it can be tested directly. Applies ENERGY_FLOW_RULES in declaration order and returns only the keys it could actually resolve — the caller merges the result without clobbering better data from the time-series endpoint. """ if not isinstance(fields, dict) or not fields: return {} mapped: dict[str, float] = {} for canonical, rules in ENERGY_FLOW_RULES.items(): for mode, sources, scale in rules: if mode in ("sum", "clamp_pos", "clamp_neg"): values = [ number for source in sources if (number := _coerce_number(fields.get(source))) is not None ] if values: total = sum(values) * scale if mode == "clamp_pos": # Positive/export half of a signed field (e.g. mains # power going positive = feeding back to the grid). total = max(0.0, total) elif mode == "clamp_neg": # Negative/import half of the same signed field, # flipped positive for display (e.g. mains power # going negative = drawing from the grid). total = max(0.0, -total) mapped[canonical] = total else: # "first" — first present source field wins for source in sources: number = _coerce_number(fields.get(source)) if number is not None: mapped[canonical] = number * scale break if canonical in mapped: break return mapped def has_realtime_values(values: Any) -> bool: """Return True if any canonical realtime key carries a usable number.""" if not isinstance(values, dict): return False return any( _coerce_number(values.get(key)) is not None for key in REALTIME_PROBE_KEYS ) # ────────────────────────────────────────────────────────────────────────────── # Main API client # ────────────────────────────────────────────────────────────────────────────── class SolarOfThingsAPI: """Solar of Things API wrapper with automatic token refresh. Parameters ---------- user_id: Siseli portal login account / user-ID (preferred auth method). password: Siseli portal password. iot_token: Legacy/manual IOT-Token (used when user_id/password absent). refresh_token: Stored refresh token (persisted between HA restarts). access_token_expires: ISO-8601 string of current access-token expiry. refresh_token_expires: ISO-8601 string of current refresh-token expiry. time_zone: IOT-Time-Zone header value. on_token_refreshed: Optional callback(access_token, refresh_token, access_expires_iso, refresh_expires_iso) called after every successful token refresh so the HA entry can persist the new tokens without restarting. """ def __init__( self, *, user_id: str | None = None, password: str | None = None, iot_token: str | None = None, refresh_token: str | None = None, access_token_expires: str | None = None, refresh_token_expires: str | None = None, time_zone: str | None = None, on_token_refreshed: Callable[[str, str, str, str], None] | None = None, ) -> None: self._user_id = user_id self._password = password self._time_zone = time_zone or _DEFAULT_TZ self._on_token_refreshed = on_token_refreshed # Token state self._access_token: str = iot_token or "" self._refresh_token: str = refresh_token or "" self._access_expires: datetime | None = _parse_expiry(access_token_expires) self._refresh_expires: datetime | None = _parse_expiry(refresh_token_expires) # Thread-safety for concurrent refresh calls self._refresh_lock = threading.Lock() # Determine auth mode if user_id and password: self._auth_mode = "password" elif iot_token and refresh_token: self._auth_mode = "token_pair" elif iot_token: self._auth_mode = "legacy" else: raise ValueError("Provide either (user_id + password) or iot_token.") # HTTP session (headers updated after every token refresh) self.session = requests.Session() self._apply_token_headers() # ─── Session headers ─────────────────────────────────────────────────────── def _apply_token_headers(self) -> None: """Write the current access token into the session headers.""" self.session.headers.update( { "Accept": "application/json", "Content-Type": "application/json; charset=utf-8", "IOT-Token": self._access_token, "IOT-Time-Zone": self._time_zone, "Origin": "https://solar.siseli.com", "Referer": "https://solar.siseli.com/", # Keep this version in sync with manifest.json on each release. "User-Agent": ( "HomeAssistant-SolarOfThings/2.5.0 " "(+https://github.com/Conexo-Casa/solar-of-things-ha)" ), } ) # ─── Public auth helpers ─────────────────────────────────────────────────── def login(self) -> None: """Authenticate with user-ID + password and store the resulting tokens. Uses the IOT Open Platform signed request format discovered from the portal JS bundle. The login endpoint is: POST https://solar.siseli.com/apis/login/account The password is sent as MD5(plaintext_password) lowercase hex — this is how the portal processes it before transmitting. Raises AuthenticationError on bad credentials, or requests.RequestException on network failure. Safe to call from a background thread. """ if self._auth_mode not in ("password",): raise RuntimeError("login() requires user_id + password auth mode.") _LOGGER.debug("SolarOfThings: logging in as %s", self._user_id) import json as _json # The Siseli portal transmits the password as MD5(plaintext) lowercase hex — # the server rejects plaintext with error code 7. This is a protocol # requirement of the upstream API, not a choice we can change. The hash # is sent over HTTPS, so the transport layer provides confidentiality. # CodeQL alert suppressed: MD5 use here is non-cryptographic (protocol-mandated # pre-hashing by the upstream service), not used for storage or key derivation. password_md5 = hashlib.md5(self._password.encode("utf-8")).hexdigest() # noqa: S324 payload = { "account": self._user_id, "password": password_md5, } body_bytes = _json.dumps(payload, separators=(",", ":")).encode("utf-8") headers = _make_signed_headers(body_bytes) resp = requests.post( f"{API_AUTH_BASE_URL}{API_LOGIN}", data=body_bytes, headers=headers, timeout=30, ) resp.raise_for_status() data = resp.json() if data.get("code") not in (0, None, "0"): msg = data.get("message") or data.get("msg") or str(data) raise AuthenticationError(f"Login failed: {msg}") self._store_tokens(data.get("data") or data) def refresh_access_token(self) -> None: """Use the stored refresh token to obtain a new access token. Raises TokenExpiredError if the refresh token is also expired or invalid. """ if not self._refresh_token: raise TokenExpiredError("No refresh token available.") _LOGGER.debug("SolarOfThings: refreshing access token") resp = requests.post( f"{API_AUTH_BASE_URL}{API_REFRESH_TOKEN_ENDPOINT}", json={"refreshToken": self._refresh_token}, headers={ "Accept": "application/json", "Content-Type": "application/json; charset=utf-8", "Origin": "https://solar.siseli.com", "Referer": "https://solar.siseli.com/", }, timeout=30, ) if resp.status_code in (401, 403): raise TokenExpiredError("Refresh token rejected by server (expired or invalid).") resp.raise_for_status() data = resp.json() if data.get("code") not in (0, None, "0"): raise TokenExpiredError( f"Refresh failed: code={data.get('code')} message={data.get('message')}" ) self._store_tokens(data.get("data") or data) # ─── Internal token management ───────────────────────────────────────────── def _store_tokens(self, payload: dict[str, Any]) -> None: """Extract tokens from a login/refresh response payload and persist them.""" access = ( payload.get("accessToken") or payload.get("iotToken") or payload.get("token") or "" ) refresh = payload.get("refreshToken") or "" access_exp = ( payload.get("accessTokenWillExpiredAt") or payload.get("accessTokenExpiredAt") or "" ) refresh_exp = ( payload.get("refreshTokenWillExpiredAt") or payload.get("refreshTokenExpiredAt") or "" ) if not access: raise AuthenticationError( f"Login/refresh response did not contain an access token. " f"Keys received: {list(payload.keys())}" ) self._access_token = access self._refresh_token = refresh self._access_expires = _parse_expiry(access_exp) self._refresh_expires = _parse_expiry(refresh_exp) # Update session header immediately self._apply_token_headers() _LOGGER.debug( "SolarOfThings: token updated, expires=%s", self._access_expires.isoformat() if self._access_expires else "unknown", ) # Notify the HA integration so it can persist the new token state if self._on_token_refreshed: try: self._on_token_refreshed( self._access_token, self._refresh_token, self._access_expires.isoformat() if self._access_expires else "", self._refresh_expires.isoformat() if self._refresh_expires else "", ) except Exception as cb_err: # pragma: no cover _LOGGER.warning("Token-refresh callback raised: %s", cb_err) def _token_needs_refresh(self) -> bool: """Return True if the access token is absent or about to expire.""" if not self._access_token: return True if self._access_expires is None: # Unknown expiry: only refresh if we already have a refresh token return bool(self._refresh_token) lead = timedelta(seconds=TOKEN_REFRESH_LEAD_SECONDS) return datetime.now(timezone.utc) >= (self._access_expires - lead) def _ensure_token_valid(self) -> None: """Proactively refresh the access token if needed. Thread-safe: uses a lock so parallel coordinator updates don't trigger multiple simultaneous refresh calls. Raises TokenExpiredError when all refresh strategies are exhausted. """ if not self._token_needs_refresh(): return with self._refresh_lock: # Double-check inside the lock (another thread may have refreshed) if not self._token_needs_refresh(): return _LOGGER.info("SolarOfThings: access token expiring; attempting refresh") # Strategy 1: use refresh token if self._refresh_token: try: self.refresh_access_token() return except TokenExpiredError: _LOGGER.warning( "SolarOfThings: refresh token expired/invalid; " "attempting re-login" ) except Exception as err: _LOGGER.error("SolarOfThings: token refresh request failed: %s", err) # Strategy 2: re-login with stored credentials if self._auth_mode == "password" and self._user_id and self._password: try: self.login() return except AuthenticationError as err: raise TokenExpiredError( f"Re-login failed (credentials rejected): {err}" ) from err except Exception as err: raise TokenExpiredError( f"Re-login failed (network error): {err}" ) from err # Strategy 3: nothing left — tell HA to trigger re-auth raise TokenExpiredError( "Access token expired and no refresh strategy succeeded. " "Please re-authenticate in Home Assistant." ) # ─── Internal HTTP helper ────────────────────────────────────────────────── def _post(self, path: str, payload: dict[str, Any], *, timeout: int = 30) -> dict[str, Any]: """Perform a POST to a data endpoint, automatically refreshing the token on 401. On second 401 (after refresh) raises TokenExpiredError. Uses API_BASE_URL (solar.siseli.com) — not the auth base URL. """ self._ensure_token_valid() resp = self.session.post(f"{API_BASE_URL}{path}", json=payload, timeout=timeout) if resp.status_code == 401: _LOGGER.warning("SolarOfThings: received 401; forcing token refresh") # Force an immediate refresh even if _token_needs_refresh() is False self._access_expires = None self._ensure_token_valid() resp = self.session.post(f"{API_BASE_URL}{path}", json=payload, timeout=timeout) resp.raise_for_status() return resp.json() def _get(self, path: str, params: dict[str, Any], *, timeout: int = 30) -> dict[str, Any]: """Perform a GET to a data endpoint, refreshing the token once on 401. Mirrors _post's retry behaviour: on a second 401 the nested _ensure_token_valid raises TokenExpiredError, which the coordinator turns into a re-auth flow. """ self._ensure_token_valid() url = f"{API_BASE_URL}{path}" resp = self.session.get(url, params=params, timeout=timeout) if resp.status_code == 401: _LOGGER.warning("SolarOfThings: received 401 on GET; forcing token refresh") # Force an immediate refresh even if _token_needs_refresh() is False self._access_expires = None self._ensure_token_valid() resp = self.session.get(url, params=params, timeout=timeout) resp.raise_for_status() return resp.json() # ─── Public properties (for persistence in HA config entry) ─────────────── @property def access_token(self) -> str: return self._access_token @property def refresh_token(self) -> str: return self._refresh_token @property def access_token_expires_iso(self) -> str: return self._access_expires.isoformat() if self._access_expires else "" @property def refresh_token_expires_iso(self) -> str: return self._refresh_expires.isoformat() if self._refresh_expires else "" # ─── Time helpers ────────────────────────────────────────────────────────── def _now(self) -> datetime: if ZoneInfo: try: return datetime.now(tz=ZoneInfo(self._time_zone)) except Exception: return datetime.now() return datetime.now() def _format_time(self, dt: datetime) -> str: if ZoneInfo: try: dt = dt.astimezone(ZoneInfo(self._time_zone)) except Exception: pass return dt.replace(microsecond=0).isoformat() # ─── Station → device listing ────────────────────────────────────────────── def list_devices(self, station_id: str, page_size: int = 50) -> list[dict[str, Any]]: """Return all devices under a station (paginated).""" devices: list[dict[str, Any]] = [] page = 1 total: int | None = None while True: data = self._post( API_DEVICE_LIST, {"page": page, "count": page_size, "stationId": station_id}, ) if data.get("code") not in (0, None): raise RuntimeError( f"Device list error code={data.get('code')} " f"message={data.get('message')}" ) d = data.get("data") or {} total = d.get("total", total) batch = d.get("list") or [] if not isinstance(batch, list): batch = [] devices.extend(batch) if total is None: if len(batch) < page_size: break else: if len(devices) >= int(total): break if not batch: break page += 1 return devices # ─── Time-series (per device) ────────────────────────────────────────────── def fetch_latest_data(self, device_id: str) -> dict[str, Any]: """Fetch the latest readings for a device (last 1 hour). The historical time-series endpoint is the primary source. Several inverter / WiFi-dongle firmware families never populate it, which leaves every realtime entity "unknown" even though the portal shows live data (issue #7). For those devices we fall back to the live energy-flow endpoint and translate its field names onto the canonical sensor keys. """ latest_values = self._fetch_time_series_values(device_id) if not has_realtime_values(latest_values): _LOGGER.debug( "SolarOfThings device %s: time-series returned no realtime " "values; trying energy-flow fallback", device_id, ) try: fields = self.fetch_energy_flow(device_id) except TokenExpiredError: # Must reach the coordinator so it can start the re-auth flow. raise except EnergyFlowRuleNotConfiguredError as err: # A portal-side setup gap, not something a field mapping can # fix — but "no sensors, no warnings" (issue #21) looked # identical to a bug, so this needs to be loud by default. _LOGGER.warning( "SolarOfThings device %s: no energy-flow rule is configured " "for this device in the Siseli portal (%s). Realtime sensors " "will stay unavailable until an energy-flow rule is set up " "for it there, or its historical time-series endpoint starts " "returning data instead.", device_id, err, ) except Exception as err: _LOGGER.debug( "SolarOfThings device %s: energy-flow fallback unavailable: %s", device_id, err, ) else: mapped = map_energy_flow_fields(fields) if mapped: _LOGGER.debug( "SolarOfThings device %s: energy-flow fallback resolved %s", device_id, sorted(mapped), ) # Only fill gaps — never clobber a time-series reading. for key, value in mapped.items(): latest_values.setdefault(key, value) elif fields: # Unknown firmware variant: surface the field names so they # can be added to ENERGY_FLOW_RULES from a bug report. _LOGGER.warning( "SolarOfThings device %s: energy-flow returned %d field(s) " "but none matched a known mapping. Please open an issue " "with these key names: %s", device_id, len(fields), sorted(fields), ) self._apply_derived_values(latest_values) return latest_values def _fetch_time_series_values(self, device_id: str) -> dict[str, Any]: """Return the latest value per key from the historical time-series API.""" end_time = self._now() start_time = end_time - timedelta(hours=1) # Some inverter models (e.g. Siseli HPVINV02 / "Inverter Top One" # gather protocol) report these three metrics under different key # names than the vendor's documented API. Request both the # documented key and the known alternate, and prefer whichever the # device actually populates — keeps this working for devices that # use either naming instead of hardcoding one over the other. ALIAS_GROUPS = [ ("pvInputPower", "pvPower"), ("acOutputActivePower", "outputActivePower"), ("batterySOC", "batteryCapacity"), ] keys = [ "pvInputPower", "pvPower", "acOutputActivePower", "outputActivePower", "batteryDischargeCurrent", "batteryChargingCurrent", "batteryVoltage", "feedInPower", "batterySOC", "batteryCapacity", ] request_body = { "deviceId": device_id, "count": 2000, "page": 1, "fromTime": self._format_time(start_time), "toTime": self._format_time(end_time), "orderByTimeAsc": True, "keys": keys, } data = self._post(API_TIME_SERIES, request_body) if data.get("code") not in (0, None): raise RuntimeError( f"Timeseries error code={data.get('code')} " f"message={data.get('message')}" ) payload_data = (data.get("data") or {}).get("payload") or {} fields = payload_data.get("fields") or {} latest_values: dict[str, Any] = {} for key, arr in fields.items(): if isinstance(arr, list) and arr: latest_values[key] = arr[-1] # Some inverter models (e.g. Siseli HPVINV02) report pvInputPower, # acOutputActivePower and batterySOC under an alternate key name # instead of the documented one. Prefer the canonical key when the # device populates it; fall back to the alternate otherwise. aliased_keys: set[str] = set() for canonical, alternate in ALIAS_GROUPS: if latest_values.get(canonical) is None and latest_values.get(alternate) is not None: latest_values[canonical] = latest_values[alternate] aliased_keys.add(canonical) latest_values.pop(alternate, None) # Unit normalisation: acOutputActivePower is always kW in API → W. # pvInputPower is W under the documented key on every device this # integration already supports (pinned by # test_working_device_is_unaffected) but kW when it arrived via the # pvPower alias (confirmed on HPVINV02) — convert only in that case, # so devices that were already working are unaffected. if "acOutputActivePower" in latest_values: converted = _coerce_number(latest_values["acOutputActivePower"]) if converted is not None: latest_values["acOutputActivePower"] = converted * 1000.0 if "pvInputPower" in aliased_keys: converted = _coerce_number(latest_values["pvInputPower"]) if converted is not None: latest_values["pvInputPower"] = converted * 1000.0 return latest_values def fetch_energy_flow(self, device_id: str) -> dict[str, Any]: """Return the live energy-flow ``fields`` mapping for a device. Fallback source for firmware that never populates the time-series endpoint (issue #7). Returns an empty dict when the endpoint responds without field data, so "no data" and "unsupported" behave identically. """ data = self._get(API_ENERGY_FLOW, {"deviceId": device_id, "dataSource": 1}) code = data.get("code") if code not in (0, None, "0"): message = data.get("message") or data.get("msg") if code in (70132, "70132"): raise EnergyFlowRuleNotConfiguredError( f"Energy-flow error code={code} message={message}" ) raise RuntimeError(f"Energy-flow error code={code} message={message}") payload = data.get("data") or {} state = payload.get("deviceAttributeState") or {} fields = state.get("fields") if not isinstance(fields, dict): # Tolerate the values being nested directly under data.fields. fields = payload.get("fields") return fields if isinstance(fields, dict) else {} @staticmethod def _apply_derived_values(latest_values: dict[str, Any]) -> None: """Compute values the API does not report directly. Gap-filling only: when a source such as the energy-flow fallback already supplied a real measurement, it is kept rather than overwritten with an estimate. On the time-series path none of these keys are ever returned by the API, so every one of them is still derived exactly as before. """ if latest_values.get("batteryPower") is None: voltage = _coerce_number(latest_values.get("batteryVoltage")) or 0.0 discharge = _coerce_number(latest_values.get("batteryDischargeCurrent")) or 0.0 charge = _coerce_number(latest_values.get("batteryChargingCurrent")) or 0.0 latest_values["batteryPower"] = (discharge - charge) * voltage ac_output = _coerce_number(latest_values.get("acOutputActivePower")) or 0.0 if latest_values.get("loadPower") is None: latest_values["loadPower"] = ac_output if latest_values.get("gridPower") is None: pv_power = _coerce_number(latest_values.get("pvInputPower")) or 0.0 feed_in = _coerce_number(latest_values.get("feedInPower")) or 0.0 battery_power = _coerce_number(latest_values.get("batteryPower")) or 0.0 latest_values["gridPower"] = max( 0.0, ac_output - pv_power + battery_power + feed_in ) # ─── Monthly summary (station) ───────────────────────────────────────────── def fetch_monthly_summary(self, station_id: str) -> dict[str, Any]: """Fetch monthly PV summary for the current month.""" now = self._now() year = now.year month_key = f"{year}-{str(now.month).zfill(2)}" self._ensure_token_valid() resp = self.session.post( f"{API_BASE_URL}{API_MONTHLY_SUMMARY}" f"?stationId={station_id}&summaryCategoryKey=pvInverterElectricityQuantityClass", json={"time": str(year)}, timeout=30, ) resp.raise_for_status() data = resp.json() if data.get("code") not in (0, None): raise RuntimeError( f"Monthly summary error code={data.get('code')} " f"message={data.get('message')}" ) props = (((data.get("data") or {}).get("properties")) or (data.get("data") or {}).get("list") or []) result: dict[str, Any] = {} for item in props if isinstance(props, list) else []: k = item.get("key") or item.get("name") v = item.get("value") if k and v is not None: result[k] = v # Extract monthly totals (fallback: look for known keys) monthly: dict[str, Any] = {} pv_total = result.get(month_key) or result.get("pvTotal") or result.get("pv") or 0 monthly["monthly_pv_generated"] = float(pv_total or 0) grid_import = result.get("gridImport") or result.get("buy") or 0 monthly["monthly_grid_import"] = float(grid_import or 0) total_consumption = result.get("totalConsumption") or result.get("load") or 0 monthly["monthly_total_consumption"] = float(total_consumption or 0) if monthly["monthly_total_consumption"] > 0: monthly["monthly_solar_percentage"] = round( 100.0 * monthly["monthly_pv_generated"] / monthly["monthly_total_consumption"], 1 ) else: monthly["monthly_solar_percentage"] = 0.0 return monthly # ─── Device settings ─────────────────────────────────────────────────────── # The remote config endpoints require only a plain IOT-Token header (which the # session already carries) and pass deviceId as a URL query parameter rather # than in the JSON body. def _write_setting(self, device_id: str, key: str, value: Any) -> None: """Write a single device setting key=value via the remote config write API.""" self._ensure_token_valid() url = f"{API_BASE_URL}{API_SETTINGS_SET}?deviceId={device_id}" payload = {"deviceId": device_id, "key": key, "value": value} resp = self.session.post(url, json=payload, timeout=30) resp.raise_for_status() data = resp.json() if data.get("code") not in (0, None): raise RuntimeError( f"Settings write error code={data.get('code')} " f"message={data.get('message')} (key={key})" ) def get_device_settings(self, device_id: str) -> dict[str, Any]: """Fetch the cached device settings from the remote config API. Returns a flat dict of {settingKey: settingObject} where each value contains at least 'key', 'value', and 'valueDisplay' fields. The endpoint accepts a plain IOT-Token header (no IOT-Open-Sign). """ self._ensure_token_valid() url = f"{API_BASE_URL}{API_SETTINGS_GET}?deviceId={device_id}" resp = self.session.post(url, json={}, timeout=30) resp.raise_for_status() data = resp.json() if data.get("code") not in (0, None): raise RuntimeError( f"Settings fetch error code={data.get('code')} " f"message={data.get('message')}" ) return data.get("data") or {} # Alias used by the coordinator in __init__.py fetch_settings = get_device_settings def update_device_settings(self, device_id: str, settings: dict[str, Any]) -> None: """Write multiple settings (one API call per key).""" for key, value in settings.items(): self._write_setting(device_id, key, value) # ─── Convenience control helpers (called by select.py / switch.py) ───────── # Key names are the real device attribute keys returned by get_device_settings. # Output Source Priority: USO=0, SUB=1, SBU=2 # Charger Source Priority: CSO=0, SNU=1, OSO=2 # batteryPowerLimitingSetting: 0=OFF, 1=ON (GRID switch) # acInputRangeSetting: 0=Appliance, 1=UPS # Operating-mode select maps HA option strings to integer values _OUTPUT_MODE_MAP: dict[str, int] = { "Utility First (USO)": 0, "Solar First (SUB)": 1, "Solar+Battery First (SBU)": 2, } _OUTPUT_MODE_REVERSE: dict[int, str] = {v: k for k, v in _OUTPUT_MODE_MAP.items()} # Charger-priority select _CHARGER_PRIORITY_MAP: dict[str, int] = { "Solar + Utility (CSO)": 0, "Solar First (SNU)": 1, "Solar Only (OSO)": 2, } _CHARGER_PRIORITY_REVERSE: dict[int, str] = {v: k for k, v in _CHARGER_PRIORITY_MAP.items()} def set_operating_mode(self, device_id: str, mode: str) -> None: """Set Output Source Priority. mode is one of _OUTPUT_MODE_MAP keys.""" value = self._OUTPUT_MODE_MAP.get(mode) if value is None: raise ValueError(f"Unknown operating mode: {mode!r}. " f"Valid options: {list(self._OUTPUT_MODE_MAP)!r}") self._write_setting(device_id, "outputSourcePrioritySetting", value) def set_battery_priority(self, device_id: str, mode: str) -> None: """Set Charger Source Priority. mode is one of _CHARGER_PRIORITY_MAP keys.""" value = self._CHARGER_PRIORITY_MAP.get(mode) if value is None: raise ValueError(f"Unknown battery priority: {mode!r}. " f"Valid options: {list(self._CHARGER_PRIORITY_MAP)!r}") self._write_setting(device_id, "chargerSourcePrioritySetting", value) def set_grid_charging(self, device_id: str, enabled: bool) -> None: """Set AC Input Range: Appliance (0, grid charging allowed) / UPS (1, bypass).""" self._write_setting(device_id, "acInputRangeSetting", 0 if enabled else 1) def set_grid_feed_in(self, device_id: str, enabled: bool) -> None: """Enable or disable the GRID grid switch (batteryPowerLimitingSetting).""" self._write_setting(device_id, "batteryPowerLimitingSetting", 1 if enabled else 0) def set_backup_mode(self, device_id: str, enabled: bool) -> None: """Set Output Source Priority to SBU (backup/off-grid priority) when True, or SUB (solar-first, grid-supplemented) when False.""" value = 2 if enabled else 1 # SBU=2 (battery before grid), SUB=1 self._write_setting(device_id, "outputSourcePrioritySetting", value) def set_battery_charge_limit(self, device_id: str, percent: int) -> None: """Set battery charge limit (0–100 %).""" self._write_setting(device_id, "batteryChargeLimit", percent) def set_battery_discharge_limit(self, device_id: str, percent: int) -> None: """Set battery discharge limit / minimum SOC (0–100 %).""" self._write_setting(device_id, "batteryDischargeLimit", percent) def set_grid_charge_limit(self, device_id: str, watts: int) -> None: """Set maximum grid charge power (0–5000 W).""" self._write_setting(device_id, "gridChargeLimit", watts) def test_connection(self, station_id: str) -> bool: """Return True if we can reach the device-list endpoint successfully.""" try: devices = self.list_devices(station_id, page_size=1) return True except Exception as err: _LOGGER.error("SolarOfThings: connection test failed: %s", err) return False