jens/ha-solaredge-optimizers
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add SolarEdge Optimizers Home Assistant integration

Browse Source
This commit is contained in:
jens
2026-08-10 09:31:43 +02:00
commit 2085316d1f
23 changed files with 3956 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+7
View File
@@ -0,0 +1,7 @@
.coverage
.pytest_cache/
.ruff_cache/
.venv/
__pycache__/
*.py[cod]
LICENSE
+22
View File
@@ -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
+62
View File
@@ -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
+54
View File
@@ -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
+225
View File
@@ -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
+194
View File
@@ -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
+15
View File
@@ -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
+41
View File
@@ -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
+27
View File
@@ -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
+12
View File
@@ -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
+145
View File
@@ -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
+57
View File
@@ -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
+57
View File
@@ -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
+57
View File
@@ -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
+6
View File
@@ -0,0 +1,6 @@
{
"name": "SolarEdge Optimizers",
"render_readme": true,
"homeassistant": "2026.6.0"
}
pyproject.toml
+35
View File
@@ -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
+1
View File
@@ -0,0 +1 @@
"""Tests for SolarEdge Optimizers."""
tests/conftest.py
+12
View File
@@ -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
+35
View File
@@ -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
+56
View File
@@ -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
+70
View File
@@ -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
+56
View File
@@ -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",
}
uv.lock
Generated
+2710
View File
File diff suppressed because it is too large Load Diff