device.py

Go to the documentation of this file.

"""Support for Broadlink devices."""
 
from contextlib import suppress
from functools import partial
import logging
from typing import Generic
 
import broadlink as blk
from broadlink.exceptions import (
    AuthenticationError,
    AuthorizationError,
    BroadlinkException,
    ConnectionClosedError,
    NetworkTimeoutError,
)
from typing_extensions import TypeVar
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CONF_HOST,
    CONF_MAC,
    CONF_NAME,
    CONF_TIMEOUT,
    CONF_TYPE,
    Platform,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
 
from .const import DEFAULT_PORT, DOMAIN, DOMAINS_AND_TYPES
from .updater import BroadlinkUpdateManager, get_update_manager
 
_ApiT = TypeVar("_ApiT", bound=blk.Device, default=blk.Device)
 
_LOGGER = logging.getLogger(__name__)
 
 
def get_domains(device_type: str) -> set[Platform]:
    """Return the domains available for a device type."""
    return {d for d, t in DOMAINS_AND_TYPES.items() if device_type in t}
 
 
class BroadlinkDevice(Generic[_ApiT]):
    """Manages a Broadlink device."""
 
    api: _ApiT
 
    def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None:
        """Initialize the device."""
        self.hasshass = hass
        self.configconfig = config
        self.update_managerupdate_manager: BroadlinkUpdateManager[_ApiT] | None = None
        self.fw_versionfw_version: int | None = None
        self.authorizedauthorized: bool | None = None
        self.reset_jobs: list[CALLBACK_TYPE] = []
 
    @property
    def name(self) -> str:
        """Return the name of the device."""
        return self.configconfig.title
 
    @property
    def unique_id(self) -> str | None:
        """Return the unique id of the device."""
        return self.configconfig.unique_id
 
    @property
    def mac_address(self) -> str:
        """Return the mac address of the device."""
        return self.configconfig.data[CONF_MAC]
 
    @property
    def available(self) -> bool | None:
        """Return True if the device is available."""
        if self.update_managerupdate_manager is None:
            return False
        return self.update_managerupdate_manager.available
 
    @staticmethod
    async def async_update(hass: HomeAssistant, entry: ConfigEntry) -> None:
        """Update the device and related entities.
 
        Triggered when the device is renamed on the frontend.
        """
        device_registry = dr.async_get(hass)
        assert entry.unique_id
        device_entry = device_registry.async_get_device(
            identifiers={(DOMAIN, entry.unique_id)}
        )
        assert device_entry
        device_registry.async_update_device(device_entry.id, name=entry.title)
        await hass.config_entries.async_reload(entry.entry_id)
 
    def _get_firmware_version(self) -> int | None:
        """Get firmware version."""
        self.apiapi.auth()
        with suppress(BroadlinkException, OSError):
            return self.apiapi.get_fwversion()
        return None
 
    async def async_setup(self) -> bool:
        """Set up the device and related entities."""
        config = self.configconfig
 
        api = blk.gendevice(
            config.data[CONF_TYPE],
            (config.data[CONF_HOST], DEFAULT_PORT),
            bytes.fromhex(config.data[CONF_MAC]),
            name=config.title,
        )
        api.timeout = config.data[CONF_TIMEOUT]
        self.apiapi = api
 
        try:
            self.fw_versionfw_version = await self.hasshass.async_add_executor_job(
                self._get_firmware_version_get_firmware_version
            )
 
        except AuthenticationError:
            await self._async_handle_auth_error_async_handle_auth_error()
            return False
 
        except (NetworkTimeoutError, OSError) as err:
            raise ConfigEntryNotReady from err
 
        except BroadlinkException as err:
            _LOGGER.error(
                "Failed to authenticate to the device at %s: %s", api.host[0], err
            )
            return False
 
        self.authorizedauthorized = True
 
        update_manager = get_update_manager(self)
        coordinator = update_manager.coordinator
        await coordinator.async_config_entry_first_refresh()
 
        self.update_managerupdate_manager = update_manager
        self.hasshass.data[DOMAIN].devices[config.entry_id] = self
        self.reset_jobs.append(config.add_update_listener(self.async_updateasync_update))
 
        # Forward entry setup to related domains.
        await self.hasshass.config_entries.async_forward_entry_setups(
            config, get_domains(self.apiapi.type)
        )
 
        return True
 
    async def async_unload(self) -> bool:
        """Unload the device and related entities."""
        if self.update_managerupdate_manager is None:
            return True
 
        while self.reset_jobs:
            self.reset_jobs.pop()()
 
        return await self.hasshass.config_entries.async_unload_platforms(
            self.configconfig, get_domains(self.apiapi.type)
        )
 
    async def async_auth(self) -> bool:
        """Authenticate to the device."""
        try:
            await self.hasshass.async_add_executor_job(self.apiapi.auth)
        except (BroadlinkException, OSError) as err:
            _LOGGER.debug(
                "Failed to authenticate to the device at %s: %s", self.apiapi.host[0], err
            )
            if isinstance(err, AuthenticationError):
                await self._async_handle_auth_error_async_handle_auth_error()
            return False
        return True
 
    async def async_request(self, function, *args, **kwargs):
        """Send a request to the device."""
        request = partial(function, *args, **kwargs)
        try:
            return await self.hasshass.async_add_executor_job(request)
        except (AuthorizationError, ConnectionClosedError):
            if not await self.async_authasync_auth():
                raise
            return await self.hasshass.async_add_executor_job(request)
 
    async def _async_handle_auth_error(self) -> None:
        """Handle an authentication error."""
        if self.authorizedauthorized is False:
            return
 
        self.authorizedauthorized = False
 
        _LOGGER.error(
            (
                "%s (%s at %s) is locked. Click Configuration in the sidebar, "
                "click Integrations, click Configure on the device and follow "
                "the instructions to unlock it"
            ),
            self.namename,
            self.apiapi.model,
            self.apiapi.host[0],
        )
 
        self.configconfig.async_start_reauth(self.hasshass, data={CONF_NAME: self.namename})