coordinator.py

Go to the documentation of this file.

"""YoLink DataUpdateCoordinator."""
 
from __future__ import annotations
 
import asyncio
from datetime import UTC, datetime, timedelta
import logging
 
from yolink.device import YoLinkDevice
from yolink.exception import YoLinkAuthFailError, YoLinkClientError
 
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import ATTR_DEVICE_STATE, DOMAIN, YOLINK_OFFLINE_TIME
 
_LOGGER = logging.getLogger(__name__)
 
 
class YoLinkCoordinator(DataUpdateCoordinator[dict]):
    """YoLink DataUpdateCoordinator."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        device: YoLinkDevice,
        paired_device: YoLinkDevice | None = None,
    ) -> None:
        """Init YoLink DataUpdateCoordinator.
 
        fetch state every 30 minutes base on yolink device heartbeat interval
        data is None before the first successful update, but we need to use
        data at first update
        """
        super().__init__(
            hass, _LOGGER, name=DOMAIN, update_interval=timedelta(minutes=30)
        )
        self.devicedevice = device
        self.paired_devicepaired_device = paired_device
        self.dev_onlinedev_online = True
 
    async def _async_update_data(self) -> dict:
        """Fetch device state."""
        try:
            async with asyncio.timeout(10):
                device_state_resp = await self.devicedevice.fetch_state()
                device_state = device_state_resp.data.get(ATTR_DEVICE_STATE)
                device_reporttime = device_state_resp.data.get("reportAt")
                if device_reporttime is not None:
                    rpt_time_delta = (
                        datetime.now(tz=UTC).replace(tzinfo=None)
                        - datetime.strptime(device_reporttime, "%Y-%m-%dT%H:%M:%S.%fZ")
                    ).total_seconds()
                    self.dev_onlinedev_online = rpt_time_delta < YOLINK_OFFLINE_TIME
                if self.paired_devicepaired_device is not None and device_state is not None:
                    paried_device_state_resp = await self.paired_devicepaired_device.fetch_state()
                    paried_device_state = paried_device_state_resp.data.get(
                        ATTR_DEVICE_STATE
                    )
                    if (
                        paried_device_state is not None
                        and ATTR_DEVICE_STATE in paried_device_state
                    ):
                        device_state[ATTR_DEVICE_STATE] = paried_device_state[
                            ATTR_DEVICE_STATE
                        ]
        except YoLinkAuthFailError as yl_auth_err:
            raise ConfigEntryAuthFailed from yl_auth_err
        except YoLinkClientError as yl_client_err:
            raise UpdateFailed from yl_client_err
        if device_state is not None:
            return device_state
        return {}