"""Real iSolarCloud client -- OAuth2.0 app type. This is a *separate* developer-portal app registration from the "V1 account login" one in client.py -- different credentials (an Application ID + redirect URI, no username/password needed at request time) and a different, browser-based consent flow. See SETUP.md. Endpoint paths, auth flow and field names below are all confirmed directly against Sungrow's own public documentation at https://audeveloper-api.isolarcloud.com/#/document (the "OAuth2.0" tab), not third-party reverse-engineering -- specifically: - Getting Started: header/body conventions (`x-access-key`, `Authorization: Bearer `, `appkey` in every body). - "Obtain token via authorization code" (`/openapi/apiManage/token`) -- its response includes `auth_ps_list`, the plant IDs the user actually granted this app access to during the consent screen. This is the explicit authorization step that iSolarCloud's V1 login apps apparently don't reliably get without it (see the E900 "Unauthorized access" investigation this project went through before switching to OAuth2). - "Refresh Token" (`/openapi/apiManage/refreshToken`). - "Query Plant List" (`/openapi/platform/queryPowerStationList`), "Query Device List of the Plant" (`/openapi/platform/getDeviceListByPsId`), "Query plant real-time measuring point data" (`/openapi/platform/getPowerStationRealTimeData`), and "Query Historical Minute-Level Measuring Point Data of Plants" (`/openapi/platform/getPowerStationPointMinuteDataList`) -- all plant-level (keyed by `ps_id`, not a per-device `ps_key` like the V1 app's endpoints), which is simpler: no device discovery needed. Two constraints from the historical-data endpoint's own docs that shape `fetch_interval_data` below: - It accepts at most a 3-hour window per call, and multiple calls are needed to cover a longer backfill range. - "Historical data endpoints do not return data for the current day" -- today's figures have to come from the real-time endpoint instead, which only returns a live snapshot (not a time series). This client derives today's interval contribution by diffing each poll's cumulative import/export counters against the previous poll's (persisted in `sync_state`), the same way multiple historical frames are diffed against each other. """ from __future__ import annotations import logging import time from datetime import date, datetime, timedelta from urllib.parse import quote_plus from zoneinfo import ZoneInfo import httpx 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__) BRISBANE_TZ = ZoneInfo("Australia/Brisbane") REGION_GATEWAYS = { "australia": "https://augateway.isolarcloud.com", "europe": "https://gateway.isolarcloud.eu", "china": "https://gateway.isolarcloud.com", "international": "https://gateway.isolarcloud.com.hk", } # (auth-page host, cloudId) per region, used to build the browser consent URL. REGION_AUTH_SERVERS = { "australia": ("auweb3.isolarcloud.com", 7), "europe": ("web3.isolarcloud.eu", 3), "china": ("web3.isolarcloud.com", 1), "international": ("web3.isolarcloud.com.hk", 2), } TOKEN_PATH = "/openapi/apiManage/token" REFRESH_PATH = "/openapi/apiManage/refreshToken" PLANT_LIST_PATH = "/openapi/platform/queryPowerStationList" DEVICE_LIST_PATH = "/openapi/platform/getDeviceListByPsId" REALTIME_DATA_PATH = "/openapi/platform/getPowerStationRealTimeData" MINUTE_HISTORY_PATH = "/openapi/platform/getPowerStationPointMinuteDataList" DEVICE_REALTIME_DATA_PATH = "/openapi/platform/getDeviceRealTimeData" MAX_HISTORY_WINDOW = timedelta(hours=3) # the endpoint's own documented limit # Battery/hybrid-inverter telemetry (SOC, charge/discharge power) lives at # the *device* level, not in the plant-aggregate points used elsewhere in # this client -- confirmed against Sungrow's own docs ("Common Energy # Storage Inverter Measuring Points") and live-tested against a real # battery plant (querying these via the plant-level realtime endpoint # just returns null). Device type 14 covers both pure hybrid inverters and # combined inverter+battery "Energy Storage System" units. ESS_DEVICE_TYPE = "14" POINT_BATTERY_SOC = "13141" # fraction 0-1, not a percentage POINT_BATTERY_CHARGE_W = "13126" POINT_BATTERY_DISCHARGE_W = "13150" class ISolarCloudAuthError(RuntimeError): pass class ISolarCloudOAuthClient: def __init__(self) -> None: self.host = REGION_GATEWAYS[settings.isolarcloud_region] self.auth_server, self.cloud_id = REGION_AUTH_SERVERS[settings.isolarcloud_region] self.appkey = settings.isolarcloud_app_key self.secret_key = settings.isolarcloud_app_secret self.app_id = settings.isolarcloud_oauth_app_id self.redirect_uri = settings.isolarcloud_oauth_redirect_uri self._client = httpx.Client(timeout=30) self._access_token: str | None = None def authorize_url(self) -> str: """URL the account owner opens once, in their own browser, to approve this app's access to their plant(s).""" return ( f"https://{self.auth_server}/#/authorized-app?cloudId={self.cloud_id}" f"&applicationId={self.app_id}&redirectUrl={quote_plus(self.redirect_uri)}" ) def _headers(self, access_token: str | None = None) -> dict[str, str]: headers = {"Content-Type": "application/json", "x-access-key": self.secret_key} if access_token: headers["Authorization"] = f"Bearer {access_token}" return headers def _post(self, path: str, payload: dict, access_token: str | None = None) -> dict: body = {"appkey": self.appkey, "lang": "_en_US", **payload} resp = self._client.post(f"{self.host}{path}", json=body, headers=self._headers(access_token)) resp.raise_for_status() return resp.json() def exchange_code(self, code: str) -> list[str]: """Completes the one-time consent flow: trades the `code` param from the redirect for an access/refresh token pair, and returns the list of plant IDs the user actually authorized (`auth_ps_list`) so Settings can show them for plant selection.""" data = self._post(TOKEN_PATH, {"code": code, "grant_type": "authorization_code", "redirect_uri": self.redirect_uri}) result_data = data.get("result_data") or {} if not result_data.get("access_token"): raise ISolarCloudAuthError(f"Authorization failed: {data}") self._store_tokens(result_data) return result_data.get("auth_ps_list") or [] def _store_tokens(self, result_data: dict) -> None: expires_at = int(time.time()) + int(result_data["expires_in"]) - 60 with get_session() as session: set_sync_state(session, "isolarcloud_oauth_access_token", result_data["access_token"]) set_sync_state(session, "isolarcloud_oauth_refresh_token", result_data["refresh_token"]) set_sync_state(session, "isolarcloud_oauth_expires_at", str(expires_at)) self._access_token = result_data["access_token"] def _get_access_token(self) -> str: with get_session() as session: access_token = get_sync_state(session, "isolarcloud_oauth_access_token") refresh_token = get_sync_state(session, "isolarcloud_oauth_refresh_token") expires_at = get_sync_state(session, "isolarcloud_oauth_expires_at") if not access_token or not refresh_token: raise ISolarCloudAuthError( "No iSolarCloud OAuth tokens stored yet -- connect your account from the Settings page first." ) if int(expires_at or 0) < int(time.time()): data = self._post(REFRESH_PATH, {"refresh_token": refresh_token}) result_data = data.get("result_data") or {} if not isinstance(result_data, dict) or not result_data.get("access_token"): raise ISolarCloudAuthError(f"Token refresh failed: {data} -- reconnect from Settings.") self._store_tokens(result_data) access_token = result_data["access_token"] return access_token def _authenticated_post(self, path: str, payload: dict) -> dict: access_token = self._get_access_token() data = self._post(path, payload, access_token=access_token) # Two distinct failure shapes are documented: the usual # {result_code, result_msg, ...} and an OAuth-specific # {"error": "invalid_token", "error_description": ...}. if data.get("error"): raise ISolarCloudAuthError(f"iSolarCloud OAuth error on {path}: {data.get('error_description') or data['error']}") 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(PLANT_LIST_PATH, {"page": 1, "size": 100}) return result.get("pageList") or [] def get_device_list(self, ps_id: str) -> list[dict]: result = self._authenticated_post(DEVICE_LIST_PATH, {"ps_id": ps_id, "page": 1, "size": 100}) return result.get("pageList") or [] def get_battery_status(self, ps_id: str) -> dict[str, float] | None: """Live battery state for plants with a hybrid inverter/ESS device, or None for solar-only plants that have no such device (e.g. a plain string-inverter system).""" ess_device = next( (d for d in self.get_device_list(ps_id) if str(d.get("device_type")) == ESS_DEVICE_TYPE), None ) if ess_device is None or not ess_device.get("ps_key"): return None result = self._authenticated_post( DEVICE_REALTIME_DATA_PATH, { "ps_key_list": [ess_device["ps_key"]], "point_id_list": [POINT_BATTERY_SOC, POINT_BATTERY_CHARGE_W, POINT_BATTERY_DISCHARGE_W], "device_type": ESS_DEVICE_TYPE, }, ) points_list = result.get("device_point_list") or [] if not points_list: return None point = points_list[0].get("device_point") or {} soc = point.get(f"p{POINT_BATTERY_SOC}") if soc is None: return None charge_w = point.get(f"p{POINT_BATTERY_CHARGE_W}") discharge_w = point.get(f"p{POINT_BATTERY_DISCHARGE_W}") return { "soc_percent": round(float(soc) * 100, 1), "charge_w": float(charge_w) if charge_w is not None else 0.0, "discharge_w": float(discharge_w) if discharge_w is not None else 0.0, } def get_realtime_points(self, ps_id: str, point_ids: list[str]) -> dict[str, float]: result = self._authenticated_post( REALTIME_DATA_PATH, {"ps_id_list": [ps_id], "point_id_list": point_ids, "is_get_point_dict": "0"} ) device_points = result.get("device_point_list") or [] if not device_points: return {} values: dict[str, float] = {} for key, value in device_points[0].items(): if key.startswith("p") and key[1:].isdigit(): try: values[key[1:]] = float(value) except (TypeError, ValueError): continue return values def get_minute_history(self, ps_id: str, point_ids: list[str], start: datetime, end: datetime) -> list[dict]: """Chunks the request into <=3-hour windows per the endpoint's documented limit, and merges the resulting frames.""" frames: list[dict] = [] chunk_start = start while chunk_start < end: chunk_end = min(chunk_start + MAX_HISTORY_WINDOW, end) result = self._authenticated_post( MINUTE_HISTORY_PATH, { "ps_id_list": [ps_id], "points": ",".join(f"p{p}" for p in point_ids), "start_time_stamp": chunk_start.strftime(TS_FORMAT), "end_time_stamp": chunk_end.strftime(TS_FORMAT), "minute_interval": "5", }, ) frames.extend(f for f in (result.get(str(ps_id)) or []) if isinstance(f, dict)) chunk_start = chunk_end return frames def _resolve_ps_id(self) -> str: 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 ps_id: return ps_id plants = self.get_plant_list() if not plants: raise ISolarCloudAuthError("No plants visible to this account.") return str(plants[0]["ps_id"]) def fetch_interval_data(self, since: datetime, until: datetime) -> list[IntervalDataPoint]: ps_id = self._resolve_ps_id() points = [POINT_GENERATION, POINT_LOAD, POINT_IMPORT, POINT_EXPORT] today_start = datetime.now(BRISBANE_TZ).replace(hour=0, minute=0, second=0, microsecond=0) results: list[IntervalDataPoint] = [] historical_end = min(until, today_start) if since < historical_end: # The endpoint takes timestamps in the plant's local time zone, # not UTC -- convert before formatting (see module docstring). frames = self.get_minute_history(ps_id, points, since.astimezone(BRISBANE_TZ), historical_end.astimezone(BRISBANE_TZ)) results.extend(merge_frames(frames)) if until > today_start: results.extend(self._poll_today(ps_id, points, until)) return results def _poll_today(self, ps_id: str, point_ids: list[str], now: datetime) -> list[IntervalDataPoint]: """Derives one interval point for "right now" by diffing this poll's cumulative import/export counters against the last poll's (historical endpoints don't cover today at all).""" values = self.get_realtime_points(ps_id, point_ids) if not values: return [] today_key = date.today().isoformat() with get_session() as session: last_day = get_sync_state(session, "isolarcloud_oauth_today") last_import_wh = get_sync_state(session, "isolarcloud_oauth_last_import_wh") last_export_wh = get_sync_state(session, "isolarcloud_oauth_last_export_wh") last_poll_ts = get_sync_state(session, "isolarcloud_oauth_last_poll_ts") import_wh = values.get(POINT_IMPORT) export_wh = values.get(POINT_EXPORT) is_new_day = last_day != today_key import_kwh = 0.0 export_kwh = 0.0 if not is_new_day and last_import_wh is not None and import_wh is not None: import_kwh = max(0.0, (import_wh - float(last_import_wh)) / 1000) if not is_new_day and last_export_wh is not None and export_wh is not None: export_kwh = max(0.0, (export_wh - float(last_export_wh)) / 1000) elapsed_minutes = 5.0 if not is_new_day and last_poll_ts: elapsed_minutes = max(1.0, (now - datetime.fromisoformat(last_poll_ts)).total_seconds() / 60) with get_session() as session: set_sync_state(session, "isolarcloud_oauth_today", today_key) if import_wh is not None: set_sync_state(session, "isolarcloud_oauth_last_import_wh", str(import_wh)) if export_wh is not None: set_sync_state(session, "isolarcloud_oauth_last_export_wh", str(export_wh)) set_sync_state(session, "isolarcloud_oauth_last_poll_ts", now.isoformat()) battery = None try: battery = self.get_battery_status(ps_id) except Exception: # A battery-status hiccup shouldn't cost the core solar # reading for this poll -- just leave it unset this round. _LOGGER.exception("Battery status fetch failed for ps_id=%s", ps_id) step_hours = elapsed_minutes / 60 return [ IntervalDataPoint( ts=now, generation_kwh=values.get(POINT_GENERATION, 0.0) * step_hours / 1000, load_kwh=values.get(POINT_LOAD, 0.0) * step_hours / 1000, import_kwh=import_kwh, export_kwh=export_kwh, resolution_minutes=round(elapsed_minutes), battery_soc_percent=battery["soc_percent"] if battery else None, battery_charge_w=battery["charge_w"] if battery else None, battery_discharge_w=battery["discharge_w"] if battery else None, ) ]