235 lines
8.2 KiB
Python
235 lines
8.2 KiB
Python
"""Client for the SolarEdge Optimizer Data App API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime
|
|
from typing import Any, Final
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
from aiohttp import ClientError, ClientResponseError, ClientSession
|
|
|
|
from .const import API_PATH, API_TIMEOUT_SECONDS
|
|
|
|
_VALID_STATUSES: Final = frozenset({"ok", "partial", "stale"})
|
|
|
|
|
|
class SolarEdgeOptimizerApiError(Exception):
|
|
"""Base error raised by the App API client."""
|
|
|
|
|
|
class SolarEdgeOptimizerConnectionError(SolarEdgeOptimizerApiError):
|
|
"""The App API could not be reached."""
|
|
|
|
|
|
class SolarEdgeOptimizerInvalidResponseError(SolarEdgeOptimizerApiError):
|
|
"""The App returned an invalid response."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class OptimizerData:
|
|
"""Optimizer readings returned by the App."""
|
|
|
|
serial: str
|
|
inverter_id: str | None
|
|
string_id: str | None
|
|
optimizer_id: str | None
|
|
daily_energy_wh: float | None
|
|
current_power_w: float | None
|
|
last_measurement: datetime | None
|
|
error: str | None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class OptimizerSnapshot:
|
|
"""A complete cached optimizer snapshot."""
|
|
|
|
status: str
|
|
site_id: str
|
|
date: date
|
|
time_zone: str
|
|
fetched_at: datetime
|
|
next_refresh_at: datetime
|
|
optimizer_count: int
|
|
successful_optimizer_count: int
|
|
failed_optimizer_count: int
|
|
total_daily_energy_wh: float
|
|
total_current_power_w: float
|
|
optimizers: tuple[OptimizerData, ...]
|
|
last_error: str | None
|
|
|
|
@property
|
|
def optimizers_by_serial(self) -> dict[str, OptimizerData]:
|
|
"""Return readings indexed by optimizer serial number."""
|
|
return {optimizer.serial: optimizer for optimizer in self.optimizers}
|
|
|
|
@property
|
|
def optimizers_by_id(self) -> dict[str, OptimizerData]:
|
|
"""Return readings indexed by logical optimizer ID."""
|
|
return {
|
|
optimizer.optimizer_id: optimizer
|
|
for optimizer in self.optimizers
|
|
if optimizer.optimizer_id is not None
|
|
}
|
|
|
|
|
|
class SolarEdgeOptimizerApiClient:
|
|
"""Read the local SolarEdge Optimizer Data App API."""
|
|
|
|
def __init__(self, base_url: str, session: ClientSession) -> None:
|
|
"""Initialize the client."""
|
|
self._base_url = normalize_base_url(base_url)
|
|
self._session = session
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
"""Return the normalized App base URL."""
|
|
return self._base_url
|
|
|
|
async def async_get_optimizers(self) -> OptimizerSnapshot:
|
|
"""Fetch and validate the latest cached snapshot."""
|
|
try:
|
|
async with asyncio.timeout(API_TIMEOUT_SECONDS):
|
|
async with self._session.get(
|
|
f"{self._base_url}{API_PATH}",
|
|
headers={"Accept": "application/json"},
|
|
) as response:
|
|
response.raise_for_status()
|
|
payload = await response.json(content_type=None)
|
|
except (TimeoutError, ClientError, ClientResponseError) as err:
|
|
raise SolarEdgeOptimizerConnectionError from err
|
|
except (ValueError, TypeError) as err:
|
|
raise SolarEdgeOptimizerInvalidResponseError from err
|
|
|
|
return parse_snapshot(payload)
|
|
|
|
|
|
def normalize_base_url(value: str) -> str:
|
|
"""Validate and normalize an App base URL."""
|
|
value = value.strip().rstrip("/")
|
|
if value.endswith(API_PATH):
|
|
value = value[: -len(API_PATH)]
|
|
|
|
parts = urlsplit(value)
|
|
if parts.scheme not in {"http", "https"} or not parts.hostname:
|
|
raise ValueError("URL must use http or https and include a host")
|
|
if parts.username or parts.password or parts.query or parts.fragment:
|
|
raise ValueError("URL must not include credentials, a query, or a fragment")
|
|
path = parts.path.rstrip("/")
|
|
return urlunsplit((parts.scheme, parts.netloc, path, "", ""))
|
|
|
|
|
|
def parse_snapshot(payload: Any) -> OptimizerSnapshot:
|
|
"""Parse and validate a snapshot payload."""
|
|
if not isinstance(payload, dict):
|
|
raise SolarEdgeOptimizerInvalidResponseError("Expected a JSON object")
|
|
|
|
try:
|
|
status = _required_string(payload, "status")
|
|
if status not in _VALID_STATUSES:
|
|
raise ValueError("Unsupported status")
|
|
|
|
raw_optimizers = payload["optimizers"]
|
|
if not isinstance(raw_optimizers, list):
|
|
raise TypeError("optimizers must be a list")
|
|
optimizers = tuple(_parse_optimizer(item) for item in raw_optimizers)
|
|
|
|
optimizer_count = _required_int(payload, "optimizerCount")
|
|
if optimizer_count != len(optimizers):
|
|
raise ValueError("optimizerCount does not match optimizers")
|
|
if len({optimizer.serial for optimizer in optimizers}) != len(optimizers):
|
|
raise ValueError("Optimizer serial numbers must be unique")
|
|
|
|
return OptimizerSnapshot(
|
|
status=status,
|
|
site_id=_required_string(payload, "siteId"),
|
|
date=date.fromisoformat(_required_string(payload, "date")),
|
|
time_zone=_required_string(payload, "timeZone"),
|
|
fetched_at=_parse_datetime(payload, "fetchedAt"),
|
|
next_refresh_at=_parse_datetime(payload, "nextRefreshAt"),
|
|
optimizer_count=optimizer_count,
|
|
successful_optimizer_count=_required_int(
|
|
payload, "successfulOptimizerCount"
|
|
),
|
|
failed_optimizer_count=_required_int(payload, "failedOptimizerCount"),
|
|
total_daily_energy_wh=_required_number(payload, "totalDailyEnergyWh"),
|
|
total_current_power_w=_required_number(payload, "totalCurrentPowerW"),
|
|
optimizers=optimizers,
|
|
last_error=_optional_string(payload, "lastError"),
|
|
)
|
|
except (KeyError, TypeError, ValueError) as err:
|
|
raise SolarEdgeOptimizerInvalidResponseError from err
|
|
|
|
|
|
def _parse_optimizer(payload: Any) -> OptimizerData:
|
|
if not isinstance(payload, dict):
|
|
raise TypeError("Optimizer must be a JSON object")
|
|
return OptimizerData(
|
|
serial=_required_string(payload, "serial"),
|
|
inverter_id=_nullable_string(payload, "inverterId"),
|
|
string_id=_nullable_string(payload, "stringId"),
|
|
optimizer_id=_nullable_string(payload, "optimizerId"),
|
|
daily_energy_wh=_nullable_number(payload, "dailyEnergyWh"),
|
|
current_power_w=_nullable_number(payload, "currentPowerW"),
|
|
last_measurement=_nullable_datetime(payload, "lastMeasurement"),
|
|
error=_optional_string(payload, "error"),
|
|
)
|
|
|
|
|
|
def _required_string(payload: dict[str, Any], key: str) -> str:
|
|
value = payload[key]
|
|
if not isinstance(value, str) or not value:
|
|
raise TypeError(f"{key} must be a non-empty string")
|
|
return value
|
|
|
|
|
|
def _optional_string(payload: dict[str, Any], key: str) -> str | None:
|
|
value = payload.get(key)
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, str):
|
|
raise TypeError(f"{key} must be a string")
|
|
return value
|
|
|
|
|
|
def _nullable_string(payload: dict[str, Any], key: str) -> str | None:
|
|
value = payload[key]
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, str):
|
|
raise TypeError(f"{key} must be a string or null")
|
|
return value
|
|
|
|
|
|
def _required_int(payload: dict[str, Any], key: str) -> int:
|
|
value = payload[key]
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
raise TypeError(f"{key} must be a non-negative integer")
|
|
return value
|
|
|
|
|
|
def _required_number(payload: dict[str, Any], key: str) -> float:
|
|
value = payload[key]
|
|
if isinstance(value, bool) or not isinstance(value, int | float):
|
|
raise TypeError(f"{key} must be a number")
|
|
return float(value)
|
|
|
|
|
|
def _nullable_number(payload: dict[str, Any], key: str) -> float | None:
|
|
value = payload[key]
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool) or not isinstance(value, int | float):
|
|
raise TypeError(f"{key} must be a number or null")
|
|
return float(value)
|
|
|
|
|
|
def _parse_datetime(payload: dict[str, Any], key: str) -> datetime:
|
|
return datetime.fromisoformat(_required_string(payload, key))
|
|
|
|
|
|
def _nullable_datetime(payload: dict[str, Any], key: str) -> datetime | None:
|
|
value = _nullable_string(payload, key)
|
|
return None if value is None else datetime.fromisoformat(value)
|