home-assistant · tronikos · Jun 2, 2025 · Jun 3, 2025 · Jun 3, 2025 · Jun 16, 2025
diff --git a/.strict-typing b/.strict-typing
@@ -469,6 +469,7 @@ homeassistant.components.smhi.*
 homeassistant.components.smlight.*
    homeassistant.components.smtp.*
 homeassistant.components.snooz.*
+homeassistant.components.solaredge_modules.*
 homeassistant.components.solarlog.*
 homeassistant.components.sonarr.*
 homeassistant.components.speedtestdotnet.*

diff --git a/CODEOWNERS b/CODEOWNERS
diff --git a/homeassistant/brands/solaredge.json b/homeassistant/brands/solaredge.json
@@ -1,5 +1,5 @@
 {
   "domain": "solaredge",
   "name": "SolarEdge",
-  "integrations": ["solaredge", "solaredge_local"]
+  "integrations": ["solaredge", "solaredge_local", "solaredge_modules"]
 }
diff --git a/homeassistant/components/solaredge_modules/__init__.py b/homeassistant/components/solaredge_modules/__init__.py
@@ -0,0 +1,31 @@
+"""The SolarEdge Modules integration."""
+
+from __future__ import annotations
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import SolarEdgeModulesConfigEntry, SolarEdgeModulesCoordinator
+
+_PLATFORMS: list[Platform] = []
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: SolarEdgeModulesConfigEntry
+) -> bool:
+    """Set up SolarEdge Modules from a config entry."""
+
+    coordinator = SolarEdgeModulesCoordinator(hass, entry)
+    await coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: SolarEdgeModulesConfigEntry
+) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)
diff --git a/homeassistant/components/solaredge_modules/config_flow.py b/homeassistant/components/solaredge_modules/config_flow.py
@@ -0,0 +1,104 @@
+"""Config flow for the SolarEdge Modules integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import aiohttp
+from solaredge_web import SolarEdgeWeb
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import aiohttp_client
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+CONF_SITE_ID = "site_id"
+DEFAULT_NAME = "SolarEdge"
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_NAME, default=DEFAULT_NAME): str,
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_PASSWORD): str,
+        vol.Required(CONF_SITE_ID): str,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+    """
+    api = SolarEdgeWeb(
+        username=data[CONF_USERNAME],
+        password=data[CONF_PASSWORD],
+        site_id=data[CONF_SITE_ID],
+        session=aiohttp_client.async_get_clientsession(hass),
+    )
+    try:
+        await api.async_get_equipment()
+    except aiohttp.ClientResponseError as err:
+        if err.status == 401:
+            _LOGGER.error("Invalid credentials")
+            raise InvalidAuth from err
+        if err.status == 403:
+            _LOGGER.error("Invalid credentials for site ID: %s", data[CONF_SITE_ID])
+            raise InvalidAuth from err
+        if err.status == 400:
+            _LOGGER.error("Invalid site ID: %s", data[CONF_SITE_ID])
+            raise CannotConnect from err
+        raise CannotConnect from err
+    except aiohttp.ClientError as err:
+        raise CannotConnect from err
+
+
+class SolarEdgeModulesConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for SolarEdge Modules."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            await self.async_set_unique_id(user_input[CONF_SITE_ID])
+            self._abort_if_unique_id_configured()
+            try:
+                await validate_input(self.hass, user_input)
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(
+                    title=user_input[CONF_NAME],
+                    data={
+                        CONF_SITE_ID: user_input[CONF_SITE_ID],
+                        CONF_USERNAME: user_input[CONF_USERNAME],
+                        CONF_PASSWORD: user_input[CONF_PASSWORD],
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+
+class CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""
diff --git a/homeassistant/components/solaredge_modules/const.py b/homeassistant/components/solaredge_modules/const.py
@@ -0,0 +1,4 @@
+"""Constants for the SolarEdge Modules integration."""
+
+DOMAIN = "solaredge_modules"
+CONF_SITE_ID = "site_id"
diff --git a/homeassistant/components/solaredge_modules/coordinator.py b/homeassistant/components/solaredge_modules/coordinator.py
@@ -0,0 +1,185 @@
+"""Provides the data update coordinator for SolarEdge Modules."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from datetime import datetime, timedelta
+import logging
+from typing import Any
+
+from solaredge_web import EnergyData, SolarEdgeWeb, TimeUnit
+
+from homeassistant.components.recorder import get_instance
+from homeassistant.components.recorder.models import (
+    StatisticData,
+    StatisticMeanType,
+    StatisticMetaData,
+)
+from homeassistant.components.recorder.statistics import (
+    async_add_external_statistics,
+    get_last_statistics,
+    statistics_during_period,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, UnitOfEnergy
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import aiohttp_client
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from homeassistant.util import dt as dt_util
+
+from .const import CONF_SITE_ID, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+type SolarEdgeModulesConfigEntry = ConfigEntry[SolarEdgeModulesCoordinator]
+
+
+class SolarEdgeModulesCoordinator(DataUpdateCoordinator[None]):
+    """Handle fetching SolarEdge Modules data and inserting statistics."""
+
+    config_entry: SolarEdgeModulesConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: SolarEdgeModulesConfigEntry,
+    ) -> None:
+        """Initialize the data handler."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=config_entry,
+            name="SolarEdge Modules",
+            # API refreshes every 15 minutes, but since we only have statistics
+            # and no sensors, refresh every 12h.
+            update_interval=timedelta(hours=12),
+        )
+        self.api = SolarEdgeWeb(
+            username=config_entry.data[CONF_USERNAME],
+            password=config_entry.data[CONF_PASSWORD],
+            site_id=config_entry.data[CONF_SITE_ID],
+            session=aiohttp_client.async_get_clientsession(hass),
+        )
+        self.site_id = config_entry.data[CONF_SITE_ID]
+        self.title = config_entry.title
+
+        @callback
+        def _dummy_listener() -> None:
+            pass
+
+        # Force the coordinator to periodically update by registering a listener.
+        # Needed because there are no sensors added.
+        self.async_add_listener(_dummy_listener)
+
+    async def _async_update_data(self) -> None:
+        """Fetch data from API endpoint and update statistics."""
+        equipment: dict[int, dict[str, Any]] = await self.api.async_get_equipment()
+        # We fetch last week's data from the API and refresh every 12h so we overwrite recent
+        # statistics. This is intended to allow adding any corrected/updated data from the API.
+        energy_data_list: list[EnergyData] = await self.api.async_get_energy_data(
+            TimeUnit.WEEK
+        )
+        if not energy_data_list:
+            _LOGGER.warning(
+                "No data received from SolarEdge API for site: %s", self.site_id
+            )
+            return
+        last_sums = await self._async_get_last_sums(
+            equipment.keys(),
+            energy_data_list[0].start_time.replace(
+                tzinfo=dt_util.get_default_time_zone()
+            ),
+        )
+        for equipment_id, equipment_data in equipment.items():
+            display_name = equipment_data.get(
+                "displayName", f"Equipment {equipment_id}"
+            )
+            statistic_id = self.get_statistic_id(equipment_id)
+            statistic_metadata = StatisticMetaData(
+                mean_type=StatisticMeanType.ARITHMETIC,
+                has_sum=True,
+                name=f"{self.title} {display_name}",
+                source=DOMAIN,
+                statistic_id=statistic_id,
+                unit_of_measurement=UnitOfEnergy.WATT_HOUR,
+            )
+            statistic_sum = last_sums[statistic_id]
+            statistics = []
+            current_hour_sum = 0.0
+            current_hour_count = 0
+            for energy_data in energy_data_list:
+                date = energy_data.start_time.replace(
+                    tzinfo=dt_util.get_default_time_zone()
+                )
+                value = energy_data.values.get(equipment_id, 0.0)
+                current_hour_sum += value
+                current_hour_count += 1
+                if date.minute != 45:
+                    continue
+                # API returns data every 15 minutes; aggregate to 1-hour statistics
+                # when we reach the energy_data for the last 15 minutes of the hour.
+                current_avg = current_hour_sum / current_hour_count
+                statistic_sum += current_avg
+                statistics.append(
+                    StatisticData(
+                        start=date - timedelta(minutes=45),
+                        state=current_avg,
+                        sum=statistic_sum,
+                    )
+                )
+                current_hour_sum = 0.0
+                current_hour_count = 0
+            _LOGGER.debug(
+                "Adding %s statistics for %s %s",
+                len(statistics),
+                statistic_id,
+                display_name,
+            )
+            async_add_external_statistics(self.hass, statistic_metadata, statistics)
+
+    def get_statistic_id(self, equipment_id: int) -> str:
+        """Return the statistic ID for this equipment_id."""
+        return f"{DOMAIN}:{self.site_id}_{equipment_id}"
+
+    async def _async_get_last_sums(
+        self, equipment_ids: Iterable[int], start_time: datetime
+    ) -> dict[str, float]:
+        """Get the last sum from the recorder before start_time for each statistic."""
+        start = start_time - timedelta(hours=1)
+        statistic_ids = {self.get_statistic_id(eq_id) for eq_id in equipment_ids}
+        _LOGGER.debug(
+            "Getting sum for %s statistic IDs at: %s", len(statistic_ids), start
+        )
+        current_stats = await get_instance(self.hass).async_add_executor_job(
+            statistics_during_period,
+            self.hass,
+            start,
+            start + timedelta(seconds=1),
+            statistic_ids,
+            "hour",
+            None,
+            {"sum"},
+        )
+        result = {}
+        for statistic_id in statistic_ids:
+            if statistic_id in current_stats:
+                statistic_sum = current_stats[statistic_id][0]["sum"]
+            else:
+                # If no statistics found right before start_time, try to get the last statistic
+                # but use it only if it's before start_time.
+                # This is needed if the integration hasn't run successfully for at least a week.
+                last_stat = await get_instance(self.hass).async_add_executor_job(
+                    get_last_statistics, self.hass, 1, statistic_id, True, {"sum"}
+                )
+                if (
+                    last_stat
+                    and last_stat[statistic_id][0]["start"] < start_time.timestamp()
+                ):
+                    statistic_sum = last_stat[statistic_id][0]["sum"]
+                else:
+                    # Expected for new installations or if the statistics were cleared,
+                    # e.g. from the developer tools
+                    statistic_sum = 0.0
+            assert isinstance(statistic_sum, float)
+            result[statistic_id] = statistic_sum
+        return result
diff --git a/homeassistant/components/solaredge_modules/manifest.json b/homeassistant/components/solaredge_modules/manifest.json
@@ -0,0 +1,19 @@
+{
+  "domain": "solaredge_modules",
+  "name": "SolarEdge Modules",
+  "codeowners": ["@tronikos"],
+  "config_flow": true,
+  "dependencies": ["recorder"],
+  "dhcp": [
+    {
+      "hostname": "target",
+      "macaddress": "002702*"
+    }
+  ],
+  "documentation": "https://www.home-assistant.io/integrations/solaredge_modules",
+  "integration_type": "service",
+  "iot_class": "cloud_polling",
+  "loggers": ["solaredge_web"],
+  "quality_scale": "bronze",
+  "requirements": ["solaredge-web==0.0.1"]
+}