device.py

Go to the documentation of this file.

"""Support for DoorBird devices."""
 
from __future__ import annotations
 
from collections import defaultdict
from dataclasses import dataclass
from http import HTTPStatus
import logging
from typing import Any
 
from aiohttp import ClientResponseError
from doorbirdpy import (
    DoorBird,
    DoorBirdScheduleEntry,
    DoorBirdScheduleEntryOutput,
    DoorBirdScheduleEntrySchedule,
)
from propcache import cached_property
 
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.helpers.network import get_url
from homeassistant.util import dt as dt_util, slugify
 
from .const import (
    API_URL,
    DEFAULT_EVENT_TYPES,
    HTTP_EVENT_TYPE,
    MAX_WEEKDAY,
    MIN_WEEKDAY,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
@dataclass(slots=True)
class DoorbirdEvent:
    """Describes a doorbird event."""
 
    event: str
    event_type: str
 
 
@dataclass(slots=True)
class DoorbirdEventConfig:
    """Describes the configuration of doorbird events."""
 
    events: list[DoorbirdEvent]
    schedule: list[DoorBirdScheduleEntry]
    unconfigured_favorites: defaultdict[str, list[str]]
 
 
class ConfiguredDoorBird:
    """Attach additional information to pass along with configured device."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        device: DoorBird,
        name: str | None,
        custom_url: str | None,
        token: str,
        event_entity_ids: dict[str, str],
    ) -> None:
        """Initialize configured device."""
        self._hass_hass = hass
        self._name_name = name
        self._device_device = device
        self._custom_url_custom_url = custom_url
        self._token_token = token
        self._event_entity_ids_event_entity_ids = event_entity_ids
        # Raw events, ie "doorbell" or "motion"
        self.eventsevents: list[str] = []
        # Event names, ie "doorbird_1234_doorbell" or "doorbird_1234_motion"
        self.door_station_eventsdoor_station_events: list[str] = []
        self.event_descriptionsevent_descriptions: list[DoorbirdEvent] = []
 
    def update_events(self, events: list[str]) -> None:
        """Update the doorbird events."""
        self.eventsevents = events
        self.door_station_eventsdoor_station_events = [
            self._get_event_name_get_event_name(event) for event in self.eventsevents
        ]
 
    @cached_property
    def name(self) -> str | None:
        """Get custom device name."""
        return self._name_name
 
    @cached_property
    def device(self) -> DoorBird:
        """Get the configured device."""
        return self._device_device
 
    @cached_property
    def custom_url(self) -> str | None:
        """Get custom url for device."""
        return self._custom_url_custom_url
 
    @cached_property
    def token(self) -> str:
        """Get token for device."""
        return self._token_token
 
    async def async_register_events(self) -> None:
        """Register events on device."""
        if not self.door_station_eventsdoor_station_events:
            # The config entry might not have any events configured yet
            return
        http_fav = await self._async_register_events_async_register_events()
        event_config = await self._async_get_event_config_async_get_event_config(http_fav)
        _LOGGER.debug("%s: Event config: %s", self.namename, event_config)
        if event_config.unconfigured_favorites:
            await self._configure_unconfigured_favorites_configure_unconfigured_favorites(event_config)
            event_config = await self._async_get_event_config_async_get_event_config(http_fav)
        self.event_descriptionsevent_descriptions = event_config.events
 
    async def _configure_unconfigured_favorites(
        self, event_config: DoorbirdEventConfig
    ) -> None:
        """Configure unconfigured favorites."""
        for entry in event_config.schedule:
            modified_schedule = False
            for identifier in event_config.unconfigured_favorites.get(entry.input, ()):
                schedule = DoorBirdScheduleEntrySchedule()
                schedule.add_weekday(MIN_WEEKDAY, MAX_WEEKDAY)
                entry.output.append(
                    DoorBirdScheduleEntryOutput(
                        enabled=True,
                        event=HTTP_EVENT_TYPE,
                        param=identifier,
                        schedule=schedule,
                    )
                )
                modified_schedule = True
 
            if modified_schedule:
                update_ok, code = await self.devicedevice.change_schedule(entry)
                if not update_ok:
                    _LOGGER.error(
                        "Unable to update schedule entry %s to %s. Error code: %s",
                        self.namename,
                        entry.export,
                        code,
                    )
 
    async def _async_register_events(self) -> dict[str, Any]:
        """Register events on device."""
        # Override url if another is specified in the configuration
        if custom_url := self.custom_urlcustom_url:
            hass_url = custom_url
        else:
            # Get the URL of this server
            hass_url = get_url(self._hass_hass, prefer_external=False)
 
        http_fav = await self._async_get_http_favorites_async_get_http_favorites()
        if any(
            # Note that a list comp is used here to ensure all
            # events are registered and the any does not short circuit
            [
                await self._async_register_event_async_register_event(hass_url, event, http_fav)
                for event in self.door_station_eventsdoor_station_events
            ]
        ):
            # If any events were registered, get the updated favorites
            http_fav = await self._async_get_http_favorites_async_get_http_favorites()
 
        return http_fav
 
    async def _async_get_event_config(
        self, http_fav: dict[str, dict[str, Any]]
    ) -> DoorbirdEventConfig:
        """Get events and unconfigured favorites from http favorites."""
        device = self.devicedevice
        events: list[DoorbirdEvent] = []
        unconfigured_favorites: defaultdict[str, list[str]] = defaultdict(list)
        try:
            schedule = await device.schedule()
        except ClientResponseError as ex:
            if ex.status == HTTPStatus.NOT_FOUND:
                # D301 models do not support schedules
                return DoorbirdEventConfig(events, [], unconfigured_favorites)
            raise
        favorite_input_type = {
            output.param: entry.input
            for entry in schedule
            for output in entry.output
            if output.event == HTTP_EVENT_TYPE
        }
        default_event_types = {
            self._get_event_name_get_event_name(event): event_type
            for event, event_type in DEFAULT_EVENT_TYPES
        }
        for identifier, data in http_fav.items():
            title: str | None = data.get("title")
            if not title or not title.startswith("Home Assistant"):
                continue
            event = title.partition("(")[2].strip(")")
            if input_type := favorite_input_type.get(identifier):
                events.append(DoorbirdEvent(event, input_type))
            elif input_type := default_event_types.get(event):
                unconfigured_favorites[input_type].append(identifier)
 
        return DoorbirdEventConfig(events, schedule, unconfigured_favorites)
 
    @cached_property
    def slug(self) -> str:
        """Get device slug."""
        return slugify(self._name_name)
 
    def _get_event_name(self, event: str) -> str:
        return f"{self.slug}_{event}"
 
    async def _async_get_http_favorites(self) -> dict[str, dict[str, Any]]:
        """Get the HTTP favorites from the device."""
        return (await self.devicedevice.favorites()).get(HTTP_EVENT_TYPE) or {}
 
    async def _async_register_event(
        self, hass_url: str, event: str, http_fav: dict[str, dict[str, Any]]
    ) -> bool:
        """Register an event.
 
        Returns True if the event was registered, False if
        the event was already registered or registration failed.
        """
        url = f"{hass_url}{API_URL}/{event}?token={self._token}"
        _LOGGER.debug("Registering URL %s for event %s", url, event)
        # If its already registered, don't register it again
        if any(fav["value"] == url for fav in http_fav.values()):
            _LOGGER.debug("URL already registered for %s", event)
            return False
 
        if not await self.devicedevice.change_favorite(
            HTTP_EVENT_TYPE, f"Home Assistant ({event})", url
        ):
            _LOGGER.warning(
                'Unable to set favorite URL "%s". Event "%s" will not fire',
                url,
                event,
            )
            return False
 
        _LOGGER.debug("Successfully registered URL for %s on %s", event, self.namename)
        return True
 
    def get_event_data(self, event: str) -> dict[str, str | None]:
        """Get data to pass along with HA event."""
        return {
            "timestamp": dt_util.utcnow().isoformat(),
            "live_video_url": self._device_device.live_video_url,
            "live_image_url": self._device_device.live_image_url,
            "rtsp_live_video_url": self._device_device.rtsp_live_video_url,
            "html5_viewer_url": self._device_device.html5_viewer_url,
            ATTR_ENTITY_ID: self._event_entity_ids_event_entity_ids.get(event),
        }
 
 
async def async_reset_device_favorites(door_station: ConfiguredDoorBird) -> None:
    """Handle clearing favorites on device."""
    door_bird = door_station.device
    favorites = await door_bird.favorites()
    for favorite_type, favorite_ids in favorites.items():
        for favorite_id in favorite_ids:
            await door_bird.delete_favorite(favorite_type, favorite_id)
    await door_station.async_register_events()