coordinator.py

Go to the documentation of this file.

"""Provides the Toon DataUpdateCoordinator."""
 
from __future__ import annotations
 
import logging
import secrets
 
from aiohttp import web
from toonapi import Status, Toon, ToonError
 
from homeassistant.components import cloud, webhook
from homeassistant.components.webhook import (
    async_register as webhook_register,
    async_unregister as webhook_unregister,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_WEBHOOK_ID, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import Event, HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.config_entry_oauth2_flow import OAuth2Session
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import CONF_CLOUDHOOK_URL, DEFAULT_SCAN_INTERVAL, DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
 
class ToonDataUpdateCoordinator(DataUpdateCoordinator[Status]):
    """Class to manage fetching Toon data from single endpoint."""
 
    def __init__(
        self, hass: HomeAssistant, *, entry: ConfigEntry, session: OAuth2Session
    ) -> None:
        """Initialize global Toon data updater."""
        self.sessionsession = session
        self.entryentry = entry
 
        async def async_token_refresh() -> str:
            await session.async_ensure_token_valid()
            return session.token["access_token"]
 
        self.toontoon = Toon(
            token=session.token["access_token"],
            session=async_get_clientsession(hass),
            token_refresh_method=async_token_refresh,
        )
 
        super().__init__(
            hass, _LOGGER, name=DOMAIN, update_interval=DEFAULT_SCAN_INTERVAL
        )
 
    async def register_webhook(self, event: Event | None = None) -> None:
        """Register a webhook with Toon to get live updates."""
        if CONF_WEBHOOK_ID not in self.entryentry.data:
            data = {**self.entryentry.data, CONF_WEBHOOK_ID: secrets.token_hex()}
            self.hasshass.config_entries.async_update_entry(self.entryentry, data=data)
 
        if cloud.async_active_subscription(self.hasshass):
            if CONF_CLOUDHOOK_URL not in self.entryentry.data:
                try:
                    webhook_url = await cloud.async_create_cloudhook(
                        self.hasshass, self.entryentry.data[CONF_WEBHOOK_ID]
                    )
                except cloud.CloudNotConnected:
                    webhook_url = webhook.async_generate_url(
                        self.hasshass, self.entryentry.data[CONF_WEBHOOK_ID]
                    )
                else:
                    data = {**self.entryentry.data, CONF_CLOUDHOOK_URL: webhook_url}
                    self.hasshass.config_entries.async_update_entry(self.entryentry, data=data)
            else:
                webhook_url = self.entryentry.data[CONF_CLOUDHOOK_URL]
        else:
            webhook_url = webhook.async_generate_url(
                self.hasshass, self.entryentry.data[CONF_WEBHOOK_ID]
            )
 
        # Ensure the webhook is not registered already
        webhook_unregister(self.hasshass, self.entryentry.data[CONF_WEBHOOK_ID])
 
        webhook_register(
            self.hasshass,
            DOMAIN,
            "Toon",
            self.entryentry.data[CONF_WEBHOOK_ID],
            self.handle_webhookhandle_webhook,
        )
 
        try:
            await self.toontoon.subscribe_webhook(
                application_id=self.entryentry.entry_id, url=webhook_url
            )
            _LOGGER.debug("Registered Toon webhook: %s", webhook_url)
        except ToonError as err:
            _LOGGER.error("Error during webhook registration - %s", err)
 
        self.hasshass.bus.async_listen_once(
            EVENT_HOMEASSISTANT_STOP, self.unregister_webhookunregister_webhook
        )
 
    async def handle_webhook(
        self, hass: HomeAssistant, webhook_id: str, request: web.Request
    ) -> None:
        """Handle webhook callback."""
        try:
            data = await request.json()
        except ValueError:
            return
 
        _LOGGER.debug("Got webhook data: %s", data)
 
        # Webhook expired notification, re-register
        if data.get("code") == 510:
            await self.register_webhookregister_webhook()
            return
 
        if (
            "updateDataSet" not in data
            or "commonName" not in data
            or self.datadata.agreement.display_common_name != data["commonName"]
        ):
            _LOGGER.warning("Received invalid data from Toon webhook - %s", data)
            return
 
        try:
            await self.toontoon.update(data["updateDataSet"])
            self.async_update_listenersasync_update_listeners()
        except ToonError as err:
            _LOGGER.error("Could not process data received from Toon webhook - %s", err)
 
    async def unregister_webhook(self, event: Event | None = None) -> None:
        """Remove / Unregister webhook for toon."""
        _LOGGER.debug(
            "Unregistering Toon webhook (%s)", self.entryentry.data[CONF_WEBHOOK_ID]
        )
        try:
            await self.toontoon.unsubscribe_webhook(self.entryentry.entry_id)
        except ToonError as err:
            _LOGGER.error("Failed unregistering Toon webhook - %s", err)
 
        webhook_unregister(self.hasshass, self.entryentry.data[CONF_WEBHOOK_ID])
 
    async def _async_update_data(self) -> Status:
        """Fetch data from Toon."""
        try:
            return await self.toontoon.update()
        except ToonError as error:
            raise UpdateFailed(f"Invalid response from API: {error}") from error