coordinator.py

Go to the documentation of this file.

"""Coordinator for the nuki component."""
 
from __future__ import annotations
 
import asyncio
from collections import defaultdict
from datetime import timedelta
import logging
 
from pynuki import NukiBridge, NukiLock, NukiOpener
from pynuki.bridge import InvalidCredentialsException
from pynuki.device import NukiDevice
from requests.exceptions import RequestException
 
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import DOMAIN, ERROR_STATES
from .helpers import parse_id
 
_LOGGER = logging.getLogger(__name__)
 
UPDATE_INTERVAL = timedelta(seconds=30)
 
 
class NukiCoordinator(DataUpdateCoordinator[None]):
    """Data Update Coordinator for the Nuki integration."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        bridge: NukiBridge,
        locks: list[NukiLock],
        openers: list[NukiOpener],
    ) -> None:
        """Initialize my coordinator."""
        super().__init__(
            hass,
            _LOGGER,
            # Name of the data. For logging purposes.
            name="nuki devices",
            # Polling interval. Will only be polled if there are subscribers.
            update_interval=UPDATE_INTERVAL,
        )
        self.bridgebridge = bridge
        self.lockslocks = locks
        self.openersopeners = openers
 
    @property
    def bridge_id(self):
        """Return the parsed id of the Nuki bridge."""
        return parse_id(self.bridgebridge.info()["ids"]["hardwareId"])
 
    async def _async_update_data(self) -> None:
        """Fetch data from Nuki bridge."""
        try:
            # Note: TimeoutError and aiohttp.ClientError are already
            # handled by the data update coordinator.
            async with asyncio.timeout(10):
                events = await self.hasshass.async_add_executor_job(
                    self.update_devicesupdate_devices, self.lockslocks + self.openersopeners
                )
        except InvalidCredentialsException as err:
            raise UpdateFailed(f"Invalid credentials for Bridge: {err}") from err
        except RequestException as err:
            raise UpdateFailed(f"Error communicating with Bridge: {err}") from err
 
        ent_reg = er.async_get(self.hasshass)
        for event, device_ids in events.items():
            for device_id in device_ids:
                entity_id = ent_reg.async_get_entity_id(
                    Platform.LOCK, DOMAIN, device_id
                )
                event_data = {
                    "entity_id": entity_id,
                    "type": event,
                }
                self.hasshass.bus.async_fire("nuki_event", event_data)
 
    def update_devices(self, devices: list[NukiDevice]) -> dict[str, set[str]]:
        """Update the Nuki devices.
 
        Returns:
            A dict with the events to be fired. The event type is the key and the device ids are the value
 
        """
 
        events: dict[str, set[str]] = defaultdict(set)
 
        for device in devices:
            for level in (False, True):
                try:
                    if isinstance(device, NukiOpener):
                        last_ring_action_state = device.ring_action_state
 
                        device.update(level)
 
                        if not last_ring_action_state and device.ring_action_state:
                            events["ring"].add(device.nuki_id)
                    else:
                        device.update(level)
                except RequestException:
                    continue
 
                if device.state not in ERROR_STATES:
                    break
 
        return events