coordinator.py

Go to the documentation of this file.

"""Helpers to help coordinate updates."""
 
from __future__ import annotations
 
from collections.abc import Callable, Coroutine
from datetime import timedelta
import logging
from typing import Any
 
from aiohttp import ClientConnectorError, ServerDisconnectedError
from pyoverkiz.client import OverkizClient
from pyoverkiz.enums import EventName, ExecutionState, Protocol
from pyoverkiz.exceptions import (
    BadCredentialsException,
    InvalidEventListenerIdException,
    MaintenanceException,
    NotAuthenticatedException,
    TooManyConcurrentRequestsException,
    TooManyRequestsException,
)
from pyoverkiz.models import Device, Event, Place
 
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util.decorator import Registry
 
from .const import DOMAIN, LOGGER, UPDATE_INTERVAL
 
EVENT_HANDLERS: Registry[
    str, Callable[[OverkizDataUpdateCoordinator, Event], Coroutine[Any, Any, None]]
] = Registry()
 
 
class OverkizDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Device]]):
    """Class to manage fetching data from Overkiz platform."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        logger: logging.Logger,
        *,
        name: str,
        client: OverkizClient,
        devices: list[Device],
        places: Place | None,
        update_interval: timedelta | None = None,
        config_entry_id: str,
    ) -> None:
        """Initialize global data updater."""
        super().__init__(
            hass,
            logger,
            name=name,
            update_interval=update_interval,
        )
 
        self.datadatadata = {}
        self.clientclient = client
        self.devicesdevices: dict[str, Device] = {d.device_url: d for d in devices}
        self.is_statelessis_stateless = all(
            device.protocol in (Protocol.RTS, Protocol.INTERNAL) for device in devices
        )
        self.executionsexecutions: dict[str, dict[str, str]] = {}
        self.areasareas = self._places_to_area_places_to_area(places) if places else None
        self.config_entry_idconfig_entry_id = config_entry_id
 
    async def _async_update_data(self) -> dict[str, Device]:
        """Fetch Overkiz data via event listener."""
        try:
            events = await self.clientclient.fetch_events()
        except BadCredentialsException as exception:
            raise ConfigEntryAuthFailed("Invalid authentication.") from exception
        except TooManyConcurrentRequestsException as exception:
            raise UpdateFailed("Too many concurrent requests.") from exception
        except TooManyRequestsException as exception:
            raise UpdateFailed("Too many requests, try again later.") from exception
        except MaintenanceException as exception:
            raise UpdateFailed("Server is down for maintenance.") from exception
        except InvalidEventListenerIdException as exception:
            raise UpdateFailed(exception) from exception
        except (TimeoutError, ClientConnectorError) as exception:
            raise UpdateFailed("Failed to connect.") from exception
        except (ServerDisconnectedError, NotAuthenticatedException):
            self.executionsexecutions = {}
 
            # During the relogin, similar exceptions can be thrown.
            try:
                await self.clientclient.login()
                self.devicesdevices = await self._get_devices_get_devices()
            except BadCredentialsException as exception:
                raise ConfigEntryAuthFailed("Invalid authentication.") from exception
            except TooManyRequestsException as exception:
                raise UpdateFailed("Too many requests, try again later.") from exception
 
            return self.devicesdevices
 
        for event in events:
            LOGGER.debug(event)
 
            if event_handler := EVENT_HANDLERS.get(event.name):
                await event_handler(self, event)
 
        if not self.executionsexecutions:
            self.update_intervalupdate_intervalupdate_intervalupdate_intervalupdate_interval = UPDATE_INTERVAL
 
        return self.devicesdevices
 
    async def _get_devices(self) -> dict[str, Device]:
        """Fetch devices."""
        LOGGER.debug("Fetching all devices and state via /setup/devices")
        return {d.device_url: d for d in await self.clientclient.get_devices(refresh=True)}
 
    def _places_to_area(self, place: Place) -> dict[str, str]:
        """Convert places with sub_places to a flat dictionary [placeoid, label])."""
        areas = {}
        if isinstance(place, Place):
            areas[place.oid] = place.label
 
        if isinstance(place.sub_places, list):
            for sub_place in place.sub_places:
                areas.update(self._places_to_area_places_to_area(sub_place))
 
        return areas
 
 
@EVENT_HANDLERS.register(EventName.DEVICE_AVAILABLE)
async def on_device_available(
    coordinator: OverkizDataUpdateCoordinator, event: Event
) -> None:
    """Handle device available event."""
    if event.device_url:
        coordinator.devices[event.device_url].available = True
 
 
@EVENT_HANDLERS.register(EventName.DEVICE_UNAVAILABLE)
@EVENT_HANDLERS.register(EventName.DEVICE_DISABLED)
async def on_device_unavailable_disabled(
    coordinator: OverkizDataUpdateCoordinator, event: Event
) -> None:
    """Handle device unavailable / disabled event."""
    if event.device_url:
        coordinator.devices[event.device_url].available = False
 
 
@EVENT_HANDLERS.register(EventName.DEVICE_CREATED)
@EVENT_HANDLERS.register(EventName.DEVICE_UPDATED)
async def on_device_created_updated(
    coordinator: OverkizDataUpdateCoordinator, event: Event
) -> None:
    """Handle device unavailable / disabled event."""
    coordinator.hass.async_create_task(
        coordinator.hass.config_entries.async_reload(coordinator.config_entry_id)
    )
 
 
@EVENT_HANDLERS.register(EventName.DEVICE_STATE_CHANGED)
async def on_device_state_changed(
    coordinator: OverkizDataUpdateCoordinator, event: Event
) -> None:
    """Handle device state changed event."""
    if not event.device_url:
        return
 
    for state in event.device_states:
        device = coordinator.devices[event.device_url]
        device.states[state.name] = state
 
 
@EVENT_HANDLERS.register(EventName.DEVICE_REMOVED)
async def on_device_removed(
    coordinator: OverkizDataUpdateCoordinator, event: Event
) -> None:
    """Handle device removed event."""
    if not event.device_url:
        return
 
    base_device_url = event.device_url.split("#")[0]
    registry = dr.async_get(coordinator.hass)
 
    if registered_device := registry.async_get_device(
        identifiers={(DOMAIN, base_device_url)}
    ):
        registry.async_remove_device(registered_device.id)
 
    if event.device_url:
        del coordinator.devices[event.device_url]
 
 
@EVENT_HANDLERS.register(EventName.EXECUTION_REGISTERED)
async def on_execution_registered(
    coordinator: OverkizDataUpdateCoordinator, event: Event
) -> None:
    """Handle execution registered event."""
    if event.exec_id and event.exec_id not in coordinator.executions:
        coordinator.executions[event.exec_id] = {}
 
    if not coordinator.is_stateless:
        coordinator.update_interval = timedelta(seconds=1)
 
 
@EVENT_HANDLERS.register(EventName.EXECUTION_STATE_CHANGED)
async def on_execution_state_changed(
    coordinator: OverkizDataUpdateCoordinator, event: Event
) -> None:
    """Handle execution changed event."""
    if event.exec_id in coordinator.executions and event.new_state in [
        ExecutionState.COMPLETED,
        ExecutionState.FAILED,
    ]:
        del coordinator.executions[event.exec_id]