Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 41 additions & 1 deletion src/solaredge_web/solaredge.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import time
from datetime import datetime
from enum import IntEnum
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, cast

import aiohttp

Expand Down Expand Up @@ -60,6 +60,7 @@ def __init__(
self.timeout = aiohttp.ClientTimeout(total=timeout)
self._equipment: dict[int, dict[str, Any]] = {}
self._last_login_time = 0.0
self._smart_home_visited = False

async def async_login(self) -> None:
"""Login to the SolarEdge Web."""
Expand All @@ -68,6 +69,7 @@ async def async_login(self) -> None:
_LOGGER.debug("Skipping login. Using existing valid SSO cookie")
return
self._equipment = {}
self._smart_home_visited = False
url = "https://monitoring.solaredge.com/solaredge-apigw/api/login"
try:
resp = await self.session.post(
Expand Down Expand Up @@ -119,6 +121,44 @@ def extract_nested_data(node: dict[Any, Any], data_dict: dict[int, dict[str, Any
_LOGGER.debug("Found %s equipment for site: %s", len(self._equipment), self.site_id)
return self._equipment

async def async_get_home_automation_devices(self) -> dict[str, Any]:
"""Get home automation devices from the SolarEdge Web API.

Returns device information from the home automation API endpoint.

Note: This endpoint requires visiting the Smart Home page first to
establish the proper session state before the API call will succeed.
"""
_LOGGER.debug("Fetching home automation devices for site: %s", self.site_id)
await self.async_login()

# Visit the Smart Home page first to establish session state
# This is required for the API to return device data
if not self._smart_home_visited:
smart_home_url = f"https://monitoring.solaredge.com/solaredge-web/p/site/{self.site_id}/#/smart-home"
_LOGGER.debug("Visiting Smart Home page to establish session: %s", smart_home_url)
try:
resp = await self.session.get(smart_home_url, timeout=self.timeout)
_LOGGER.debug("Smart Home page returned %s", resp.status)
resp.raise_for_status()
self._smart_home_visited = True
except aiohttp.ClientError:
_LOGGER.exception("Error visiting Smart Home page %s", smart_home_url)
raise

url = f"https://monitoring.solaredge.com/services/api/homeautomation/v1.0/sites/{self.site_id}/devices"

try:
resp = await self.session.get(url, timeout=self.timeout)
_LOGGER.debug("Got %s from %s", resp.status, url)
resp.raise_for_status()
except aiohttp.ClientError:
_LOGGER.exception("Error fetching home automation devices from %s", url)
raise
resp_json = await resp.json()
_LOGGER.debug("Found home automation devices for site: %s", self.site_id)
return cast("dict[str, Any]", resp_json)

async def async_get_energy_data(self, time_unit: TimeUnit = TimeUnit.WEEK) -> list[EnergyData]:
"""Get energy data from the SolarEdge Web API.

Expand Down