From a489ceebb69b7e26e09d351322898a492f7e782c Mon Sep 17 00:00:00 2001 From: petretiandrea Date: Mon, 8 Apr 2024 22:23:52 +0200 Subject: [PATCH] feat(core): update entity for check device firmware --- custom_components/tapo/const.py | 1 + custom_components/tapo/coordinators.py | 35 ++----- custom_components/tapo/helpers.py | 16 +++- custom_components/tapo/update.py | 128 +++++++++++++++++++++++++ 4 files changed, 153 insertions(+), 27 deletions(-) create mode 100644 custom_components/tapo/update.py diff --git a/custom_components/tapo/const.py b/custom_components/tapo/const.py index ff35ae1..5e139f4 100755 --- a/custom_components/tapo/const.py +++ b/custom_components/tapo/const.py @@ -70,6 +70,7 @@ Platform.SIREN, Platform.CLIMATE, Platform.NUMBER, + Platform.UPDATE, ] diff --git a/custom_components/tapo/coordinators.py b/custom_components/tapo/coordinators.py index fd27777..2f7783e 100644 --- a/custom_components/tapo/coordinators.py +++ b/custom_components/tapo/coordinators.py @@ -1,11 +1,9 @@ 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 @@ -13,24 +11,16 @@ 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] diff --git a/custom_components/tapo/helpers.py b/custom_components/tapo/helpers.py index f889c11..fbcc1df 100644 --- a/custom_components/tapo/helpers.py +++ b/custom_components/tapo/helpers.py @@ -1,6 +1,9 @@ +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, ) @@ -8,6 +11,7 @@ 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 diff --git a/custom_components/tapo/update.py b/custom_components/tapo/update.py new file mode 100644 index 0000000..aed7888 --- /dev/null +++ b/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__)