home-assistant · fedus · Jun 1, 2025 · Jun 2, 2025 · Jun 10, 2025 · pantherale0
diff --git a/CODEOWNERS b/CODEOWNERS
diff --git a/homeassistant/components/leneda/__init__.py b/homeassistant/components/leneda/__init__.py
@@ -0,0 +1,53 @@
+"""The Leneda integration."""
+
+from __future__ import annotations
+
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import DeviceEntry
+
+from .const import CONF_API_TOKEN, CONF_ENERGY_ID
+from .coordinator import LenedaCoordinator
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+type LenedaConfigEntry = ConfigEntry[LenedaCoordinator]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+async def async_setup_entry(hass: HomeAssistant, entry: LenedaConfigEntry) -> bool:
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+async def async_setup_entry(hass: HomeAssistant, entry: LenedaConfigEntry) -> bool:
+    """Set up Leneda from a config entry or subentry."""
+    _LOGGER.debug("Setting up entry %s", entry.entry_id)
+
+    api_token, energy_id = entry.data[CONF_API_TOKEN], entry.data[CONF_ENERGY_ID]
+    coordinator = LenedaCoordinator(hass, entry, api_token, energy_id)
+    entry.runtime_data = coordinator
+
+    await coordinator.async_config_entry_first_refresh()
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    # Set up a listener for subentry changes
+    entry.async_on_unload(entry.add_update_listener(_async_update_listener))
+
+    return True
+
+
+async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Handle update."""
+    await hass.config_entries.async_reload(entry.entry_id)
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+
+async def async_remove_config_entry_device(
+    hass: HomeAssistant, config_entry: ConfigEntry, device_entry: DeviceEntry
+) -> bool:
+    """Remove a metering point device (only relevant for subentries)."""
+    return True
diff --git a/homeassistant/components/leneda/config_flow.py b/homeassistant/components/leneda/config_flow.py
@@ -0,0 +1,155 @@
+"""Config flow for Leneda integration."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from typing import Any, Final
+
+from leneda import LenedaClient
+from leneda.exceptions import ForbiddenException, UnauthorizedException
+from leneda.obis_codes import ObisCode
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.config_entries import ConfigFlowResult
+from homeassistant.core import callback
+from homeassistant.helpers import selector
+
+from .const import CONF_API_TOKEN, CONF_ENERGY_ID, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+# Setup types
+SETUP_TYPE_PROBE: Final = "probe"
+SETUP_TYPE_MANUAL: Final = "manual"
+
+# Error messages
+ERROR_INVALID_METERING_POINT: Final = "invalid_metering_point"
+ERROR_SELECT_AT_LEAST_ONE: Final = "select_at_least_one"
+ERROR_DUPLICATE_METERING_POINT: Final = "duplicate_metering_point"
+ERROR_FORBIDDEN: Final = "forbidden"
+ERROR_UNAUTHORIZED: Final = "unauthorized"
+
+
+class LenedaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
-class LenedaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+class LenedaConfigFlow(ConfigFlow, domain=DOMAIN):
-class LenedaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+class LenedaConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Leneda (main entry: authentication only)."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._api_token: str = ""
+        self._energy_id: str = ""
+
+    async def async_step_reauth(
+        self, entry_data: Mapping[str, Any]
+    ) -> ConfigFlowResult:
+        """Handle reauthentication flow when API token becomes invalid."""
+        self._energy_id = entry_data[CONF_ENERGY_ID]
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle reauthentication confirmation step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            self._api_token = user_input[CONF_API_TOKEN]
+
+            # Validate new API token
+            try:
+                client = LenedaClient(
+                    api_key=self._api_token,
+                    energy_id=self._energy_id,
+                )
+                # Use a dummy metering point ID to test authentication
+                await client.probe_metering_point_obis_code(
+                    "dummy-metering-point", ObisCode.ELEC_CONSUMPTION_ACTIVE
+                )
+            except UnauthorizedException:
+                errors = {"base": ERROR_UNAUTHORIZED}
+            except ForbiddenException:
+                errors = {"base": ERROR_FORBIDDEN}
+            else:
+                # Update the config entry with new token
+                return self.async_update_reload_and_abort(
+                    self._get_reauth_entry(),
+                    data_updates={CONF_API_TOKEN: self._api_token},
+                )
+
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_API_TOKEN): selector.TextSelector(
+                        selector.TextSelectorConfig(
+                            type=selector.TextSelectorType.PASSWORD,
+                            autocomplete="leneda-api-token",
+                        )
+                    ),
+                }
+            ),
+            description_placeholders={"energy_id": self._energy_id},
+            errors=errors,
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step of the config flow."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            await self.async_set_unique_id(user_input[CONF_ENERGY_ID])
+            self._abort_if_unique_id_configured()
+            self._api_token = user_input[CONF_API_TOKEN]
+            self._energy_id = user_input[CONF_ENERGY_ID]
+
+            # Validate authentication by making a test API call
+            try:
+                client = LenedaClient(
+                    api_key=self._api_token,
+                    energy_id=self._energy_id,
+                )
+                # Use a dummy metering point ID to test authentication
+                await client.probe_metering_point_obis_code(
+                    "dummy-metering-point", ObisCode.ELEC_CONSUMPTION_ACTIVE
+                )
+            except UnauthorizedException:
+                errors = {"base": ERROR_UNAUTHORIZED}
+            except ForbiddenException:
+                errors = {"base": ERROR_FORBIDDEN}
+            else:
+                return self.async_create_entry(
+                    title=self._energy_id,
+                    data={
+                        CONF_API_TOKEN: self._api_token,
+                        CONF_ENERGY_ID: self._energy_id,
+                    },
+                )
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_API_TOKEN): selector.TextSelector(
+                        selector.TextSelectorConfig(
+                            type=selector.TextSelectorType.PASSWORD,
+                            autocomplete="leneda-api-token",
+                        )
+                    ),
+                    vol.Required(CONF_ENERGY_ID): selector.TextSelector(
+                        selector.TextSelectorConfig(
+                            type=selector.TextSelectorType.TEXT,
+                            autocomplete="leneda-energy-id",
+                        )
+                    ),
+                }
+            ),
+            errors=errors,
+        )
+
+    @staticmethod
+    @callback
+    def async_get_config_entry_title(config_entry: config_entries.ConfigEntry) -> str:
+        """Get the title for the config entry."""
+        return config_entry.data.get(CONF_ENERGY_ID, "Leneda")
diff --git a/homeassistant/components/leneda/const.py b/homeassistant/components/leneda/const.py
@@ -0,0 +1,118 @@
+"""Constants for the Leneda integration."""
+
+from datetime import timedelta
+
+from leneda.obis_codes import ObisCode
+
+from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
+from homeassistant.const import (
+    UnitOfEnergy,
+    UnitOfPower,
+    UnitOfReactiveEnergy,
+    UnitOfReactivePower,
+)
+
+DOMAIN = "leneda"
+
+CONF_API_TOKEN = "api_token"
+CONF_ENERGY_ID = "energy_id"
+CONF_METERING_POINT = "metering_point"
+
+SCAN_INTERVAL = timedelta(hours=1)
+
+# Sensor types and their corresponding OBIS codes
+SENSOR_TYPES = {
+    # Electricity Consumption
+    "electricity_consumption_active": {
+        "obis_code": ObisCode.ELEC_CONSUMPTION_ACTIVE,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_consumption_reactive": {
+        "obis_code": ObisCode.ELEC_CONSUMPTION_REACTIVE,
+        "device_class": SensorDeviceClass.REACTIVE_ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_consumption_covered_layer1": {
+        "obis_code": ObisCode.ELEC_CONSUMPTION_COVERED_LAYER1,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_consumption_covered_layer2": {
+        "obis_code": ObisCode.ELEC_CONSUMPTION_COVERED_LAYER2,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_consumption_covered_layer3": {
+        "obis_code": ObisCode.ELEC_CONSUMPTION_COVERED_LAYER3,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_consumption_covered_layer4": {
+        "obis_code": ObisCode.ELEC_CONSUMPTION_COVERED_LAYER4,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_consumption_remaining": {
+        "obis_code": ObisCode.ELEC_CONSUMPTION_REMAINING,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    # Electricity Production
+    "electricity_production_active": {
+        "obis_code": ObisCode.ELEC_PRODUCTION_ACTIVE,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_production_reactive": {
+        "obis_code": ObisCode.ELEC_PRODUCTION_REACTIVE,
+        "device_class": SensorDeviceClass.REACTIVE_ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_production_shared_layer1": {
+        "obis_code": ObisCode.ELEC_PRODUCTION_SHARED_LAYER1,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_production_shared_layer2": {
+        "obis_code": ObisCode.ELEC_PRODUCTION_SHARED_LAYER2,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_production_shared_layer3": {
+        "obis_code": ObisCode.ELEC_PRODUCTION_SHARED_LAYER3,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_production_shared_layer4": {
+        "obis_code": ObisCode.ELEC_PRODUCTION_SHARED_LAYER4,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "electricity_production_remaining": {
+        "obis_code": ObisCode.ELEC_PRODUCTION_REMAINING,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    # Gas Consumption
+    "gas_consumption_volume": {
+        "obis_code": ObisCode.GAS_CONSUMPTION_VOLUME,
+        "device_class": SensorDeviceClass.GAS,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "gas_consumption_standard_volume": {
+        "obis_code": ObisCode.GAS_CONSUMPTION_STANDARD_VOLUME,
+        "device_class": None,  # Ideally SensorDeviceClass.GAS, but Nm3 not supported by Hass
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+    "gas_consumption_energy": {
+        "obis_code": ObisCode.GAS_CONSUMPTION_ENERGY,
+        "device_class": SensorDeviceClass.ENERGY,
+        "state_class": SensorStateClass.TOTAL_INCREASING,
+    },
+}
+
+UNIT_TO_AGGREGATED_UNIT = {
+    UnitOfPower.KILO_WATT.lower(): UnitOfEnergy.KILO_WATT_HOUR,
+    UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE.lower(): UnitOfReactiveEnergy.KILO_VOLT_AMPERE_REACTIVE_HOUR,
+}