dlawler489/power-compare
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
power-compare/app/isolarcloud/oauth_client.py
dlawler489 1dbfe3007b
All checks were successful
Build and Push Docker Image / build-and-push (push) Successful in 24s
Details
Share one iSolarCloud OAuth token across both household instances 
Sungrow's OAuth server keeps only one active token session per (app,
account), not per plant -- with Kallangur and McDowall each managing their
own independent token against the same account, whichever one refreshed
last was silently invalidating the other's refresh token. Both instances
now share one token file (a common bind-mounted volume) instead of each
holding their own.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-13 08:04:30 +10:00

393 lines
19 KiB
Python
Raw Blame History

"""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 <access_token>`, `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 import shared_token_store
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
token_state = {
"access_token": result_data["access_token"],
"refresh_token": result_data["refresh_token"],
"expires_at": expires_at,
}
if shared_token_store.enabled():
with shared_token_store.locked() as fd:
shared_token_store.write(fd, token_state)
else:
with get_session() as session:
set_sync_state(session, "isolarcloud_oauth_access_token", token_state["access_token"])
set_sync_state(session, "isolarcloud_oauth_refresh_token", token_state["refresh_token"])
set_sync_state(session, "isolarcloud_oauth_expires_at", str(expires_at))
self._access_token = token_state["access_token"]
def _get_access_token(self) -> str:
if shared_token_store.enabled():
# Holds the lock across the whole check-and-maybe-refresh
# sequence (including the network round-trip), not just the
# file I/O either side of it -- otherwise two instances waking
# up around the same moment could both read the same
# about-to-expire refresh_token and race to redeem it, and
# only one of those calls will succeed.
with shared_token_store.locked() as fd:
state = shared_token_store.read(fd)
if not state or not state.get("access_token") or not state.get("refresh_token"):
raise ISolarCloudAuthError(
"No iSolarCloud OAuth tokens stored yet -- connect your account from the Settings page first."
)
if int(state.get("expires_at") or 0) < int(time.time()):
data = self._post(REFRESH_PATH, {"refresh_token": state["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.")
expires_at = int(time.time()) + int(result_data["expires_in"]) - 60
state = {
"access_token": result_data["access_token"],
"refresh_token": result_data["refresh_token"],
"expires_at": expires_at,
}
shared_token_store.write(fd, state)
self._access_token = state["access_token"]
return state["access_token"]
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,
)
]
Reference in a new issue View git blame Copy permalink