api_data.py

Go to the documentation of this file.

"""Data for all Kaiterra devices."""
 
import asyncio
from logging import getLogger
 
from aiohttp.client_exceptions import ClientConnectorError, ClientResponseError
from kaiterra_async_client import AQIStandard, KaiterraAPIClient, Units
 
from homeassistant.const import CONF_API_KEY, CONF_DEVICE_ID, CONF_DEVICES, CONF_TYPE
from homeassistant.helpers.dispatcher import async_dispatcher_send
 
from .const import (
    AQI_LEVEL,
    AQI_SCALE,
    CONF_AQI_STANDARD,
    CONF_PREFERRED_UNITS,
    DISPATCHER_KAITERRA,
)
 
_LOGGER = getLogger(__name__)
 
POLLUTANTS = {"rpm25c": "PM2.5", "rpm10c": "PM10", "rtvoc": "TVOC", "rco2": "CO2"}
 
 
class KaiterraApiData:
    """Get data from Kaiterra API."""
 
    def __init__(self, hass, config, session):
        """Initialize the API data object."""
 
        api_key = config[CONF_API_KEY]
        aqi_standard = config[CONF_AQI_STANDARD]
        devices = config[CONF_DEVICES]
        units = config[CONF_PREFERRED_UNITS]
 
        self._hass_hass = hass
        self._api_api = KaiterraAPIClient(
            session,
            api_key=api_key,
            aqi_standard=AQIStandard.from_str(aqi_standard),
            preferred_units=[Units.from_str(unit) for unit in units],
        )
        self._devices_ids_devices_ids = [device[CONF_DEVICE_ID] for device in devices]
        self._devices_devices = [
            f"/{device[CONF_TYPE]}s/{device[CONF_DEVICE_ID]}" for device in devices
        ]
        self._scale_scale = AQI_SCALE[aqi_standard]
        self._level_level = AQI_LEVEL[aqi_standard]
        self._update_listeners_update_listeners = []
        self.datadata = {}
 
    async def async_update(self) -> None:
        """Get the data from Kaiterra API."""
 
        try:
            async with asyncio.timeout(10):
                data = await self._api_api.get_latest_sensor_readings(self._devices_devices)
        except (ClientResponseError, ClientConnectorError, TimeoutError) as err:
            _LOGGER.debug("Couldn't fetch data from Kaiterra API: %s", err)
            self.datadata = {}
            async_dispatcher_send(self._hass_hass, DISPATCHER_KAITERRA)
            return
 
        _LOGGER.debug("New data retrieved: %s", data)
 
        try:
            self.datadata = {}
            for i, device in enumerate(data):
                if not device:
                    self.datadata[self._devices_ids_devices_ids[i]] = {}
                    continue
 
                aqi, main_pollutant = None, None
                for sensor_name, sensor in device.items():
                    if not (points := sensor.get("points")):
                        continue
 
                    point = points[0]
                    sensor["value"] = point.get("value")
 
                    if "aqi" not in point:
                        continue
 
                    sensor["aqi"] = point["aqi"]
                    if not aqi or aqi < point["aqi"]:
                        aqi = point["aqi"]
                        main_pollutant = POLLUTANTS.get(sensor_name)
 
                level = None
                if aqi is not None:
                    for j in range(1, len(self._scale_scale)):
                        if aqi <= self._scale_scale[j]:
                            level = self._level_level[j - 1]
                            break
 
                device["aqi"] = {"value": aqi}
                device["aqi_level"] = {"value": level}
                device["aqi_pollutant"] = {"value": main_pollutant}
 
                self.datadata[self._devices_ids_devices_ids[i]] = device
        except IndexError as err:
            _LOGGER.error("Parsing error %s", err)
        except TypeError as err:
            _LOGGER.error("Type error %s", err)
 
        async_dispatcher_send(self._hass_hass, DISPATCHER_KAITERRA)