coordinator.py

Go to the documentation of this file.

"""Coordinator for the venstar component."""
 
from __future__ import annotations
 
import asyncio
from datetime import timedelta
 
from requests import RequestException
from venstarcolortouch import VenstarColorTouch
 
from homeassistant.core import HomeAssistant
from homeassistant.helpers import update_coordinator
 
from .const import _LOGGER, DOMAIN, VENSTAR_SLEEP
 
 
class VenstarDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator[None]):
    """Class to manage fetching Venstar data."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        *,
        venstar_connection: VenstarColorTouch,
    ) -> None:
        """Initialize global Venstar data updater."""
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=60),
        )
        self.clientclient = venstar_connection
        self.runtimesruntimes: list[dict[str, int]] = []
 
    async def _async_update_data(self) -> None:
        """Update the state."""
        try:
            await self.hass.async_add_executor_job(self.clientclient.update_info)
        except (OSError, RequestException) as ex:
            raise update_coordinator.UpdateFailed(
                f"Exception during Venstar info update: {ex}"
            ) from ex
 
        # older venstars sometimes cannot handle rapid sequential connections
        await asyncio.sleep(VENSTAR_SLEEP)
 
        try:
            await self.hass.async_add_executor_job(self.clientclient.update_sensors)
        except (OSError, RequestException) as ex:
            raise update_coordinator.UpdateFailed(
                f"Exception during Venstar sensor update: {ex}"
            ) from ex
 
        # older venstars sometimes cannot handle rapid sequential connections
        await asyncio.sleep(VENSTAR_SLEEP)
 
        try:
            await self.hass.async_add_executor_job(self.clientclient.update_alerts)
        except (OSError, RequestException) as ex:
            raise update_coordinator.UpdateFailed(
                f"Exception during Venstar alert update: {ex}"
            ) from ex
 
        # older venstars sometimes cannot handle rapid sequential connections
        await asyncio.sleep(VENSTAR_SLEEP)
 
        try:
            self.runtimesruntimes = await self.hass.async_add_executor_job(
                self.clientclient.get_runtimes
            )
        except (OSError, RequestException) as ex:
            raise update_coordinator.UpdateFailed(
                f"Exception during Venstar runtime update: {ex}"
            ) from ex