coordinator.py

Go to the documentation of this file.

"""Helper and wrapper classes for Gree module."""
 
from __future__ import annotations
 
from datetime import datetime, timedelta
import logging
from typing import Any
 
from greeclimate.device import Device, DeviceInfo
from greeclimate.discovery import Discovery, Listener
from greeclimate.exceptions import DeviceNotBoundError, DeviceTimeoutError
from greeclimate.network import Response
 
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.json import json_dumps
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util.dt import utcnow
 
from .const import (
    COORDINATORS,
    DISCOVERY_TIMEOUT,
    DISPATCH_DEVICE_DISCOVERED,
    DOMAIN,
    MAX_ERRORS,
    UPDATE_INTERVAL,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
class DeviceDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
    """Manages polling for state changes from the device."""
 
    def __init__(self, hass: HomeAssistant, device: Device) -> None:
        """Initialize the data update coordinator."""
        DataUpdateCoordinator.__init__(
            self,
            hass,
            _LOGGER,
            name=f"{DOMAIN}-{device.device_info.name}",
            update_interval=timedelta(seconds=UPDATE_INTERVAL),
            always_update=False,
        )
        self.devicedevice = device
        self.devicedevice.add_handler(Response.DATA, self.device_state_updateddevice_state_updated)
        self.devicedevice.add_handler(Response.RESULT, self.device_state_updateddevice_state_updated)
 
        self._error_count_error_count: int = 0
        self._last_response_time_last_response_time: datetime = utcnow()
        self._last_error_time_last_error_time: datetime | None = None
 
    def device_state_updated(self, *args: Any) -> None:
        """Handle device state updates."""
        _LOGGER.debug("Device state updated: %s", json_dumps(args))
        self._error_count_error_count = 0
        self._last_response_time_last_response_time = utcnow()
        self.async_set_updated_dataasync_set_updated_data(self.devicedevice.raw_properties)
 
    async def _async_update_data(self) -> dict[str, Any]:
        """Update the state of the device."""
        _LOGGER.debug(
            "Updating device state: %s, error count: %d", self.namename, self._error_count_error_count
        )
        try:
            await self.devicedevice.update_state()
        except DeviceNotBoundError as error:
            raise UpdateFailed(
                f"Device {self.name} is unavailable, device is not bound."
            ) from error
        except DeviceTimeoutError as error:
            self._error_count_error_count += 1
 
            # Under normal conditions GREE units timeout every once in a while
            if self.last_update_successlast_update_success and self._error_count_error_count >= MAX_ERRORS:
                _LOGGER.warning(
                    "Device %s is unavailable: %s", self.namename, self.devicedevice.device_info
                )
                raise UpdateFailed(
                    f"Device {self.name} is unavailable, could not send update request"
                ) from error
        else:
            # raise update failed if time for more than MAX_ERRORS has passed since last update
            now = utcnow()
            elapsed_success = now - self._last_response_time_last_response_time
            if self.update_intervalupdate_intervalupdate_intervalupdate_interval and elapsed_success >= self.update_intervalupdate_intervalupdate_intervalupdate_interval:
                if not self._last_error_time_last_error_time or (
                    (now - self.update_intervalupdate_intervalupdate_intervalupdate_interval) >= self._last_error_time_last_error_time
                ):
                    self._last_error_time_last_error_time = now
                    self._error_count_error_count += 1
 
                _LOGGER.warning(
                    "Device %s is unresponsive for %s seconds",
                    self.namename,
                    elapsed_success,
                )
            if self.last_update_successlast_update_success and self._error_count_error_count >= MAX_ERRORS:
                raise UpdateFailed(
                    f"Device {self.name} is unresponsive for too long and now unavailable"
                )
 
        return self.devicedevice.raw_properties
 
    async def push_state_update(self):
        """Send state updates to the physical device."""
        try:
            return await self.devicedevice.push_state_update()
        except DeviceTimeoutError:
            _LOGGER.warning(
                "Timeout send state update to: %s (%s)",
                self.namename,
                self.devicedevice.device_info,
            )
 
 
class DiscoveryService(Listener):
    """Discovery event handler for gree devices."""
 
    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize discovery service."""
        super().__init__()
        self.hasshass = hass
 
        self.discoverydiscovery = Discovery(DISCOVERY_TIMEOUT)
        self.discoverydiscovery.add_listener(self)
 
        hass.data[DOMAIN].setdefault(COORDINATORS, [])
 
    async def device_found(self, device_info: DeviceInfo) -> None:
        """Handle new device found on the network."""
 
        device = Device(device_info)
        try:
            await device.bind()
        except DeviceNotBoundError:
            _LOGGER.error("Unable to bind to gree device: %s", device_info)
        except DeviceTimeoutError:
            _LOGGER.error("Timeout trying to bind to gree device: %s", device_info)
 
        _LOGGER.debug(
            "Adding Gree device %s at %s:%i",
            device.device_info.name,
            device.device_info.ip,
            device.device_info.port,
        )
        coordo = DeviceDataUpdateCoordinator(self.hasshass, device)
        self.hasshass.data[DOMAIN][COORDINATORS].append(coordo)
        await coordo.async_refresh()
 
        async_dispatcher_send(self.hasshass, DISPATCH_DEVICE_DISCOVERED, coordo)
 
    async def device_update(self, device_info: DeviceInfo) -> None:
        """Handle updates in device information, update if ip has changed."""
        for coordinator in self.hasshass.data[DOMAIN][COORDINATORS]:
            if coordinator.device.device_info.mac == device_info.mac:
                coordinator.device.device_info.ip = device_info.ip
                await coordinator.async_refresh()