Add new integration for AiDot

homeassistant/components/aidot/__init__.py

@@ -0,0 +1,26 @@
+"""The aidot integration."""
+
+from __future__ import annotations
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import AidotConfigEntry, AidotDeviceManagerCoordinator
+
+PLATFORMS: list[Platform] = [Platform.LIGHT]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: AidotConfigEntry) -> bool:
+    """Set up aidot from a config entry."""
+
+    coordinator = AidotDeviceManagerCoordinator(hass, entry)
+    await coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: AidotConfigEntry) -> bool:
+    """Unload a config entry."""
+    await entry.runtime_data.async_cleanup()
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/aidot/config_flow.py

@@ -0,0 +1,67 @@
+"""Config flow for Aidot integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from aidot.client import AidotClient
+from aidot.const import CONF_LOGIN_INFO, DEFAULT_COUNTRY_CODE, SUPPORTED_COUNTRY_CODES
+from aidot.exceptions import AidotUserOrPassIncorrect
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_COUNTRY_CODE, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.helpers import selector
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(
+            CONF_COUNTRY_CODE,
+            default=DEFAULT_COUNTRY_CODE,
+        ): selector.CountrySelector(
+            selector.CountrySelectorConfig(
+                countries=SUPPORTED_COUNTRY_CODES,
+            )
+        ),
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+class AidotConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle aidot config flow."""
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            client = AidotClient(
+                session=async_get_clientsession(self.hass),
+                country_code=user_input[CONF_COUNTRY_CODE],
+                username=user_input[CONF_USERNAME],
+                password=user_input[CONF_PASSWORD],
+            )
+            await self.async_set_unique_id(client.get_identifier())
+            self._abort_if_unique_id_configured()
+            try:
+                login_info = await client.async_post_login()
+            except AidotUserOrPassIncorrect:
+                errors["base"] = "invalid_auth"
+
+            if not errors:
+                return self.async_create_entry(
+                    title=f"{user_input[CONF_USERNAME]} {user_input[CONF_COUNTRY_CODE]}",
+                    data={
+                        CONF_LOGIN_INFO: login_info,
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+        )

homeassistant/components/aidot/const.py

@@ -0,0 +1,3 @@
+"""Constants for the aidot integration."""
+
+DOMAIN = "aidot"

homeassistant/components/aidot/coordinator.py

@@ -0,0 +1,162 @@
+"""Coordinator for Aidot."""
+
+from datetime import timedelta
+import logging
+
+from aidot.client import AidotClient
+from aidot.const import (
+    CONF_ACCESS_TOKEN,
+    CONF_AES_KEY,
+    CONF_DEVICE_LIST,
+    CONF_ID,
+    CONF_LOGIN_INFO,
+    CONF_TYPE,
+)
+from aidot.device_client import DeviceClient, DeviceStatusData
+from aidot.exceptions import AidotAuthFailed, AidotUserOrPassIncorrect
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryError
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN
+
+type AidotConfigEntry = ConfigEntry[AidotDeviceManagerCoordinator]
+_LOGGER = logging.getLogger(__name__)
+
+UPDATE_DEVICE_LIST_INTERVAL = timedelta(hours=6)
+
+
+class AidotDeviceUpdateCoordinator(DataUpdateCoordinator[DeviceStatusData]):
+    """Class to manage Aidot data."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: AidotConfigEntry,
+        device_client: DeviceClient,
+    ) -> None:
+        """Initialize coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=config_entry,
+            name=DOMAIN,
+            update_interval=None,
+        )
+        self.device_client = device_client
+
+    async def _async_setup(self) -> None:
+        """Set up the coordinator."""
+        self.device_client.on_status_update = self._handle_status_update
+
+    def _handle_status_update(self, status: DeviceStatusData) -> None:
+        """Handle status callback."""
+        self.async_set_updated_data(status)
+
+    async def _async_update_data(self) -> DeviceStatusData:
+        """Return current status."""
+        return self.device_client.status
+
+
+class AidotDeviceManagerCoordinator(DataUpdateCoordinator[None]):
+    """Class to manage fetching Aidot data."""
+
+    config_entry: AidotConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: AidotConfigEntry,
+    ) -> None:
+        """Initialize coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=config_entry,
+            name=DOMAIN,
+            update_interval=UPDATE_DEVICE_LIST_INTERVAL,
+        )
+        self.client = AidotClient(
+            session=async_get_clientsession(hass),
+            token=config_entry.data[CONF_LOGIN_INFO],
+        )
+        self.client.set_token_fresh_cb(self.token_fresh_cb)
+        self.device_coordinators: dict[str, AidotDeviceUpdateCoordinator] = {}
+
+    async def _async_setup(self) -> None:
+        """Set up the coordinator."""
+        try:
+            await self.async_auto_login()
+        except AidotUserOrPassIncorrect as error:
+            raise ConfigEntryAuthFailed from error
+
+    async def _async_update_data(self) -> None:
+        """Update data async."""
+        try:
+            data = await self.client.async_get_all_device()
+        except AidotAuthFailed as error:
+            self.token_fresh_cb()
+            raise ConfigEntryError from error
+        current_devices = {
+            device[CONF_ID]: device
+            for device in data[CONF_DEVICE_LIST]
+            if (
+                device[CONF_TYPE] == "light"
+                and CONF_AES_KEY in device
+                and device[CONF_AES_KEY][0] is not None
+            )
+        }
+
+        removed_ids = set(self.device_coordinators) - set(current_devices)
+        for dev_id in removed_ids:
+            del self.device_coordinators[dev_id]
+        if removed_ids:
+            self._purge_deleted_lists()
+
+        for dev_id, device in current_devices.items():
+            if dev_id not in self.device_coordinators:
+                device_client = self.client.get_device_client(device)
+                device_coordinator = AidotDeviceUpdateCoordinator(
+                    self.hass, self.config_entry, device_client
+                )
+                await device_coordinator.async_config_entry_first_refresh()
+                self.device_coordinators[dev_id] = device_coordinator
+
+    async def async_cleanup(self) -> None:
+        """Perform cleanup actions."""
+        await self.client.async_cleanup()
+
+    def token_fresh_cb(self) -> None:
+        """Update token."""
+        self.hass.config_entries.async_update_entry(
+            self.config_entry, data={CONF_LOGIN_INFO: self.client.login_info.copy()}
+        )
+
+    async def async_auto_login(self) -> None:
+        """Async auto login."""
+        if self.client.login_info.get(CONF_ACCESS_TOKEN) is None:
+            await self.client.async_post_login()
+
+    def _purge_deleted_lists(self) -> None:
+        """Purge device entries of deleted lists."""
+
+        device_reg = dr.async_get(self.hass)
+        identifiers = {
+            (
+                DOMAIN,
+                f"{device_coordinator.device_client.info.dev_id}",
+            )
+            for device_coordinator in self.device_coordinators.values()
+        }
+        for device in dr.async_entries_for_config_entry(
+            device_reg, self.config_entry.entry_id
+        ):
+            if not set(device.identifiers) & identifiers:
+                _LOGGER.debug("Removing obsolete device entry %s", device.name)
+                device_reg.async_update_device(
+                    device.id, remove_config_entry_id=self.config_entry.entry_id
+                )

homeassistant/components/aidot/light.py

@@ -0,0 +1,122 @@
+"""Support for Aidot lights."""
+
+from typing import Any
+
+from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
+    ATTR_COLOR_TEMP_KELVIN,
+    ATTR_RGBW_COLOR,
+    ColorMode,
+    LightEntity,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import (
+    CONNECTION_NETWORK_MAC,
+    DeviceInfo,
+    format_mac,
+)
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import AidotConfigEntry, AidotDeviceUpdateCoordinator
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: AidotConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Light."""
+    coordinator = entry.runtime_data
+    async_add_entities(
+        AidotLight(device_coordinator)
+        for device_coordinator in coordinator.device_coordinators.values()
+    )
+
+
+class AidotLight(CoordinatorEntity[AidotDeviceUpdateCoordinator], LightEntity):
+    """Representation of a Aidot Wi-Fi Light."""
+
+    _attr_has_entity_name = True
+    _attr_name = None
+
+    def __init__(self, coordinator: AidotDeviceUpdateCoordinator) -> None:
+        """Initialize the light."""
+        super().__init__(coordinator)
+        self._attr_unique_id = coordinator.device_client.info.dev_id
+        if hasattr(coordinator.device_client.info, "cct_max"):
+            self._attr_max_color_temp_kelvin = coordinator.device_client.info.cct_max
+        if hasattr(coordinator.device_client.info, "cct_min"):
+            self._attr_min_color_temp_kelvin = coordinator.device_client.info.cct_min
+
+        model_id = coordinator.device_client.info.model_id
+        manufacturer = model_id.split(".")[0]
+        model = model_id[len(manufacturer) + 1 :]
+        mac = format_mac(coordinator.device_client.info.mac)
+
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, self._attr_unique_id)},
+            connections={(CONNECTION_NETWORK_MAC, mac)},
+            manufacturer=manufacturer,
+            model=model,
+            name=coordinator.device_client.info.name,
+            hw_version=coordinator.device_client.info.hw_version,
+        )
+        if coordinator.device_client.info.enable_rgbw:
+            self._attr_color_mode = ColorMode.RGBW
+            self._attr_supported_color_modes = {ColorMode.RGBW, ColorMode.COLOR_TEMP}
+        elif coordinator.device_client.info.enable_cct:
+            self._attr_color_mode = ColorMode.COLOR_TEMP
+            self._attr_supported_color_modes = {ColorMode.COLOR_TEMP}
+        else:
+            self._attr_color_mode = ColorMode.BRIGHTNESS
+            self._attr_supported_color_modes = {ColorMode.BRIGHTNESS}
+        self._update_status()
+
+    def _update_status(self) -> None:
+        """Update light status from coordinator data."""
+        self._attr_available = self.coordinator.data.online
+        self._attr_is_on = self.coordinator.data.on
+        self._attr_brightness = self.coordinator.data.dimming
+        self._attr_color_temp_kelvin = self.coordinator.data.cct
+        self._attr_rgbw_color = self.coordinator.data.rgbw
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        return self.coordinator.data.online
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Update."""
+        self._update_status()
+        super()._handle_coordinator_update()
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Fix brightness state synchronization by updating the coordinator's `dimming` field."""
+        if ATTR_BRIGHTNESS in kwargs:
+            brightness = kwargs.get(ATTR_BRIGHTNESS, 255)
+            await self.coordinator.device_client.async_set_brightness(brightness)
+            self.coordinator.data.dimming = brightness
+        elif ATTR_COLOR_TEMP_KELVIN in kwargs:
+            color_temp_kelvin = kwargs.get(ATTR_COLOR_TEMP_KELVIN)
+            await self.coordinator.device_client.async_set_cct(color_temp_kelvin)
+            self.coordinator.data.cct = color_temp_kelvin
+            self._attr_color_mode = ColorMode.COLOR_TEMP
+        elif ATTR_RGBW_COLOR in kwargs:
+            rgbw_color = kwargs.get(ATTR_RGBW_COLOR)
+            await self.coordinator.device_client.async_set_rgbw(rgbw_color)
+            self.coordinator.data.rgbw = rgbw_color
+            self._attr_color_mode = ColorMode.RGBW
+        else:
+            await self.coordinator.device_client.async_turn_on()
+
+        self.coordinator.data.on = True
+        self._attr_is_on = True
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn the light off."""
+        await self.coordinator.device_client.async_turn_off()
+        self.coordinator.data.on = False
+        self._attr_is_on = False