data_handler.py

Go to the documentation of this file.

"""The Netatmo data handler."""
 
from __future__ import annotations
 
from collections import deque
from dataclasses import dataclass
from datetime import datetime, timedelta
from itertools import islice
import logging
from time import time
from typing import Any
 
import aiohttp
import pyatmo
from pyatmo.modules.device_types import (
    DeviceCategory as NetatmoDeviceCategory,
    DeviceType as NetatmoDeviceType,
)
 
from homeassistant.components import cloud
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.helpers.dispatcher import (
    async_dispatcher_connect,
    async_dispatcher_send,
)
from homeassistant.helpers.event import async_track_time_interval
 
from .const import (
    AUTH,
    DATA_PERSONS,
    DATA_SCHEDULES,
    DOMAIN,
    MANUFACTURER,
    NETATMO_CREATE_BATTERY,
    NETATMO_CREATE_CAMERA,
    NETATMO_CREATE_CAMERA_LIGHT,
    NETATMO_CREATE_CLIMATE,
    NETATMO_CREATE_COVER,
    NETATMO_CREATE_FAN,
    NETATMO_CREATE_LIGHT,
    NETATMO_CREATE_ROOM_SENSOR,
    NETATMO_CREATE_SELECT,
    NETATMO_CREATE_SENSOR,
    NETATMO_CREATE_SWITCH,
    NETATMO_CREATE_WEATHER_SENSOR,
    PLATFORMS,
    WEBHOOK_ACTIVATION,
    WEBHOOK_DEACTIVATION,
    WEBHOOK_NACAMERA_CONNECTION,
    WEBHOOK_PUSH_TYPE,
)
 
_LOGGER = logging.getLogger(__name__)
 
SIGNAL_NAME = "signal_name"
ACCOUNT = "account"
HOME = "home"
WEATHER = "weather"
AIR_CARE = "air_care"
PUBLIC = NetatmoDeviceType.public
EVENT = "event"
 
PUBLISHERS = {
    ACCOUNT: "async_update_topology",
    HOME: "async_update_status",
    WEATHER: "async_update_weather_stations",
    AIR_CARE: "async_update_air_care",
    PUBLIC: "async_update_public_weather",
    EVENT: "async_update_events",
}
 
BATCH_SIZE = 3
DEV_FACTOR = 7
DEV_LIMIT = 400
CLOUD_FACTOR = 2
CLOUD_LIMIT = 150
DEFAULT_INTERVALS = {
    ACCOUNT: 10800,
    HOME: 300,
    WEATHER: 600,
    AIR_CARE: 300,
    PUBLIC: 600,
    EVENT: 600,
}
SCAN_INTERVAL = 60
 
 
@dataclass
class NetatmoDevice:
    """Netatmo device class."""
 
    data_handler: NetatmoDataHandler
    device: pyatmo.modules.Module
    parent_id: str
    signal_name: str
 
 
@dataclass
class NetatmoHome:
    """Netatmo home class."""
 
    data_handler: NetatmoDataHandler
    home: pyatmo.Home
    parent_id: str
    signal_name: str
 
 
@dataclass
class NetatmoRoom:
    """Netatmo room class."""
 
    data_handler: NetatmoDataHandler
    room: pyatmo.Room
    parent_id: str
    signal_name: str
 
 
@dataclass
class NetatmoPublisher:
    """Class for keeping track of Netatmo data class metadata."""
 
    name: str
    interval: int
    next_scan: float
    subscriptions: set[CALLBACK_TYPE | None]
    method: str
    kwargs: dict
 
 
class NetatmoDataHandler:
    """Manages the Netatmo data handling."""
 
    account: pyatmo.AsyncAccount
    _interval_factor: int
 
    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
        """Initialize self."""
        self.hasshass = hass
        self.config_entryconfig_entry = config_entry
        self._auth_auth = hass.data[DOMAIN][config_entry.entry_id][AUTH]
        self.publisher: dict[str, NetatmoPublisher] = {}
        self._queue: deque = deque()
        self._webhook_webhook: bool = False
        if config_entry.data["auth_implementation"] == cloud.DOMAIN:
            self._interval_factor_interval_factor = CLOUD_FACTOR
            self._rate_limit_rate_limit = CLOUD_LIMIT
        else:
            self._interval_factor_interval_factor = DEV_FACTOR
            self._rate_limit_rate_limit = DEV_LIMIT
        self.poll_startpoll_start = time()
        self.poll_countpoll_count = 0
 
    async def async_setup(self) -> None:
        """Set up the Netatmo data handler."""
        self.config_entryconfig_entry.async_on_unload(
            async_track_time_interval(
                self.hasshass, self.async_updateasync_update, timedelta(seconds=SCAN_INTERVAL)
            )
        )
 
        self.config_entryconfig_entry.async_on_unload(
            async_dispatcher_connect(
                self.hasshass,
                f"signal-{DOMAIN}-webhook-None",
                self.handle_eventhandle_event,
            )
        )
 
        self.accountaccount = pyatmo.AsyncAccount(self._auth_auth)
 
        await self.subscribesubscribe(ACCOUNT, ACCOUNT, None)
 
        await self.hasshass.config_entries.async_forward_entry_setups(
            self.config_entryconfig_entry, PLATFORMS
        )
        await self.async_dispatchasync_dispatch()
 
    async def async_update(self, event_time: datetime) -> None:
        """Update device.
 
        We do up to BATCH_SIZE calls in one update in order
        to minimize the calls on the api service.
        """
        for data_class in islice(self._queue, 0, BATCH_SIZE * self._interval_factor_interval_factor):
            if data_class.next_scan > time():
                continue
 
            if publisher := data_class.name:
                error = await self.async_fetch_dataasync_fetch_data(publisher)
 
                if error:
                    self.publisher[publisher].next_scan = (
                        time() + data_class.interval * 10
                    )
                else:
                    self.publisher[publisher].next_scan = time() + data_class.interval
 
        self._queue.rotate(BATCH_SIZE)
        cph = self.poll_countpoll_count / (time() - self.poll_startpoll_start) * 3600
        _LOGGER.debug("Calls per hour: %i", cph)
        if cph > self._rate_limit_rate_limit:
            for publisher in self.publisher.values():
                publisher.next_scan += 60
        if (time() - self.poll_startpoll_start) > 3600:
            self.poll_startpoll_start = time()
            self.poll_countpoll_count = 0
 
    @callback
    def async_force_update(self, signal_name: str) -> None:
        """Prioritize data retrieval for given data class entry."""
        self.publisher[signal_name].next_scan = time()
        self._queue.rotate(-(self._queue.index(self.publisher[signal_name])))
 
    async def handle_event(self, event: dict) -> None:
        """Handle webhook events."""
        if event["data"][WEBHOOK_PUSH_TYPE] == WEBHOOK_ACTIVATION:
            _LOGGER.debug("%s webhook successfully registered", MANUFACTURER)
            self._webhook_webhook = True
 
        elif event["data"][WEBHOOK_PUSH_TYPE] == WEBHOOK_DEACTIVATION:
            _LOGGER.debug("%s webhook unregistered", MANUFACTURER)
            self._webhook_webhook = False
 
        elif event["data"][WEBHOOK_PUSH_TYPE] == WEBHOOK_NACAMERA_CONNECTION:
            _LOGGER.debug("%s camera reconnected", MANUFACTURER)
            self.async_force_updateasync_force_update(ACCOUNT)
 
    async def async_fetch_data(self, signal_name: str) -> bool:
        """Fetch data and notify."""
        self.poll_countpoll_count += 1
        has_error = False
        try:
            await getattr(self.accountaccount, self.publisher[signal_name].method)(
                **self.publisher[signal_name].kwargs
            )
 
        except (pyatmo.NoDevice, pyatmo.ApiError) as err:
            _LOGGER.debug(err)
            has_error = True
 
        except (TimeoutError, aiohttp.ClientConnectorError) as err:
            _LOGGER.debug(err)
            return True
 
        for update_callback in self.publisher[signal_name].subscriptions:
            if update_callback:
                update_callback()
 
        return has_error
 
    async def subscribe(
        self,
        publisher: str,
        signal_name: str,
        update_callback: CALLBACK_TYPE | None,
        **kwargs: Any,
    ) -> None:
        """Subscribe to publisher."""
        if signal_name in self.publisher:
            if update_callback not in self.publisher[signal_name].subscriptions:
                self.publisher[signal_name].subscriptions.add(update_callback)
            return
 
        if publisher == "public":
            kwargs = {"area_id": self.accountaccount.register_public_weather_area(**kwargs)}
 
        interval = int(DEFAULT_INTERVALS[publisher] / self._interval_factor_interval_factor)
        self.publisher[signal_name] = NetatmoPublisher(
            name=signal_name,
            interval=interval,
            next_scan=time() + interval,
            subscriptions={update_callback},
            method=PUBLISHERS[publisher],
            kwargs=kwargs,
        )
 
        try:
            await self.async_fetch_dataasync_fetch_data(signal_name)
        except KeyError:
            self.publisher.pop(signal_name)
            raise
 
        self._queue.append(self.publisher[signal_name])
        _LOGGER.debug("Publisher %s added", signal_name)
 
    async def unsubscribe(
        self, signal_name: str, update_callback: CALLBACK_TYPE | None
    ) -> None:
        """Unsubscribe from publisher."""
        if update_callback not in self.publisher[signal_name].subscriptions:
            return
 
        self.publisher[signal_name].subscriptions.remove(update_callback)
 
        if not self.publisher[signal_name].subscriptions:
            self._queue.remove(self.publisher[signal_name])
            self.publisher.pop(signal_name)
            _LOGGER.debug("Publisher %s removed", signal_name)
 
    @property
    def webhook(self) -> bool:
        """Return the webhook state."""
        return self._webhook_webhook
 
    async def async_dispatch(self) -> None:
        """Dispatch the creation of entities."""
        await self.subscribesubscribe(WEATHER, WEATHER, None)
        await self.subscribesubscribe(AIR_CARE, AIR_CARE, None)
 
        self.setup_air_caresetup_air_care()
 
        for home in self.accountaccount.homes.values():
            signal_home = f"{HOME}-{home.entity_id}"
 
            await self.subscribesubscribe(HOME, signal_home, None, home_id=home.entity_id)
            await self.subscribesubscribe(EVENT, signal_home, None, home_id=home.entity_id)
 
            self.setup_climate_schedule_selectsetup_climate_schedule_select(home, signal_home)
            self.setup_roomssetup_rooms(home, signal_home)
            self.setup_modulessetup_modules(home, signal_home)
 
            self.hasshass.data[DOMAIN][DATA_PERSONS][home.entity_id] = {
                person.entity_id: person.pseudo for person in home.persons.values()
            }
 
        await self.unsubscribeunsubscribe(WEATHER, None)
        await self.unsubscribeunsubscribe(AIR_CARE, None)
 
    def setup_air_care(self) -> None:
        """Set up home coach/air care modules."""
        for module in self.accountaccount.modules.values():
            if module.device_category is NetatmoDeviceCategory.air_care:
                async_dispatcher_send(
                    self.hasshass,
                    NETATMO_CREATE_WEATHER_SENSOR,
                    NetatmoDevice(
                        self,
                        module,
                        AIR_CARE,
                        AIR_CARE,
                    ),
                )
 
    def setup_modules(self, home: pyatmo.Home, signal_home: str) -> None:
        """Set up modules."""
        netatmo_type_signal_map = {
            NetatmoDeviceCategory.camera: [
                NETATMO_CREATE_CAMERA,
                NETATMO_CREATE_CAMERA_LIGHT,
            ],
            NetatmoDeviceCategory.dimmer: [NETATMO_CREATE_LIGHT],
            NetatmoDeviceCategory.shutter: [NETATMO_CREATE_COVER],
            NetatmoDeviceCategory.switch: [
                NETATMO_CREATE_LIGHT,
                NETATMO_CREATE_SWITCH,
                NETATMO_CREATE_SENSOR,
            ],
            NetatmoDeviceCategory.meter: [NETATMO_CREATE_SENSOR],
            NetatmoDeviceCategory.fan: [NETATMO_CREATE_FAN],
        }
        for module in home.modules.values():
            if not module.device_category:
                continue
 
            for signal in netatmo_type_signal_map.get(module.device_category, []):
                async_dispatcher_send(
                    self.hasshass,
                    signal,
                    NetatmoDevice(
                        self,
                        module,
                        home.entity_id,
                        signal_home,
                    ),
                )
            if module.device_category is NetatmoDeviceCategory.weather:
                async_dispatcher_send(
                    self.hasshass,
                    NETATMO_CREATE_WEATHER_SENSOR,
                    NetatmoDevice(
                        self,
                        module,
                        home.entity_id,
                        WEATHER,
                    ),
                )
 
    def setup_rooms(self, home: pyatmo.Home, signal_home: str) -> None:
        """Set up rooms."""
        for room in home.rooms.values():
            if NetatmoDeviceCategory.climate in room.features:
                async_dispatcher_send(
                    self.hasshass,
                    NETATMO_CREATE_CLIMATE,
                    NetatmoRoom(
                        self,
                        room,
                        home.entity_id,
                        signal_home,
                    ),
                )
 
                for module in room.modules.values():
                    if module.device_category is NetatmoDeviceCategory.climate:
                        async_dispatcher_send(
                            self.hasshass,
                            NETATMO_CREATE_BATTERY,
                            NetatmoDevice(
                                self,
                                module,
                                room.entity_id,
                                signal_home,
                            ),
                        )
 
                if "humidity" in room.features:
                    async_dispatcher_send(
                        self.hasshass,
                        NETATMO_CREATE_ROOM_SENSOR,
                        NetatmoRoom(
                            self,
                            room,
                            room.entity_id,
                            signal_home,
                        ),
                    )
 
    def setup_climate_schedule_select(
        self, home: pyatmo.Home, signal_home: str
    ) -> None:
        """Set up climate schedule per home."""
        if NetatmoDeviceCategory.climate in [
            next(iter(x)) for x in [room.features for room in home.rooms.values()] if x
        ]:
            self.hasshass.data[DOMAIN][DATA_SCHEDULES][home.entity_id] = self.accountaccount.homes[
                home.entity_id
            ].schedules
 
            async_dispatcher_send(
                self.hasshass,
                NETATMO_CREATE_SELECT,
                NetatmoHome(
                    self,
                    home,
                    home.entity_id,
                    signal_home,
                ),
            )