coordinator.py

Go to the documentation of this file.

"""Define an object to manage fetching AirGradient data."""
 
from __future__ import annotations
 
from dataclasses import dataclass
from datetime import timedelta
from typing import TYPE_CHECKING
 
from airgradient import AirGradientClient, AirGradientError, Config, Measures
 
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import DOMAIN, LOGGER
 
if TYPE_CHECKING:
    from . import AirGradientConfigEntry
 
 
@dataclass
class AirGradientData:
    """Class for AirGradient data."""
 
    measures: Measures
    config: Config
 
 
class AirGradientCoordinator(DataUpdateCoordinator[AirGradientData]):
    """Class to manage fetching AirGradient data."""
 
    config_entry: AirGradientConfigEntry
    _current_version: str
 
    def __init__(self, hass: HomeAssistant, client: AirGradientClient) -> None:
        """Initialize coordinator."""
        super().__init__(
            hass,
            logger=LOGGER,
            name=f"AirGradient {client.host}",
            update_interval=timedelta(minutes=1),
        )
        self.clientclient = client
        assert self.config_entryconfig_entry.unique_id
        self.serial_numberserial_number = self.config_entryconfig_entry.unique_id
 
    async def _async_setup(self) -> None:
        """Set up the coordinator."""
        self._current_version_current_version = (
            await self.clientclient.get_current_measures()
        ).firmware_version
 
    async def _async_update_data(self) -> AirGradientData:
        try:
            measures = await self.clientclient.get_current_measures()
            config = await self.clientclient.get_config()
        except AirGradientError as error:
            raise UpdateFailed(error) from error
        if measures.firmware_version != self._current_version_current_version:
            device_registry = dr.async_get(self.hasshass)
            device_entry = device_registry.async_get_device(
                identifiers={(DOMAIN, self.serial_numberserial_number)}
            )
            assert device_entry
            device_registry.async_update_device(
                device_entry.id,
                sw_version=measures.firmware_version,
            )
            self._current_version_current_version = measures.firmware_version
        return AirGradientData(measures, config)