Add lektrico integration (home-assistant#102371)

* Add Lektrico Integration

* Make the changes proposed by Lash-L: new coordinator.py, new entity.py; use: translation_key, last_update_sucess, PlatformNotReady; remove: global variables

* Replace FlowResult with ConfigFlowResult and add tests.

* Remove unused lines.

* Remove Options from condif_flow

* Fix ruff and mypy.

* Fix CODEOWNERS.

* Run python3 -m script.hassfest.

* Correct rebase mistake.

* Make modifications suggested by emontnemery.

* Add pytest fixtures.

* Remove meaningless patches.

* Update .coveragerc

* Replace CONF_FRIENDLY_NAME with CONF_NAME.

* Remove underscores.

* Update tests.

* Update test file with is and no config_entries. .

* Set serial_number in DeviceInfo and add return type of the async_update_data to DataUpdateCoordinator.

* Use suggested_unit_of_measurement for KILO_WATT and replace Any in value_fn (sensor file).

* Add device class duration to charging_time sensor.

* Change raising  PlatformNotReady to raising IntegrationError.

* Test the unique id of the entry.

* Rename PF Lx with Power factor Lx and remove PF from strings.json.

* Remove comment.

* Make state and limit reason sensors to be enum sensors.

* Use result variable to check unique_id in test.

* Remove CONF_NAME from entry and __init__ from LektricoFlowHandler.

* Remove session parameter from LektricoDeviceDataUpdateCoordinator.

* Use config_entry: ConfigEntry in coordinator.

* Replace Connected,NeedAuth with Waiting for Authentication.

* Use lektricowifi 0.0.29.

* Use lektricowifi 0.0.39

* Use lektricowifi 0.0.40

* Use lektricowifi 0.0.41

* Replace hass.data with entry.runtime_data

* Delete .coveragerc

* Restructure the user step

* Fix tests

* Add returned value of _async_update_data to class DataUpdateCoordinator

* Use hw_version at DeviceInfo

* Remove a variable

* Use StateType

* Replace friendly_name with device_name

* Use sentence case in translation strings

* Uncomment and fix test_discovered_zeroconf

* Add type LektricoConfigEntry

* Remove commented code

* Remove the type of coordinator in sensor async_setup_entry

* Make zeroconf test end in ABORT, not FORM

* Remove all async_block_till_done from tests

* End test_user_setup_device_offline with CREATE_ENTRY

* Patch the full Device

* Add snapshot tests

* Overwrite the type LektricoSensorEntityDescription outside of the constructor

* Test separate already_configured for zeroconf

---------

Co-authored-by: mihaela.tarjoianu <[email protected]>
Co-authored-by: Erik Montnemery <[email protected]>

.strict-typing

@@ -279,6 +279,7 @@ homeassistant.components.lawn_mower.*
 homeassistant.components.lcn.*
 homeassistant.components.ld2410_ble.*
 homeassistant.components.led_ble.*
+homeassistant.components.lektrico.*
 homeassistant.components.lidarr.*
 homeassistant.components.lifx.*
 homeassistant.components.light.*

CODEOWNERS

@@ -799,6 +799,8 @@ build.json @home-assistant/supervisor
 /tests/components/leaone/ @bdraco
 /homeassistant/components/led_ble/ @bdraco
 /tests/components/led_ble/ @bdraco
+/homeassistant/components/lektrico/ @lektrico
+/tests/components/lektrico/ @lektrico
 /homeassistant/components/lg_netcast/ @Drafteed @splinter98
 /tests/components/lg_netcast/ @Drafteed @splinter98
 /homeassistant/components/lidarr/ @tkdrob

homeassistant/components/lektrico/__init__.py

@@ -0,0 +1,51 @@
+"""The Lektrico Charging Station integration."""
+
+from __future__ import annotations
+
+from lektricowifi import Device
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import ATTR_SERIAL_NUMBER, CONF_TYPE, Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import LektricoDeviceDataUpdateCoordinator
+
+# List the platforms that charger supports.
+CHARGERS_PLATFORMS = [Platform.SENSOR]
+
+# List the platforms that load balancer device supports.
+LB_DEVICES_PLATFORMS = [Platform.SENSOR]
+
+type LektricoConfigEntry = ConfigEntry[LektricoDeviceDataUpdateCoordinator]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: LektricoConfigEntry) -> bool:
+    """Set up Lektrico Charging Station from a config entry."""
+    coordinator = LektricoDeviceDataUpdateCoordinator(
+        hass,
+        f"{entry.data[CONF_TYPE]}_{entry.data[ATTR_SERIAL_NUMBER]}",
+    )
+
+    await coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, _get_platforms(entry))
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+
+    return await hass.config_entries.async_unload_platforms(
+        entry, _get_platforms(entry)
+    )
+
+
+def _get_platforms(entry: ConfigEntry) -> list[Platform]:
+    """Return the platforms for this type of device."""
+    _device_type: str = entry.data[CONF_TYPE]
+    if _device_type in (Device.TYPE_1P7K, Device.TYPE_3P22K):
+        return CHARGERS_PLATFORMS
+    return LB_DEVICES_PLATFORMS

homeassistant/components/lektrico/config_flow.py

@@ -0,0 +1,138 @@
+"""Config flow for Lektrico Charging Station."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from lektricowifi import Device, DeviceConnectionError
+import voluptuous as vol
+
+from homeassistant.components.zeroconf import ZeroconfServiceInfo
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import (
+    ATTR_HW_VERSION,
+    ATTR_SERIAL_NUMBER,
+    CONF_HOST,
+    CONF_TYPE,
+)
+from homeassistant.core import callback
+from homeassistant.helpers.httpx_client import get_async_client
+
+from .const import DOMAIN
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): str,
+    }
+)
+
+
+class LektricoFlowHandler(ConfigFlow, domain=DOMAIN):
+    """Handle a Lektrico config flow."""
+
+    VERSION = 1
+
+    _host: str
+    _name: str
+    _serial_number: str
+    _board_revision: str
+    _device_type: str
+
+    async def async_step_user(
+        self, user_input: dict[str, str] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a flow initiated by the user."""
+        errors = None
+
+        if user_input is not None:
+            self._host = user_input[CONF_HOST]
+
+            # obtain serial number
+            try:
+                await self._get_lektrico_device_settings_and_treat_unique_id()
+                return self._async_create_entry()
+            except DeviceConnectionError:
+                errors = {CONF_HOST: "cannot_connect"}
+
+        return self._async_show_setup_form(user_input=user_input, errors=errors)
+
+    @callback
+    def _async_show_setup_form(
+        self,
+        user_input: dict[str, Any] | None = None,
+        errors: dict[str, str] | None = None,
+    ) -> ConfigFlowResult:
+        """Show the setup form to the user."""
+        if user_input is None:
+            user_input = {}
+
+        schema = self.add_suggested_values_to_schema(STEP_USER_DATA_SCHEMA, user_input)
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=schema,
+            errors=errors or {},
+        )
+
+    @callback
+    def _async_create_entry(self) -> ConfigFlowResult:
+        return self.async_create_entry(
+            title=self._name,
+            data={
+                CONF_HOST: self._host,
+                ATTR_SERIAL_NUMBER: self._serial_number,
+                CONF_TYPE: self._device_type,
+                ATTR_HW_VERSION: self._board_revision,
+            },
+        )
+
+    async def async_step_zeroconf(
+        self, discovery_info: ZeroconfServiceInfo
+    ) -> ConfigFlowResult:
+        """Handle zeroconf discovery."""
+        self._host = discovery_info.host  # 192.168.100.11
+
+        # read settings from the device
+        try:
+            await self._get_lektrico_device_settings_and_treat_unique_id()
+        except DeviceConnectionError:
+            return self.async_abort(reason="cannot_connect")
+
+        self.context["title_placeholders"] = {
+            "serial_number": self._serial_number,
+            "name": self._name,
+        }
+
+        return await self.async_step_confirm()
+
+    async def _get_lektrico_device_settings_and_treat_unique_id(self) -> None:
+        """Get device's serial number from a Lektrico device."""
+        device = Device(
+            _host=self._host,
+            asyncClient=get_async_client(self.hass),
+        )
+
+        settings = await device.device_config()
+        self._serial_number = str(settings["serial_number"])
+        self._device_type = settings["type"]
+        self._board_revision = settings["board_revision"]
+        self._name = f"{settings["type"]}_{self._serial_number}"
+
+        # Check if already configured
+        # Set unique id
+        await self.async_set_unique_id(self._serial_number, raise_on_progress=True)
+        # Abort if already configured, but update the last-known host
+        self._abort_if_unique_id_configured(
+            updates={CONF_HOST: self._host}, reload_on_update=True
+        )
+
+    async def async_step_confirm(
+        self, user_input: dict[str, str] | None = None
+    ) -> ConfigFlowResult:
+        """Allow the user to confirm adding the device."""
+
+        if user_input is not None:
+            return self._async_create_entry()
+
+        self._set_confirm_only()
+        return self.async_show_form(step_id="confirm")

homeassistant/components/lektrico/const.py

@@ -0,0 +1,9 @@
+"""Constants for the Lektrico Charging Station integration."""
+
+from logging import Logger, getLogger
+
+# Integration domain
+DOMAIN = "lektrico"
+
+# Logger
+LOGGER: Logger = getLogger(__package__)

homeassistant/components/lektrico/coordinator.py

@@ -0,0 +1,52 @@
+"""Coordinator for the Lektrico Charging Station integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import Any
+
+from lektricowifi import Device, DeviceConnectionError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    ATTR_HW_VERSION,
+    ATTR_SERIAL_NUMBER,
+    CONF_HOST,
+    CONF_TYPE,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.httpx_client import get_async_client
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import LOGGER
+
+SCAN_INTERVAL = timedelta(seconds=10)
+
+
+class LektricoDeviceDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
+    """Data update coordinator for Lektrico device."""
+
+    config_entry: ConfigEntry
+
+    def __init__(self, hass: HomeAssistant, device_name: str) -> None:
+        """Initialize a Lektrico Device."""
+        super().__init__(
+            hass,
+            LOGGER,
+            name=device_name,
+            update_interval=SCAN_INTERVAL,
+        )
+        self.device = Device(
+            self.config_entry.data[CONF_HOST],
+            asyncClient=get_async_client(hass),
+        )
+        self.serial_number: str = self.config_entry.data[ATTR_SERIAL_NUMBER]
+        self.board_revision: str = self.config_entry.data[ATTR_HW_VERSION]
+        self.device_type: str = self.config_entry.data[CONF_TYPE]
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Async Update device state."""
+        try:
+            return await self.device.device_info(self.device_type)
+        except DeviceConnectionError as lek_ex:
+            raise UpdateFailed(lek_ex) from lek_ex

homeassistant/components/lektrico/entity.py

@@ -0,0 +1,33 @@
+"""Entity classes for the Lektrico integration."""
+
+from __future__ import annotations
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import LektricoDeviceDataUpdateCoordinator
+from .const import DOMAIN
+
+
+class LektricoEntity(CoordinatorEntity[LektricoDeviceDataUpdateCoordinator]):
+    """Define an Lektrico entity."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: LektricoDeviceDataUpdateCoordinator,
+        device_name: str,
+    ) -> None:
+        """Initialize."""
+        super().__init__(coordinator)
+
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, coordinator.serial_number)},
+            model=coordinator.device_type.upper(),
+            name=device_name,
+            manufacturer="Lektrico",
+            sw_version=coordinator.data["fw_version"],
+            hw_version=coordinator.board_revision,
+            serial_number=coordinator.serial_number,
+        )

homeassistant/components/lektrico/manifest.json

@@ -0,0 +1,16 @@
+{
+  "domain": "lektrico",
+  "name": "Lektrico Charging Station",
+  "codeowners": ["@lektrico"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/lektrico",
+  "integration_type": "device",
+  "iot_class": "local_polling",
+  "requirements": ["lektricowifi==0.0.41"],
+  "zeroconf": [
+    {
+      "type": "_http._tcp.local.",
+      "name": "lektrico*"
+    }
+  ]
+}