solar-of-things-ha/api.py

"""API client for Solar of Things (solar.siseli.com).

Authentication strategy
─────────────────────────────────────────────────────────────────────────────
The Siseli portal uses a dual-token system discovered via JS bundle analysis:

  • POST /login/account          → accessToken + refreshToken + expiry timestamps
  • POST /login/refresh/access/token → new accessToken (+ new refreshToken) using
                                        the current refreshToken

accessToken  = the "IOT-Token" HTTP header value; short-lived (hours/days)
refreshToken = long-lived; used only to mint a new accessToken

This class supports three auth modes, tried in priority order:

  1. Email + 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, matching 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(
      email="user@example.com",
      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 logging
import threading
from datetime import datetime, timezone, timedelta
from typing import Any, Callable

import requests

try:
    from zoneinfo import ZoneInfo
except Exception:  # pragma: no cover
    ZoneInfo = None  # type: ignore[assignment]

from .const import (
    API_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,
    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."""


# ──────────────────────────────────────────────────────────────────────────────
# 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


# ──────────────────────────────────────────────────────────────────────────────
# Main API client
# ──────────────────────────────────────────────────────────────────────────────

class SolarOfThingsAPI:
    """Solar of Things API wrapper with automatic token refresh.

    Parameters
    ----------
    email:              Siseli portal login e-mail (preferred auth method).
    password:           Siseli portal password.
    iot_token:          Legacy/manual IOT-Token (used when email/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,
        *,
        email: 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._email = email
        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 email 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 (email + 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/",
                "User-Agent": (
                    "HomeAssistant-SolarOfThings/2.2.0 "
                    "(+https://github.com/conexocasa/solar-of-things-ha)"
                ),
            }
        )

    # ─── Public auth helpers ───────────────────────────────────────────────────

    def login(self) -> None:
        """Authenticate with email + password and store the resulting tokens.

        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 email + password auth mode.")

        _LOGGER.debug("SolarOfThings: logging in as %s", self._email)

        payload = {
            "email": self._email,
            "password": self._password,
            "loginType": "email",
        }

        # Auth endpoints must NOT carry the IOT-Token header yet
        resp = requests.post(
            f"{API_BASE_URL}{API_LOGIN}",
            json=payload,
            headers={
                "Accept": "application/json",
                "Content-Type": "application/json; charset=utf-8",
                "Origin": "https://solar.siseli.com",
                "Referer": "https://solar.siseli.com/",
            },
            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_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._email 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, automatically refreshing the token on 401.

        On second 401 (after refresh) raises TokenExpiredError.
        """
        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()

    # ─── 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)."""
        end_time = self._now()
        start_time = end_time - timedelta(hours=1)

        keys = [
            "pvInputPower",
            "acOutputActivePower",
            "batteryDischargeCurrent",
            "batteryChargingCurrent",
            "batteryVoltage",
            "feedInPower",
            "batterySOC",
        ]

        data = self._post(
            API_TIME_SERIES,
            {
                "deviceId": device_id,
                "count": 2000,
                "page": 1,
                "fromTime": self._format_time(start_time),
                "toTime": self._format_time(end_time),
                "orderByTimeAsc": True,
                "keys": keys,
            },
        )

        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]

        # Unit normalisation: acOutputActivePower is kW in API → W
        if "acOutputActivePower" in latest_values:
            try:
                latest_values["acOutputActivePower"] = (
                    float(latest_values["acOutputActivePower"]) * 1000.0
                )
            except Exception:
                pass

        # Derived values
        voltage = float(latest_values.get("batteryVoltage") or 0)
        discharge = float(latest_values.get("batteryDischargeCurrent") or 0)
        charge = float(latest_values.get("batteryChargingCurrent") or 0)
        latest_values["batteryPower"] = (discharge - charge) * voltage

        pv_power = float(latest_values.get("pvInputPower") or 0)
        ac_output = float(latest_values.get("acOutputActivePower") or 0)
        feed_in = float(latest_values.get("feedInPower") or 0)
        battery_power = float(latest_values.get("batteryPower") or 0)

        latest_values["gridPower"] = max(0.0, ac_output - pv_power + battery_power + feed_in)
        latest_values["loadPower"] = ac_output

        return latest_values

    # ─── 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 [])
        out: dict[str, Any] = {}

        for prop in props:
            prop_key = ((prop.get("property") or {}).get("key"))
            if not prop_key:
                continue
            for tp in prop.get("timePoints") or []:
                if tp.get("time") == month_key and tp.get("isRealValue"):
                    out[prop_key] = tp.get("value")

        if "pvGeneratedEnergy" in out:
            return {"monthly_pv_generated": float(out.get("pvGeneratedEnergy") or 0)}

        return {}

    # ─── Settings / control (per device) ──────────────────────────────────────

    def fetch_settings(self, device_id: str) -> dict[str, Any]:
        """Fetch current device settings."""
        try:
            data = self._post(API_SETTINGS_GET, {"deviceId": device_id})
            settings: dict[str, Any] = {}
            if "data" in data:
                sd = data["data"]
                settings["batteryChargeLimit"] = sd.get("batteryChargeLimit", 100)
                settings["batteryDischargeLimit"] = sd.get("batteryDischargeLimit", 10)
                settings["gridChargeLimit"] = sd.get("gridChargeLimit", 0)
                settings["operatingMode"] = sd.get("operatingMode", "Self-Use")
                settings["batteryPriority"] = sd.get("batteryPriority", "Solar First")
                settings["gridChargingEnabled"] = sd.get("gridChargingEnabled", False)
                settings["gridFeedInEnabled"] = sd.get("gridFeedInEnabled", True)
                settings["backupModeEnabled"] = sd.get("backupModeEnabled", False)
            return settings
        except requests.exceptions.RequestException as err:
            _LOGGER.error("Error fetching settings for device %s: %s", device_id, err)
            return {}

    def _update_setting(self, device_id: str, setting_key: str, value: Any) -> bool:
        try:
            data = self._post(
                API_SETTINGS_SET,
                {"deviceId": device_id, "settings": {setting_key: value}},
            )
            if data.get("success") is True or data.get("code") in (0, None):
                _LOGGER.info("Updated %s=%s for device %s", setting_key, value, device_id)
                return True
            _LOGGER.error("Failed to update %s for device %s: %s", setting_key, device_id, data)
            return False
        except requests.exceptions.RequestException as err:
            _LOGGER.error("Error updating %s for device %s: %s", setting_key, device_id, err)
            return False

    def set_battery_charge_limit(self, device_id: str, limit: int) -> bool:
        return self._update_setting(device_id, "batteryChargeLimit", limit)

    def set_battery_discharge_limit(self, device_id: str, limit: int) -> bool:
        return self._update_setting(device_id, "batteryDischargeLimit", limit)

    def set_grid_charge_limit(self, device_id: str, limit: int) -> bool:
        return self._update_setting(device_id, "gridChargeLimit", limit)

    def set_operating_mode(self, device_id: str, mode: str) -> bool:
        return self._update_setting(device_id, "operatingMode", mode)

    def set_battery_priority(self, device_id: str, priority: str) -> bool:
        return self._update_setting(device_id, "batteryPriority", priority)

    def set_grid_charging(self, device_id: str, enabled: bool) -> bool:
        return self._update_setting(device_id, "gridChargingEnabled", enabled)

    def set_grid_feed_in(self, device_id: str, enabled: bool) -> bool:
        return self._update_setting(device_id, "gridFeedInEnabled", enabled)

    def set_backup_mode(self, device_id: str, enabled: bool) -> bool:
        return self._update_setting(device_id, "backupModeEnabled", enabled)

    # ─── Validation ───────────────────────────────────────────────────────────

    def test_connection(self, station_id: str) -> bool:
        """Validate credentials + station by listing devices and querying telemetry."""
        try:
            devices = self.list_devices(station_id)
            if not devices:
                return False
            device_id = str((devices[0].get("id") or ""))
            if not device_id:
                return False
            self.fetch_latest_data(device_id)
            return True
        except TokenExpiredError:
            raise
        except Exception as err:
            _LOGGER.error("Connection test failed: %s", err)
            return False