"""Real iSolarCloud client -- "V1 account login" (non-OAuth2) app. Implements the exact same `fetch_interval_data(since, until)` interface as `mock_client.MockISolarCloudClient` so the scheduler can swap between them via `settings.use_mock_isolarcloud` with no other code changes. Auth flow, headers and the core endpoints (`/openapi/login`, `/openapi/getPowerStationList`, `/openapi/getDeviceList`, `/openapi/getDevicePointMinuteDataList`, `/openapi/getDeviceRealTimeData`) are taken from steven-g-w/HA-Sungrow (github.com/steven-g-w/HA-Sungrow), a Home Assistant integration built directly against https://developer-api.isolarcloud.com/ using this same "V1 login" app type (as opposed to the separate OAuth2 app type, which uses different `/openapi/platform/...` endpoints and an authorization-code browser flow -- not what an app with an App Key + Secret Key + username/password uses). What's NOT independently verified here (confirm with scripts/verify_isolarcloud_cli.py against your real account before trusting the numbers -- see SETUP.md): - RSA password encryption. Some V1-login apps are issued an RSA public key (Settings -> Open API in iSolarCloud) and are expected to encrypt the password before sending it; others aren't and send it as plain text over HTTPS. The exact padding scheme Sungrow expects isn't published anywhere public -- `_encrypt_password` below uses the common RSA/PKCS1v1.5 + base64 convention as a best effort. If login fails with an RSA key configured, try without one (some accounts don't need it) or adjust the padding here based on what the portal's own "Documentation" page (visible after logging in) specifies for the login endpoint. - Which device serves generation vs. import/export/load. iSolarCloud reports these per *device* (inverter, meter, ...), not per plant, in this API family. `fetch_interval_data` picks the first METER device for import/export/load and the first INVERTER for generation, which is correct for the common single-inverter, grid-meter household setup this app targets, but should be confirmed with `scripts/verify_isolarcloud_cli.py devices` against your real plant. """ from __future__ import annotations import base64 import logging from datetime import datetime import httpx from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.primitives.serialization import load_der_public_key, load_pem_public_key from app.config import settings from app.db.repositories import get_sync_state, set_sync_state from app.db.session import get_session from app.isolarcloud.parsing import POINT_EXPORT, POINT_GENERATION, POINT_IMPORT, POINT_LOAD, TS_FORMAT, merge_frames from app.isolarcloud.schemas import IntervalDataPoint _LOGGER = logging.getLogger(__name__) REGION_GATEWAYS = { "australia": "https://augateway.isolarcloud.com", "europe": "https://gateway.isolarcloud.eu", "china": "https://gateway.isolarcloud.com", "international": "https://gateway.isolarcloud.com.hk", } LOGIN_PATH = "/openapi/login" POWER_STATION_LIST_PATH = "/openapi/getPowerStationList" DEVICE_LIST_PATH = "/openapi/getDeviceList" MINUTE_HISTORY_PATH = "/openapi/getDevicePointMinuteDataList" DEVICE_TYPE_INVERTER = 1 DEVICE_TYPE_METER = 7 # result_code values meaning the token is invalid/expired -- worth one # re-login + retry. Per HA-Sungrow, E900 covers both an expired token and # "no permission", so a retry can be futile, but it only costs one call. TOKEN_ERROR_CODES = {"E900"} class ISolarCloudAuthError(RuntimeError): pass def _load_public_key(key_str: str): """Accepts either a full PEM key or the raw base64 (standard or URL-safe -- the developer portal's "Overview" page gives it without BEGIN/END headers and with -/_ in place of +//) DER SubjectPublicKeyInfo that the iSolarCloud portal actually hands out.""" key_str = key_str.strip() if "BEGIN" in key_str: return load_pem_public_key(key_str.encode()) # Raw base64: normalize URL-safe alphabet and padding, then DER-decode. normalized = key_str.replace("-", "+").replace("_", "/") normalized += "=" * (-len(normalized) % 4) der_bytes = base64.b64decode(normalized) return load_der_public_key(der_bytes) def _encrypt_password(password: str, public_key_str: str) -> str: """Best-effort RSA/PKCS1v1.5 encryption, base64-encoded. See the module docstring -- this is the common convention for this class of API, not an independently confirmed one for Sungrow specifically.""" public_key = _load_public_key(public_key_str) ciphertext = public_key.encrypt(password.encode(), padding.PKCS1v15()) return base64.b64encode(ciphertext).decode() class ISolarCloudClient: def __init__(self) -> None: self.host = REGION_GATEWAYS[settings.isolarcloud_region] self.appkey = settings.isolarcloud_app_key self.secret_key = settings.isolarcloud_app_secret self.username = settings.isolarcloud_username self.password = settings.isolarcloud_password self.rsa_public_key = settings.isolarcloud_rsa_public_key self._client = httpx.Client(timeout=30) self._token: str | None = None def _headers(self) -> dict[str, str]: headers = { "Content-Type": "application/json", "x-access-key": self.secret_key, "sys_code": "901", } if self._token: headers["token"] = self._token return headers def _post(self, path: str, payload: dict) -> dict: body = {"appkey": self.appkey, "lang": "_en_US", **payload} if self._token and "user_password" not in payload: body["token"] = self._token resp = self._client.post(f"{self.host}{path}", json=body, headers=self._headers()) resp.raise_for_status() return resp.json() def login(self) -> None: if not self.username or not self.password: raise ISolarCloudAuthError("iSolarCloud username/password not configured -- set them in .env.") password = self.password if self.rsa_public_key: password = _encrypt_password(password, self.rsa_public_key) data = self._post(LOGIN_PATH, {"user_account": self.username, "user_password": password, "login_type": "1"}) result_code = str(data.get("result_code")) result_data = data.get("result_data") or {} if result_code != "1": raise ISolarCloudAuthError(f"Login failed (result_code={result_code}): {data.get('result_msg')}") if str(result_data.get("login_state")) not in ("1", "None", ""): raise ISolarCloudAuthError( f"Login rejected (login_state={result_data.get('login_state')}): " f"{result_data.get('msg') or data.get('result_msg')}" ) token = result_data.get("token") if not token: raise ISolarCloudAuthError("Login succeeded but no token was returned") self._token = token with get_session() as session: set_sync_state(session, "isolarcloud_token", token) _LOGGER.debug("Logged in to iSolarCloud, token acquired") def _load_cached_token(self) -> None: if self._token is not None: return with get_session() as session: self._token = get_sync_state(session, "isolarcloud_token") def _authenticated_post(self, path: str, payload: dict) -> dict: self._load_cached_token() if self._token is None: self.login() data = self._post(path, payload) result_code = str(data.get("result_code")) result_msg = str(data.get("result_msg") or "") if result_code != "1" and (result_code in TOKEN_ERROR_CODES or "token" in result_msg.lower()): _LOGGER.debug("Token rejected (result_code=%s, msg=%s); re-authenticating", result_code, result_msg) self._token = None self.login() data = self._post(path, payload) result_code = str(data.get("result_code")) if result_code != "1": raise ISolarCloudAuthError(f"iSolarCloud error on {path} (result_code={result_code}): {data.get('result_msg')}") return data.get("result_data") or {} def get_plant_list(self) -> list[dict]: result = self._authenticated_post(POWER_STATION_LIST_PATH, {"curPage": 1, "size": 100}) return result.get("pageList") or result.get("page_list") or [] def get_device_list(self, ps_id: str) -> list[dict]: result = self._authenticated_post(DEVICE_LIST_PATH, {"ps_id": ps_id, "curPage": 1, "size": 100}) return result.get("pageList") or result.get("page_list") or [] def get_minute_history(self, ps_key: str, point_ids: list[str], start: str, end: str, minute_interval: int = 5) -> list[dict]: result = self._authenticated_post( MINUTE_HISTORY_PATH, { "ps_key_list": [ps_key], "points": ",".join(f"p{p}" for p in point_ids), "start_time_stamp": start, "end_time_stamp": end, "minute_interval": str(minute_interval), }, ) frames = result.get(ps_key) or [] return [f for f in frames if isinstance(f, dict)] def fetch_interval_data(self, since: datetime, until: datetime) -> list[IntervalDataPoint]: from app.db.repositories import get_household_config with get_session() as session: ps_id = get_household_config(session).isolarcloud_ps_id or settings.isolarcloud_plant_id if not ps_id: plants = self.get_plant_list() if not plants: raise ISolarCloudAuthError("No plants visible to this account.") ps_id = plants[0]["ps_id"] devices = self.get_device_list(ps_id) meter = next((d for d in devices if int(d.get("device_type", -1)) == DEVICE_TYPE_METER), None) inverter = next((d for d in devices if int(d.get("device_type", -1)) == DEVICE_TYPE_INVERTER), None) if meter is None or inverter is None: raise ISolarCloudAuthError( "Could not find both a meter and an inverter device on this plant -- " "run `python scripts/verify_isolarcloud_cli.py devices` and adjust " "app/isolarcloud/client.py's device selection for your setup." ) # The endpoint takes timestamps in the plant's local time zone, not # UTC (consistent with the OAuth2 app family's equivalent endpoint, # confirmed against Sungrow's official docs -- see oauth_client.py). from app.isolarcloud.parsing import PLANT_TZ start_s = since.astimezone(PLANT_TZ).strftime(TS_FORMAT) end_s = until.astimezone(PLANT_TZ).strftime(TS_FORMAT) meter_frames = self.get_minute_history(meter["ps_key"], [POINT_LOAD, POINT_IMPORT, POINT_EXPORT], start_s, end_s) inverter_frames = self.get_minute_history(inverter["ps_key"], [POINT_GENERATION], start_s, end_s) return merge_frames(meter_frames, inverter_frames)