coordinator.py

Go to the documentation of this file.

"""Data coordinator for WeatherFlow Cloud Data."""
 
from datetime import timedelta
 
from aiohttp import ClientResponseError
from weatherflow4py.api import WeatherFlowRestAPI
from weatherflow4py.models.rest.unified import WeatherFlowDataREST
 
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import DOMAIN, LOGGER
 
 
class WeatherFlowCloudDataUpdateCoordinator(
    DataUpdateCoordinator[dict[int, WeatherFlowDataREST]]
):
    """Class to manage fetching REST Based WeatherFlow Forecast data."""
 
    def __init__(self, hass: HomeAssistant, api_token: str) -> None:
        """Initialize global WeatherFlow forecast data updater."""
        self.weather_apiweather_api = WeatherFlowRestAPI(api_token=api_token)
        super().__init__(
            hass,
            LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=60),
        )
 
    async def _async_update_data(self) -> dict[int, WeatherFlowDataREST]:
        """Fetch data from WeatherFlow Forecast."""
        try:
            async with self.weather_apiweather_api:
                return await self.weather_apiweather_api.get_all_data()
        except ClientResponseError as err:
            if err.status == 401:
                raise ConfigEntryAuthFailed(err) from err
            raise UpdateFailed(f"Update failed: {err}") from err