feat(core): update entity for check device firmware

custom_components/tapo/const.py

@@ -70,6 +70,7 @@
     Platform.SIREN,
     Platform.CLIMATE,
     Platform.NUMBER,
+    Platform.UPDATE,
 ]
 
 

custom_components/tapo/coordinators.py

@@ -1,36 +1,26 @@
 import logging
 from abc import ABC
-from abc import abstractmethod
 from dataclasses import dataclass
 from datetime import timedelta
 from typing import Dict
 from typing import List
-from typing import Optional
 from typing import Type
 from typing import TypeVar
 
 import aiohttp
 import async_timeout
 from homeassistant.core import CALLBACK_TYPE
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import ConfigEntryAuthFailed
 from homeassistant.helpers.debounce import Debouncer
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 from homeassistant.helpers.update_coordinator import UpdateFailed
-from plugp100.api.tapo_client import TapoClient
 from plugp100.new.tapodevice import TapoDevice
 from plugp100.new.tapohub import TapoHub
 from plugp100.responses.child_device_list import PowerStripChild
-from plugp100.responses.components import Components
-from plugp100.responses.tapo_exception import TapoError
 from plugp100.responses.tapo_exception import TapoException
 
 from custom_components.tapo.const import DOMAIN
-from custom_components.tapo.const import SUPPORTED_DEVICE_AS_LED_STRIP
-from custom_components.tapo.const import SUPPORTED_DEVICE_AS_LIGHT
-from custom_components.tapo.const import SUPPORTED_DEVICE_AS_SWITCH
-from custom_components.tapo.const import SUPPORTED_HUB_DEVICE_MODEL
-from custom_components.tapo.const import SUPPORTED_POWER_STRIP_DEVICE_MODEL
+from custom_components.tapo.helpers import _raise_from_tapo_exception
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -66,10 +56,10 @@ class HassTapoDeviceData:
 
 class TapoDataCoordinator(ABC, DataUpdateCoordinator[StateMap]):
     def __init__(
-        self,
-        hass: HomeAssistant,
-        device: TapoDevice,
-        polling_interval: timedelta,
+            self,
+            hass: HomeAssistant,
+            device: TapoDevice,
+            polling_interval: timedelta,
     ):
         self._device = device
         super().__init__(
@@ -95,21 +85,16 @@ def is_hub(self) -> bool:
     async def _async_update_data(self) -> StateMap:
         try:
             async with async_timeout.timeout(10):
-                return await self.device.update()
+                return await self.poll_update()
         except TapoException as error:
-            _raise_from_tapo_exception(error)
+            _raise_from_tapo_exception(error, _LOGGER)
         except aiohttp.ClientError as error:
             raise UpdateFailed(f"Error communication with API: {str(error)}") from error
         except Exception as exception:
             raise UpdateFailed(f"Unexpected exception: {str(exception)}") from exception
 
-
-PowerStripChildrenState = dict[str, PowerStripChild]
+    async def poll_update(self):
+        return await self.device.update()
 
 
-def _raise_from_tapo_exception(exception: TapoException):
-    _LOGGER.error("Tapo exception: %s", str(exception))
-    if exception.error_code == TapoError.INVALID_CREDENTIAL.value:
-        raise ConfigEntryAuthFailed from exception
-    else:
-        raise UpdateFailed(f"Error tapo exception: {exception}") from exception
+PowerStripChildrenState = dict[str, PowerStripChild]

custom_components/tapo/helpers.py

@@ -1,13 +1,17 @@
+from logging import Logger
 from typing import Optional
 from typing import TypeVar
 
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.update_coordinator import UpdateFailed
 from homeassistant.util.color import (
     color_temperature_kelvin_to_mired as kelvin_to_mired,
 )
 from homeassistant.util.color import (
     color_temperature_mired_to_kelvin as mired_to_kelvin,
 )
 from plugp100.common.functional.tri import Try
+from plugp100.responses.tapo_exception import TapoException, TapoError
 
 T = TypeVar("T")
 
@@ -38,7 +42,7 @@ def tapo_to_hass_brightness(brightness: float | None) -> float | None:
 
 # Mireds and Kelving are min, max tuple
 def hass_to_tapo_color_temperature(
-    color_temp: int | None, mireds: (int, int), kelvin: (int, int)
+        color_temp: int | None, mireds: (int, int), kelvin: (int, int)
 ) -> int | None:
     if color_temp is not None:
         constraint_color_temp = clamp(color_temp, mireds[0], mireds[1])
@@ -51,7 +55,7 @@ def hass_to_tapo_color_temperature(
 
 
 def tapo_to_hass_color_temperature(
-    color_temp: int | None, mireds: (int, int)
+        color_temp: int | None, mireds: (int, int)
 ) -> int | None:
     if color_temp is not None and color_temp > 0:
         return clamp(
@@ -60,3 +64,11 @@ def tapo_to_hass_color_temperature(
             max_value=mireds[1],
         )
     return None
+
+
+def _raise_from_tapo_exception(exception: TapoException, logger: Logger):
+    logger.error("Tapo exception: %s", str(exception))
+    if exception.error_code == TapoError.INVALID_CREDENTIAL.value:
+        raise ConfigEntryAuthFailed from exception
+    else:
+        raise UpdateFailed(f"Error tapo exception: {exception}") from exception

custom_components/tapo/update.py

@@ -0,0 +1,128 @@
+import logging
+from datetime import timedelta
+from typing import cast, Any, Optional
+
+from homeassistant.components.update import UpdateEntity, UpdateEntityFeature, UpdateDeviceClass
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from plugp100.new.tapodevice import TapoDevice
+from plugp100.responses.firmware import LatestFirmware, FirmwareDownloadProgress, FirmwareDownloadStatus
+from plugp100.responses.tapo_exception import TapoException
+
+from custom_components.tapo import DOMAIN, HassTapoDeviceData
+from custom_components.tapo.coordinators import TapoDataCoordinator
+from custom_components.tapo.entity import CoordinatedTapoEntity
+
+POLL_DELAY_IDLE = timedelta(seconds=6 * 60 * 60)
+POLL_DELAY_UPGRADE = timedelta(seconds=60)
+
+
+async def async_setup_entry(
+        hass: HomeAssistant,
+        entry: ConfigEntry,
+        async_add_entities: AddEntitiesCallback,
+) -> None:
+    data = cast(HassTapoDeviceData, hass.data[DOMAIN][entry.entry_id])
+    if data.coordinator.is_hub:
+        coordinators = [
+            TapoDeviceFirmwareDataCoordinator(hass, coordinator.device, POLL_DELAY_IDLE) \
+            for coordinator in data.child_coordinators
+        ]
+    else:
+        coordinators = [TapoDeviceFirmwareDataCoordinator(hass, data.coordinator.device, POLL_DELAY_IDLE)]
+
+    async_add_entities([
+        TapoDeviceFirmwareEntity(coordinator, coordinator.device) for coordinator in coordinators
+    ], True)
+
+
+class TapoDeviceFirmwareDataCoordinator(TapoDataCoordinator):
+
+    def __init__(self, hass: HomeAssistant, device: TapoDevice, polling_interval: timedelta):
+        super().__init__(hass, device, polling_interval)
+        self._latest_firmware: Optional[LatestFirmware] = None
+        self._download_status: Optional[FirmwareDownloadProgress] = None
+
+    @property
+    def latest_firmware(self) -> Optional[LatestFirmware]:
+        return self._latest_firmware
+
+    @property
+    def download_progress(self) -> Optional[FirmwareDownloadProgress]:
+        return self._download_status
+
+    async def poll_update(self):
+        self._latest_firmware = (await self.device.get_latest_firmware()).get_or_raise()
+        self._download_status = (await self.device.get_firmware_download_state()).get_or_raise()
+        if self._download_status.status == FirmwareDownloadStatus.DOWNLOADING or \
+                self._download_status == FirmwareDownloadStatus.PREPARING:
+            self.update_interval = POLL_DELAY_UPGRADE
+        else:
+            self.update_interval = POLL_DELAY_IDLE
+        return self._latest_firmware
+
+
+class TapoDeviceFirmwareEntity(CoordinatedTapoEntity, UpdateEntity):
+    _attr_has_entity_name = True
+    _attr_supported_features = (
+            UpdateEntityFeature.INSTALL
+            | UpdateEntityFeature.PROGRESS
+            | UpdateEntityFeature.RELEASE_NOTES
+    )
+    _attr_device_class = UpdateDeviceClass.FIRMWARE
+
+    coordinator: TapoDeviceFirmwareDataCoordinator
+
+    def __init__(self, coordinator: TapoDeviceFirmwareDataCoordinator, device: TapoDevice):
+        super().__init__(coordinator, device)
+        self._attr_name = "Firmware"
+
+    def release_notes(self) -> str | None:
+        """Get the release notes for the latest update."""
+        status = self.coordinator.latest_firmware
+        if status.need_to_upgrade:
+            return status.release_note
+        return None
+
+    async def async_install(
+            self, version: str | None, backup: bool, **kwargs: Any
+    ) -> None:
+        """Install a firmware update."""
+        try:
+            result = (await self.device.start_firmware_upgrade())
+        except TapoException as ex:
+            raise HomeAssistantError("Unable to send Firmware update request. Check the controller is online.") from ex
+        except Exception as ex:
+            raise HomeAssistantError("Firmware update request rejected") from ex
+        finally:
+            await self.coordinator.async_request_refresh()
+
+        if not result:
+            raise HomeAssistantError(
+                "Unable to send Firmware update request. Check the controller is online.")
+
+    @property
+    def installed_version(self) -> str | None:
+        return self.device.firmware_version
+
+    @property
+    def latest_version(self) -> str | None:
+        status = self.coordinator.latest_firmware
+        return status.firmware_version \
+            if status.firmware_version and status.need_to_upgrade \
+            else self.device.firmware_version
+
+    @property
+    def in_progress(self) -> bool | int | None:
+        download_progress = self.coordinator.download_progress
+        return download_progress.download_in_progress if download_progress else 0
+
+    @property
+    def auto_update(self) -> bool:
+        download_progress = self.coordinator.download_progress
+        return download_progress.auto_upgrade if download_progress else False
+
+
+_LOGGER = logging.getLogger(__name__)