Add SolarEdge Optimizers Home Assistant integration

.gitignore

@@ -0,0 +1,7 @@
+.coverage
+.pytest_cache/
+.ruff_cache/
+.venv/
+__pycache__/
+*.py[cod]
+

LICENSE

@@ -0,0 +1,22 @@
+MIT License
+
+Copyright (c) 2026 Jens Neuber
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+

README.md

@@ -0,0 +1,62 @@
+# SolarEdge Optimizers for Home Assistant
+
+Diese Custom Integration liest den lokalen Cache der
+[SolarEdge Optimizer Data App](https://git.jensneuber.de/jens/solaredge-optimizers)
+und stellt jeden Leistungsoptimierer als eigenes Home-Assistant-Gerät bereit.
+
+Pro Optimierer werden zwei Sensoren angelegt:
+
+- **Tagesenergie** in Wh (`total_increasing`)
+- **Aktuelle Leistung** in W (`measurement`)
+
+Die Seriennummer bleibt die stabile Gerätekennung. Optimierer-ID, String-ID,
+Wechselrichter-ID und Zeitpunkt der letzten Messung stehen als Attribute zur
+Verfügung.
+
+## Voraussetzungen
+
+- Home Assistant 2026.6 oder neuer
+- SolarEdge Optimizer Data App 0.2.0 oder neuer
+- eine laufende und konfigurierte App mit bereits verfügbaren Optimizer-Daten
+
+Die SolarEdge-Zugangsdaten bleiben ausschließlich in der App-Konfiguration.
+Die Integration spricht nur mit der lokalen App-API.
+
+## Installation mit HACS
+
+1. In HACS **Integrationen** öffnen.
+2. Über das Menü **Benutzerdefinierte Repositories** öffnen.
+3. `https://git.jensneuber.de/jens/ha-solaredge-optimizers` als Typ
+   **Integration** hinzufügen.
+4. **SolarEdge Optimizers** installieren.
+5. Home Assistant neu starten.
+
+Alternativ den Ordner `custom_components/solaredge_optimizers` nach
+`/config/custom_components/solaredge_optimizers` kopieren und Home Assistant
+neu starten.
+
+## Einrichtung
+
+Bei Home Assistant OS/Supervised meldet sich die App automatisch bei Home
+Assistant. Unter **Einstellungen → Geräte & Dienste** erscheint anschließend
+eine gefundene Integration **SolarEdge Optimizers**, die nur noch bestätigt
+werden muss.
+
+Falls keine automatische Erkennung verfügbar ist:
+
+1. **Einstellungen → Geräte & Dienste → Integration hinzufügen** öffnen.
+2. Nach **SolarEdge Optimizers** suchen.
+3. Die Basis-URL der App eintragen, beispielsweise `http://host:8099`.
+
+Die Integration liest den Cache standardmäßig alle 600 Sekunden. Dieser Wert
+kann über **Neu konfigurieren** geändert werden. Dadurch entstehen keine
+zusätzlichen SolarEdge-Portalabrufe; diese steuert die App selbst.
+
+## Entwicklung
+
+```shell
+uv sync
+uv run ruff check .
+uv run pytest
+```
+

custom_components/solaredge_optimizers/__init__.py

@@ -0,0 +1,54 @@
+"""SolarEdge Optimizers integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_URL, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .api import SolarEdgeOptimizerApiClient
+from .const import CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL
+from .coordinator import SolarEdgeOptimizerCoordinator
+
+PLATFORMS = [Platform.SENSOR]
+type SolarEdgeOptimizerConfigEntry = ConfigEntry[SolarEdgeOptimizerCoordinator]
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: SolarEdgeOptimizerConfigEntry
+) -> bool:
+    """Set up SolarEdge Optimizers from a config entry."""
+    scan_seconds = entry.data.get(
+        CONF_SCAN_INTERVAL, int(DEFAULT_SCAN_INTERVAL.total_seconds())
+    )
+    client = SolarEdgeOptimizerApiClient(
+        entry.data[CONF_URL], async_get_clientsession(hass)
+    )
+    coordinator = SolarEdgeOptimizerCoordinator(
+        hass,
+        client,
+        timedelta(seconds=scan_seconds),
+    )
+    await coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = coordinator
+    entry.async_on_unload(entry.add_update_listener(_async_reload_entry))
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: SolarEdgeOptimizerConfigEntry
+) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+
+async def _async_reload_entry(
+    hass: HomeAssistant, entry: SolarEdgeOptimizerConfigEntry
+) -> None:
+    """Reload after the entry is reconfigured."""
+    await hass.config_entries.async_reload(entry.entry_id)

custom_components/solaredge_optimizers/api.py

@@ -0,0 +1,225 @@
+"""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}
+
+
+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)

custom_components/solaredge_optimizers/config_flow.py

@@ -0,0 +1,194 @@
+"""Config flow for SolarEdge Optimizers."""
+
+from __future__ import annotations
+
+from typing import Any, override
+
+import voluptuous as vol
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_URL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.service_info.hassio import HassioServiceInfo
+
+from .api import (
+    OptimizerSnapshot,
+    SolarEdgeOptimizerApiClient,
+    SolarEdgeOptimizerConnectionError,
+    SolarEdgeOptimizerInvalidResponseError,
+    normalize_base_url,
+)
+from .const import (
+    CONF_SCAN_INTERVAL,
+    DEFAULT_PORT,
+    DEFAULT_SCAN_INTERVAL,
+    DOMAIN,
+    LOGGER,
+    MIN_SCAN_INTERVAL_SECONDS,
+)
+
+
+async def validate_input(
+    hass: HomeAssistant, data: dict[str, Any]
+) -> tuple[str, OptimizerSnapshot]:
+    """Validate user input and return the normalized URL and snapshot."""
+    url = normalize_base_url(data[CONF_URL])
+    client = SolarEdgeOptimizerApiClient(url, async_get_clientsession(hass))
+    return url, await client.async_get_optimizers()
+
+
+def _schema(defaults: dict[str, Any]) -> vol.Schema:
+    """Return the manual and reconfigure schema."""
+    url_marker: vol.Marker
+    if CONF_URL in defaults:
+        url_marker = vol.Required(CONF_URL, default=defaults[CONF_URL])
+    else:
+        url_marker = vol.Required(CONF_URL)
+    return vol.Schema(
+        {
+            url_marker: str,
+            vol.Required(
+                CONF_SCAN_INTERVAL,
+                default=defaults.get(
+                    CONF_SCAN_INTERVAL,
+                    int(DEFAULT_SCAN_INTERVAL.total_seconds()),
+                ),
+            ): vol.All(vol.Coerce(int), vol.Range(min=MIN_SCAN_INTERVAL_SECONDS)),
+        }
+    )
+
+
+class SolarEdgeOptimizersConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for SolarEdge Optimizers."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the flow."""
+        self._discovered_url: str | None = None
+        self._discovered_name: str | None = None
+
+    @override
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle manual setup."""
+        return await self._async_handle_form("user", user_input)
+
+    @override
+    async def async_step_hassio(
+        self, discovery_info: HassioServiceInfo
+    ) -> ConfigFlowResult:
+        """Handle discovery from the Home Assistant App."""
+        host = discovery_info.config.get("host")
+        port = discovery_info.config.get("port", DEFAULT_PORT)
+        if not isinstance(host, str) or not host:
+            return self.async_abort(reason="invalid_discovery")
+        try:
+            port = int(port)
+            self._discovered_url = normalize_base_url(f"http://{host}:{port}")
+        except TypeError, ValueError:
+            return self.async_abort(reason="invalid_discovery")
+
+        self._discovered_name = discovery_info.name
+        self.context["title_placeholders"] = {"name": discovery_info.name}
+        return await self.async_step_hassio_confirm()
+
+    async def async_step_hassio_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Confirm App discovery."""
+        if user_input is None:
+            self._abort_if_unique_id_configured()
+            return self.async_show_form(
+                step_id="hassio_confirm",
+                description_placeholders={"name": self._discovered_name or "App"},
+            )
+
+        assert self._discovered_url is not None
+        data = {
+            CONF_URL: self._discovered_url,
+            CONF_SCAN_INTERVAL: int(DEFAULT_SCAN_INTERVAL.total_seconds()),
+        }
+        try:
+            url, snapshot = await validate_input(self.hass, data)
+        except SolarEdgeOptimizerConnectionError:
+            return self.async_abort(reason="cannot_connect")
+        except SolarEdgeOptimizerInvalidResponseError:
+            return self.async_abort(reason="invalid_response")
+        except Exception:  # noqa: BLE001
+            LOGGER.exception("Unexpected exception during App discovery")
+            return self.async_abort(reason="unknown")
+
+        await self.async_set_unique_id(snapshot.site_id)
+        self._abort_if_unique_id_configured(updates={CONF_URL: url})
+        return self.async_create_entry(
+            title=f"SolarEdge Site {snapshot.site_id}", data={**data, CONF_URL: url}
+        )
+
+    @override
+    async def async_step_reconfigure(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Reconfigure the App connection."""
+        entry = self._get_reconfigure_entry()
+        if user_input is None:
+            return self.async_show_form(
+                step_id="reconfigure", data_schema=_schema(dict(entry.data))
+            )
+
+        result = await self._async_validate_form(user_input)
+        if isinstance(result, dict):
+            return self.async_show_form(
+                step_id="reconfigure",
+                data_schema=_schema(user_input),
+                errors=result,
+            )
+        url, snapshot = result
+        if entry.unique_id != snapshot.site_id:
+            return self.async_show_form(
+                step_id="reconfigure",
+                data_schema=_schema(user_input),
+                errors={"base": "site_id_mismatch"},
+            )
+        return self.async_update_reload_and_abort(
+            entry,
+            data_updates={**user_input, CONF_URL: url},
+        )
+
+    async def _async_handle_form(
+        self, step_id: str, user_input: dict[str, Any] | None
+    ) -> ConfigFlowResult:
+        if user_input is None:
+            return self.async_show_form(step_id=step_id, data_schema=_schema({}))
+
+        result = await self._async_validate_form(user_input)
+        if isinstance(result, dict):
+            return self.async_show_form(
+                step_id=step_id,
+                data_schema=_schema(user_input),
+                errors=result,
+            )
+
+        url, snapshot = result
+        await self.async_set_unique_id(snapshot.site_id)
+        self._abort_if_unique_id_configured(updates={CONF_URL: url})
+        return self.async_create_entry(
+            title=f"SolarEdge Site {snapshot.site_id}",
+            data={**user_input, CONF_URL: url},
+        )
+
+    async def _async_validate_form(
+        self, user_input: dict[str, Any]
+    ) -> tuple[str, OptimizerSnapshot] | dict[str, str]:
+        try:
+            return await validate_input(self.hass, user_input)
+        except ValueError:
+            return {"base": "invalid_url"}
+        except SolarEdgeOptimizerConnectionError:
+            return {"base": "cannot_connect"}
+        except SolarEdgeOptimizerInvalidResponseError:
+            return {"base": "invalid_response"}
+        except Exception:  # noqa: BLE001
+            LOGGER.exception("Unexpected exception while validating configuration")
+            return {"base": "unknown"}

custom_components/solaredge_optimizers/const.py

@@ -0,0 +1,15 @@
+"""Constants for the SolarEdge Optimizers integration."""
+
+import logging
+from datetime import timedelta
+
+DOMAIN = "solaredge_optimizers"
+LOGGER = logging.getLogger(__package__)
+
+CONF_SCAN_INTERVAL = "scan_interval"
+DEFAULT_SCAN_INTERVAL = timedelta(minutes=10)
+MIN_SCAN_INTERVAL_SECONDS = 60
+DEFAULT_PORT = 8099
+
+API_PATH = "/api/optimizers"
+API_TIMEOUT_SECONDS = 20

custom_components/solaredge_optimizers/coordinator.py

@@ -0,0 +1,41 @@
+"""DataUpdateCoordinator for SolarEdge Optimizers."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .api import (
+    OptimizerSnapshot,
+    SolarEdgeOptimizerApiClient,
+    SolarEdgeOptimizerApiError,
+)
+from .const import DOMAIN, LOGGER
+
+
+class SolarEdgeOptimizerCoordinator(DataUpdateCoordinator[OptimizerSnapshot]):
+    """Coordinate shared polling of the App API."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        client: SolarEdgeOptimizerApiClient,
+        update_interval: timedelta,
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            LOGGER,
+            name=DOMAIN,
+            update_interval=update_interval,
+            always_update=False,
+        )
+        self.client = client
+
+    async def _async_update_data(self) -> OptimizerSnapshot:
+        try:
+            return await self.client.async_get_optimizers()
+        except SolarEdgeOptimizerApiError as err:
+            raise UpdateFailed("Error communicating with optimizer App") from err

custom_components/solaredge_optimizers/diagnostics.py

@@ -0,0 +1,27 @@
+"""Diagnostics for SolarEdge Optimizers."""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+from typing import Any
+
+from homeassistant.core import HomeAssistant
+
+from . import SolarEdgeOptimizerConfigEntry
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, entry: SolarEdgeOptimizerConfigEntry
+) -> dict[str, Any]:
+    """Return diagnostics for a config entry."""
+    return {
+        "entry": {
+            "title": entry.title,
+            "unique_id": entry.unique_id,
+            "data": dict(entry.data),
+        },
+        "coordinator": {
+            "last_update_success": entry.runtime_data.last_update_success,
+            "snapshot": asdict(entry.runtime_data.data),
+        },
+    }

custom_components/solaredge_optimizers/manifest.json

@@ -0,0 +1,12 @@
+{
+  "domain": "solaredge_optimizers",
+  "name": "SolarEdge Optimizers",
+  "codeowners": ["@jensneuber"],
+  "config_flow": true,
+  "documentation": "https://git.jensneuber.de/jens/ha-solaredge-optimizers",
+  "integration_type": "hub",
+  "iot_class": "local_polling",
+  "issue_tracker": "https://git.jensneuber.de/jens/ha-solaredge-optimizers/issues",
+  "requirements": [],
+  "version": "0.1.0"
+}

custom_components/solaredge_optimizers/sensor.py

@@ -0,0 +1,145 @@
+"""Sensors for SolarEdge Optimizers."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.const import UnitOfEnergy, UnitOfPower
+from homeassistant.core import callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import SolarEdgeOptimizerConfigEntry
+from .api import OptimizerData
+from .const import DOMAIN
+from .coordinator import SolarEdgeOptimizerCoordinator
+
+type OptimizerValue = float | None
+
+
+@dataclass(frozen=True, kw_only=True)
+class SolarEdgeOptimizerSensorDescription(SensorEntityDescription):
+    """Describe an optimizer sensor."""
+
+    value_fn: Callable[[OptimizerData], OptimizerValue]
+
+
+SENSOR_DESCRIPTIONS = (
+    SolarEdgeOptimizerSensorDescription(
+        key="daily_energy",
+        translation_key="daily_energy",
+        device_class=SensorDeviceClass.ENERGY,
+        native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
+        state_class=SensorStateClass.TOTAL_INCREASING,
+        suggested_display_precision=1,
+        value_fn=lambda optimizer: optimizer.daily_energy_wh,
+    ),
+    SolarEdgeOptimizerSensorDescription(
+        key="current_power",
+        translation_key="current_power",
+        device_class=SensorDeviceClass.POWER,
+        native_unit_of_measurement=UnitOfPower.WATT,
+        state_class=SensorStateClass.MEASUREMENT,
+        suggested_display_precision=1,
+        value_fn=lambda optimizer: optimizer.current_power_w,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: Any,
+    entry: SolarEdgeOptimizerConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up optimizer sensors from a config entry."""
+    coordinator = entry.runtime_data
+    known_serials: set[str] = set()
+
+    @callback
+    def async_add_new_optimizers() -> None:
+        new_optimizers = [
+            optimizer
+            for optimizer in coordinator.data.optimizers
+            if optimizer.serial not in known_serials
+        ]
+        if not new_optimizers:
+            return
+        known_serials.update(optimizer.serial for optimizer in new_optimizers)
+        async_add_entities(
+            SolarEdgeOptimizerSensor(coordinator, optimizer.serial, description)
+            for optimizer in new_optimizers
+            for description in SENSOR_DESCRIPTIONS
+        )
+
+    async_add_new_optimizers()
+    entry.async_on_unload(coordinator.async_add_listener(async_add_new_optimizers))
+
+
+class SolarEdgeOptimizerSensor(
+    CoordinatorEntity[SolarEdgeOptimizerCoordinator], SensorEntity
+):
+    """Represent one optimizer reading."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: SolarEdgeOptimizerCoordinator,
+        serial: str,
+        description: SolarEdgeOptimizerSensorDescription,
+    ) -> None:
+        """Initialize a sensor."""
+        super().__init__(coordinator)
+        self.entity_description = description
+        self._serial = serial
+        self._attr_unique_id = f"{coordinator.data.site_id}_{serial}_{description.key}"
+
+    @property
+    def _optimizer(self) -> OptimizerData | None:
+        return self.coordinator.data.optimizers_by_serial.get(self._serial)
+
+    @property
+    def native_value(self) -> OptimizerValue:
+        """Return the current sensor value."""
+        if (optimizer := self._optimizer) is None:
+            return None
+        return self.entity_description.value_fn(optimizer)
+
+    @property
+    def available(self) -> bool:
+        """Return whether this individual reading is available."""
+        return super().available and self.native_value is not None
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return optimizer device information."""
+        optimizer = self._optimizer
+        optimizer_id = optimizer.optimizer_id if optimizer else None
+        return DeviceInfo(
+            identifiers={(DOMAIN, f"{self.coordinator.data.site_id}_{self._serial}")},
+            manufacturer="SolarEdge",
+            model="Power Optimizer",
+            name=f"Optimizer {optimizer_id or self._serial}",
+            serial_number=self._serial,
+        )
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Return stable optimizer metadata."""
+        if (optimizer := self._optimizer) is None:
+            return {}
+        return {
+            "optimizer_id": optimizer.optimizer_id,
+            "string_id": optimizer.string_id,
+            "inverter_id": optimizer.inverter_id,
+            "last_measurement": optimizer.last_measurement,
+        }

custom_components/solaredge_optimizers/strings.json

@@ -0,0 +1,57 @@
+{
+  "title": "SolarEdge Optimizers",
+  "config": {
+    "step": {
+      "user": {
+        "title": "Connect to SolarEdge Optimizer Data",
+        "description": "Enter the base URL of the SolarEdge Optimizer Data App API.",
+        "data": {
+          "url": "App URL",
+          "scan_interval": "Polling interval (seconds)"
+        },
+        "data_description": {
+          "url": "For example http://host:8099. The /api/optimizers path is added automatically.",
+          "scan_interval": "How often Home Assistant reads the App cache. The App refreshes SolarEdge data independently."
+        }
+      },
+      "hassio_confirm": {
+        "title": "Set up {name}",
+        "description": "Use the automatically discovered {name} App?"
+      },
+      "reconfigure": {
+        "title": "Reconfigure SolarEdge Optimizers",
+        "description": "Update the connection to the local App API.",
+        "data": {
+          "url": "App URL",
+          "scan_interval": "Polling interval (seconds)"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "Failed to connect to the App",
+      "invalid_response": "The App returned an invalid response",
+      "invalid_url": "Enter a valid HTTP or HTTPS URL",
+      "site_id_mismatch": "The new App URL belongs to a different SolarEdge site",
+      "unknown": "Unexpected error"
+    },
+    "abort": {
+      "already_configured": "This SolarEdge site is already configured",
+      "cannot_connect": "Failed to connect to the discovered App",
+      "invalid_discovery": "The App supplied invalid discovery data",
+      "invalid_response": "The discovered App returned an invalid response",
+      "reconfigure_successful": "The integration was reconfigured",
+      "unknown": "Unexpected error"
+    }
+  },
+  "entity": {
+    "sensor": {
+      "daily_energy": {
+        "name": "Daily energy"
+      },
+      "current_power": {
+        "name": "Current power"
+      }
+    }
+  }
+}
+

custom_components/solaredge_optimizers/translations/de.json

@@ -0,0 +1,57 @@
+{
+  "title": "SolarEdge-Optimierer",
+  "config": {
+    "step": {
+      "user": {
+        "title": "Mit SolarEdge Optimizer Data verbinden",
+        "description": "Gib die Basis-URL der API der SolarEdge Optimizer Data App ein.",
+        "data": {
+          "url": "App-URL",
+          "scan_interval": "Abrufintervall (Sekunden)"
+        },
+        "data_description": {
+          "url": "Zum Beispiel http://host:8099. Der Pfad /api/optimizers wird automatisch ergänzt.",
+          "scan_interval": "Wie oft Home Assistant den App-Cache liest. Die App aktualisiert die SolarEdge-Daten unabhängig davon."
+        }
+      },
+      "hassio_confirm": {
+        "title": "{name} einrichten",
+        "description": "Die automatisch gefundene App {name} verwenden?"
+      },
+      "reconfigure": {
+        "title": "SolarEdge-Optimierer neu konfigurieren",
+        "description": "Verbindung zur lokalen App-API aktualisieren.",
+        "data": {
+          "url": "App-URL",
+          "scan_interval": "Abrufintervall (Sekunden)"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "Verbindung zur App fehlgeschlagen",
+      "invalid_response": "Die App hat eine ungültige Antwort geliefert",
+      "invalid_url": "Gib eine gültige HTTP- oder HTTPS-URL ein",
+      "site_id_mismatch": "Die neue App-URL gehört zu einer anderen SolarEdge-Site",
+      "unknown": "Unerwarteter Fehler"
+    },
+    "abort": {
+      "already_configured": "Diese SolarEdge-Site ist bereits eingerichtet",
+      "cannot_connect": "Verbindung zur gefundenen App fehlgeschlagen",
+      "invalid_discovery": "Die App hat ungültige Erkennungsdaten geliefert",
+      "invalid_response": "Die gefundene App hat eine ungültige Antwort geliefert",
+      "reconfigure_successful": "Die Integration wurde neu konfiguriert",
+      "unknown": "Unerwarteter Fehler"
+    }
+  },
+  "entity": {
+    "sensor": {
+      "daily_energy": {
+        "name": "Tagesenergie"
+      },
+      "current_power": {
+        "name": "Aktuelle Leistung"
+      }
+    }
+  }
+}
+

custom_components/solaredge_optimizers/translations/en.json

@@ -0,0 +1,57 @@
+{
+  "title": "SolarEdge Optimizers",
+  "config": {
+    "step": {
+      "user": {
+        "title": "Connect to SolarEdge Optimizer Data",
+        "description": "Enter the base URL of the SolarEdge Optimizer Data App API.",
+        "data": {
+          "url": "App URL",
+          "scan_interval": "Polling interval (seconds)"
+        },
+        "data_description": {
+          "url": "For example http://host:8099. The /api/optimizers path is added automatically.",
+          "scan_interval": "How often Home Assistant reads the App cache. The App refreshes SolarEdge data independently."
+        }
+      },
+      "hassio_confirm": {
+        "title": "Set up {name}",
+        "description": "Use the automatically discovered {name} App?"
+      },
+      "reconfigure": {
+        "title": "Reconfigure SolarEdge Optimizers",
+        "description": "Update the connection to the local App API.",
+        "data": {
+          "url": "App URL",
+          "scan_interval": "Polling interval (seconds)"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "Failed to connect to the App",
+      "invalid_response": "The App returned an invalid response",
+      "invalid_url": "Enter a valid HTTP or HTTPS URL",
+      "site_id_mismatch": "The new App URL belongs to a different SolarEdge site",
+      "unknown": "Unexpected error"
+    },
+    "abort": {
+      "already_configured": "This SolarEdge site is already configured",
+      "cannot_connect": "Failed to connect to the discovered App",
+      "invalid_discovery": "The App supplied invalid discovery data",
+      "invalid_response": "The discovered App returned an invalid response",
+      "reconfigure_successful": "The integration was reconfigured",
+      "unknown": "Unexpected error"
+    }
+  },
+  "entity": {
+    "sensor": {
+      "daily_energy": {
+        "name": "Daily energy"
+      },
+      "current_power": {
+        "name": "Current power"
+      }
+    }
+  }
+}
+

hacs.json

@@ -0,0 +1,6 @@
+{
+  "name": "SolarEdge Optimizers",
+  "render_readme": true,
+  "homeassistant": "2026.6.0"
+}
+

pyproject.toml

@@ -0,0 +1,35 @@
+[project]
+name = "ha-solaredge-optimizers"
+version = "0.1.0"
+description = "Home Assistant custom integration for SolarEdge optimizer data"
+readme = "README.md"
+requires-python = ">=3.14.2"
+license = "MIT"
+authors = [{ name = "Jens Neuber" }]
+
+[dependency-groups]
+dev = [
+  "homeassistant>=2026.6.0",
+  "pytest>=9.0.0",
+  "pytest-homeassistant-custom-component>=0.13.300",
+  "ruff>=0.14.0",
+]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]
+
+[tool.ruff]
+target-version = "py314"
+line-length = 88
+
+[tool.ruff.lint]
+select = ["ASYNC", "B", "E", "F", "I", "SIM", "UP"]
+ignore = [
+  "COM812",
+  "D203",
+  "D213",
+]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/*" = ["S101"]

tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for SolarEdge Optimizers."""

tests/conftest.py

@@ -0,0 +1,12 @@
+"""Test fixtures for SolarEdge Optimizers."""
+
+from __future__ import annotations
+
+import pytest
+
+pytest_plugins = "pytest_homeassistant_custom_component"
+
+
+@pytest.fixture(autouse=True)
+def auto_enable_custom_integrations(enable_custom_integrations: None) -> None:
+    """Enable loading custom integrations in all tests."""

tests/sample_data.py

@@ -0,0 +1,35 @@
+"""Sample App API data."""
+
+SAMPLE_PAYLOAD = {
+    "status": "ok",
+    "siteId": "4886699",
+    "date": "2026-08-10",
+    "timeZone": "Europe/Berlin",
+    "fetchedAt": "2026-08-10T10:00:00.000Z",
+    "nextRefreshAt": "2026-08-10T10:10:00.000Z",
+    "optimizerCount": 2,
+    "successfulOptimizerCount": 2,
+    "failedOptimizerCount": 0,
+    "totalDailyEnergyWh": 2468.5,
+    "totalCurrentPowerW": 321.4,
+    "optimizers": [
+        {
+            "serial": "14F28854-E2",
+            "inverterId": "1",
+            "stringId": "1.1",
+            "optimizerId": "1.1.1",
+            "dailyEnergyWh": 1234.5,
+            "currentPowerW": 160.7,
+            "lastMeasurement": "2026-08-10T09:59:30.000Z",
+        },
+        {
+            "serial": "14F28855-E2",
+            "inverterId": "1",
+            "stringId": "1.1",
+            "optimizerId": "1.1.2",
+            "dailyEnergyWh": 1234.0,
+            "currentPowerW": 160.7,
+            "lastMeasurement": "2026-08-10T09:59:30.000Z",
+        },
+    ],
+}

tests/test_api.py

@@ -0,0 +1,56 @@
+"""Tests for the App API parser."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+
+import pytest
+
+from custom_components.solaredge_optimizers.api import (
+    SolarEdgeOptimizerInvalidResponseError,
+    normalize_base_url,
+    parse_snapshot,
+)
+
+from .sample_data import SAMPLE_PAYLOAD
+
+
+def test_parse_snapshot() -> None:
+    """Parse a valid snapshot."""
+    snapshot = parse_snapshot(SAMPLE_PAYLOAD)
+
+    assert snapshot.site_id == "4886699"
+    assert snapshot.optimizer_count == 2
+    assert snapshot.optimizers[0].optimizer_id == "1.1.1"
+    assert snapshot.optimizers[0].daily_energy_wh == 1234.5
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("http://app:8099/", "http://app:8099"),
+        ("http://app:8099/api/optimizers", "http://app:8099"),
+        ("https://example.test/prefix/", "https://example.test/prefix"),
+    ],
+)
+def test_normalize_base_url(value: str, expected: str) -> None:
+    """Normalize supported URLs."""
+    assert normalize_base_url(value) == expected
+
+
+@pytest.mark.parametrize(
+    "value", ["app:8099", "ftp://app/data", "http://user:password@app:8099"]
+)
+def test_reject_invalid_url(value: str) -> None:
+    """Reject invalid or credential-bearing URLs."""
+    with pytest.raises(ValueError):
+        normalize_base_url(value)
+
+
+def test_reject_duplicate_serial() -> None:
+    """Reject ambiguous optimizer records."""
+    payload = deepcopy(SAMPLE_PAYLOAD)
+    payload["optimizers"][1]["serial"] = payload["optimizers"][0]["serial"]
+
+    with pytest.raises(SolarEdgeOptimizerInvalidResponseError):
+        parse_snapshot(payload)

tests/test_config_flow.py

@@ -0,0 +1,70 @@
+"""Tests for the SolarEdge Optimizers config flow."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_URL
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+from homeassistant.helpers.service_info.hassio import HassioServiceInfo
+
+from custom_components.solaredge_optimizers.api import parse_snapshot
+from custom_components.solaredge_optimizers.const import (
+    CONF_SCAN_INTERVAL,
+    DOMAIN,
+)
+
+from .sample_data import SAMPLE_PAYLOAD
+
+
+async def test_user_flow(hass: HomeAssistant) -> None:
+    """Configure the integration manually."""
+    with patch(
+        "custom_components.solaredge_optimizers.config_flow.validate_input",
+        AsyncMock(return_value=("http://app:8099", parse_snapshot(SAMPLE_PAYLOAD))),
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+            data={CONF_URL: "http://app:8099", CONF_SCAN_INTERVAL: 600},
+        )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["title"] == "SolarEdge Site 4886699"
+    assert result["data"][CONF_URL] == "http://app:8099"
+
+
+async def test_hassio_discovery_flow(hass: HomeAssistant) -> None:
+    """Configure the integration from App discovery."""
+    discovery = HassioServiceInfo(
+        config={"host": "abc-solaredge-optimizer-data", "port": 8099},
+        name="SolarEdge Optimizer Data",
+        slug="abc_solaredge_optimizer_data",
+        uuid="discovery-uuid",
+    )
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_HASSIO},
+        data=discovery,
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "hassio_confirm"
+
+    with patch(
+        "custom_components.solaredge_optimizers.config_flow.validate_input",
+        AsyncMock(
+            return_value=(
+                "http://abc-solaredge-optimizer-data:8099",
+                parse_snapshot(SAMPLE_PAYLOAD),
+            )
+        ),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={}
+        )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["data"][CONF_URL] == "http://abc-solaredge-optimizer-data:8099"

tests/test_sensor.py

@@ -0,0 +1,56 @@
+"""Tests for SolarEdge optimizer sensors."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+
+from homeassistant.const import CONF_URL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers import entity_registry as er
+from pytest_homeassistant_custom_component.common import MockConfigEntry
+
+from custom_components.solaredge_optimizers.api import parse_snapshot
+from custom_components.solaredge_optimizers.const import CONF_SCAN_INTERVAL, DOMAIN
+
+from .sample_data import SAMPLE_PAYLOAD
+
+
+async def test_optimizer_sensors(hass: HomeAssistant) -> None:
+    """Create energy and power sensors for every optimizer."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        title="SolarEdge Site 4886699",
+        unique_id="4886699",
+        data={CONF_URL: "http://app:8099", CONF_SCAN_INTERVAL: 600},
+    )
+    entry.add_to_hass(hass)
+
+    with patch(
+        "custom_components.solaredge_optimizers.api."
+        "SolarEdgeOptimizerApiClient.async_get_optimizers",
+        AsyncMock(return_value=parse_snapshot(SAMPLE_PAYLOAD)),
+    ):
+        assert await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    entity_registry = er.async_get(hass)
+    entities = er.async_entries_for_config_entry(entity_registry, entry.entry_id)
+    assert len(entities) == 4
+
+    energy_entity = next(
+        entity for entity in entities if entity.unique_id.endswith("_daily_energy")
+    )
+    power_entity = next(
+        entity for entity in entities if entity.unique_id.endswith("_current_power")
+    )
+    assert hass.states.get(energy_entity.entity_id).state == "1234.5"
+    assert hass.states.get(power_entity.entity_id).state == "160.7"
+
+    device_registry = dr.async_get(hass)
+    devices = dr.async_entries_for_config_entry(device_registry, entry.entry_id)
+    assert len(devices) == 2
+    assert {device.serial_number for device in devices} == {
+        "14F28854-E2",
+        "14F28855-E2",
+    }