84c4922bbc
FastAPI app that pulls interval usage from Sungrow iSolarCloud (OAuth2), caches AER Consumer Data Right electricity plan data, and compares actual metered usage against every available plan to estimate real annual cost (TOU, tiered feed-in, and flat tariffs). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
236 lines
11 KiB
Python
236 lines
11 KiB
Python
"""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)
|