From ab7c4244d2745b1915bc5ce5b6623672c21e91f4 Mon Sep 17 00:00:00 2001
From: "J. Nick Koston" <nick@koston.org>
Date: Thu, 11 Jul 2024 04:31:29 -0500
Subject: [PATCH] Pre-configure default doorbird events (#121692)
---
.../components/doorbird/config_flow.py | 15 +-
homeassistant/components/doorbird/const.py | 13 ++
homeassistant/components/doorbird/device.py | 169 +++++++++++++-----
tests/components/doorbird/test_config_flow.py | 10 +-
4 files changed, 154 insertions(+), 53 deletions(-)
diff --git a/homeassistant/components/doorbird/config_flow.py b/homeassistant/components/doorbird/config_flow.py
index 13e7d151d2f..f91e498b5e7 100644
--- a/homeassistant/components/doorbird/config_flow.py
+++ b/homeassistant/components/doorbird/config_flow.py
@@ -22,11 +22,19 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
-from .const import CONF_EVENTS, DOMAIN, DOORBIRD_OUI
+from .const import (
+ CONF_EVENTS,
+ DEFAULT_DOORBELL_EVENT,
+ DEFAULT_MOTION_EVENT,
+ DOMAIN,
+ DOORBIRD_OUI,
+)
from .util import get_mac_address_from_door_station_info
_LOGGER = logging.getLogger(__name__)
+DEFAULT_OPTIONS = {CONF_EVENTS: [DEFAULT_DOORBELL_EVENT, DEFAULT_MOTION_EVENT]}
+
def _schema_with_defaults(
host: str | None = None, name: str | None = None
@@ -99,7 +107,9 @@ class DoorBirdConfigFlow(ConfigFlow, domain=DOMAIN):
if not errors:
await self.async_set_unique_id(info["mac_addr"])
self._abort_if_unique_id_configured()
- return self.async_create_entry(title=info["title"], data=user_input)
+ return self.async_create_entry(
+ title=info["title"], data=user_input, options=DEFAULT_OPTIONS
+ )
data = self.discovery_schema or _schema_with_defaults()
return self.async_show_form(step_id="user", data_schema=data, errors=errors)
@@ -176,7 +186,6 @@ class OptionsFlowHandler(OptionsFlow):
"""Handle options flow."""
if user_input is not None:
events = [event.strip() for event in user_input[CONF_EVENTS].split(",")]
-
return self.async_create_entry(title="", data={CONF_EVENTS: events})
current_events = self.config_entry.options.get(CONF_EVENTS, [])
diff --git a/homeassistant/components/doorbird/const.py b/homeassistant/components/doorbird/const.py
index 4985b9ac9ea..40dafb5bdc8 100644
--- a/homeassistant/components/doorbird/const.py
+++ b/homeassistant/components/doorbird/const.py
@@ -22,3 +22,16 @@ DOORBIRD_INFO_KEY_WIFI_MAC_ADDR = "WIFI_MAC_ADDR"
UNDO_UPDATE_LISTENER = "undo_update_listener"
API_URL = f"/api/{DOMAIN}"
+
+
+DEFAULT_DOORBELL_EVENT = "doorbell"
+DEFAULT_MOTION_EVENT = "motion"
+
+DEFAULT_EVENT_TYPES = (
+ (DEFAULT_DOORBELL_EVENT, "doorbell"),
+ (DEFAULT_MOTION_EVENT, "motion"),
+)
+
+HTTP_EVENT_TYPE = "http"
+MIN_WEEKDAY = 104400
+MAX_WEEKDAY = 104399
diff --git a/homeassistant/components/doorbird/device.py b/homeassistant/components/doorbird/device.py
index 84a2d3abeab..9bb3397d0ff 100644
--- a/homeassistant/components/doorbird/device.py
+++ b/homeassistant/components/doorbird/device.py
@@ -2,19 +2,31 @@
from __future__ import annotations
+from collections import defaultdict
from dataclasses import dataclass
from functools import cached_property
import logging
from typing import Any
-from doorbirdpy import DoorBird, DoorBirdScheduleEntry
+from doorbirdpy import (
+ DoorBird,
+ DoorBirdScheduleEntry,
+ DoorBirdScheduleEntryOutput,
+ DoorBirdScheduleEntrySchedule,
+)
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
+from .const import (
+ API_URL,
+ DEFAULT_EVENT_TYPES,
+ HTTP_EVENT_TYPE,
+ MAX_WEEKDAY,
+ MIN_WEEKDAY,
+)
_LOGGER = logging.getLogger(__name__)
@@ -27,6 +39,15 @@ class DoorbirdEvent:
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."""
@@ -46,7 +67,9 @@ class ConfiguredDoorBird:
self._custom_url = custom_url
self._token = token
self._event_entity_ids = event_entity_ids
+ # Raw events, ie "doorbell" or "motion"
self.events: list[str] = []
+ # Event names, ie "doorbird_1234_doorbell" or "doorbird_1234_motion"
self.door_station_events: list[str] = []
self.event_descriptions: list[DoorbirdEvent] = []
@@ -79,34 +102,88 @@ class ConfiguredDoorBird:
async def async_register_events(self) -> None:
"""Register events on device."""
- hass = self._hass
- # Override url if another is specified in the configuration
- if custom_url := self.custom_url:
- hass_url = custom_url
- else:
- # Get the URL of this server
- hass_url = get_url(hass, prefer_external=False)
-
if not self.door_station_events:
# User may not have permission to get the favorites
return
- favorites = await self.device.favorites()
- for event in self.door_station_events:
- if await self._async_register_event(hass_url, event, favs=favorites):
- _LOGGER.info(
- "Successfully registered URL for %s on %s", event, self.name
- )
+ http_fav = await self._async_register_events()
+ event_config = await self._async_get_event_config(http_fav)
+ _LOGGER.debug("%s: Event config: %s", self.name, event_config)
+ if event_config.unconfigured_favorites:
+ await self._configure_unconfigured_favorites(event_config)
+ event_config = await self._async_get_event_config(http_fav)
+ self.event_descriptions = event_config.events
- schedule: list[DoorBirdScheduleEntry] = await self.device.schedule()
- http_fav: dict[str, dict[str, Any]] = favorites.get("http") or {}
- favorite_input_type: dict[str, str] = {
+ 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.device.change_schedule(entry)
+ if not update_ok:
+ _LOGGER.error(
+ "Unable to update schedule entry %s to %s. Error code: %s",
+ self.name,
+ 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_url:
+ hass_url = custom_url
+ else:
+ # Get the URL of this server
+ hass_url = get_url(self._hass, prefer_external=False)
+
+ http_fav = await self._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(hass_url, event, http_fav)
+ for event in self.door_station_events
+ ]
+ ):
+ # If any events were registered, get the updated favorites
+ http_fav = await self._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.device
+ schedule = await device.schedule()
+ favorite_input_type = {
output.param: entry.input
for entry in schedule
for output in entry.output
- if output.event == "http"
+ if output.event == HTTP_EVENT_TYPE
}
events: list[DoorbirdEvent] = []
+ unconfigured_favorites: defaultdict[str, list[str]] = defaultdict(list)
+ default_event_types = {
+ self._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"):
@@ -114,8 +191,10 @@ class ConfiguredDoorBird:
event = title.split("(")[1].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)
- self.event_descriptions = events
+ return DoorbirdEventConfig(events, schedule, unconfigured_favorites)
@cached_property
def slug(self) -> str:
@@ -125,45 +204,37 @@ class ConfiguredDoorBird:
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.device.favorites()).get(HTTP_EVENT_TYPE) or {}
+
async def _async_register_event(
- self, hass_url: str, event: str, favs: dict[str, Any] | None = None
+ self, hass_url: str, event: str, http_fav: dict[str, dict[str, Any]]
) -> bool:
- """Add a schedule entry in the device for a sensor."""
- url = f"{hass_url}{API_URL}/{event}?token={self._token}"
+ """Register an event.
- # Register HA URL as webhook if not already, then get the ID
- if await self.async_webhook_is_registered(url, favs=favs):
- return True
+ 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
- await self.device.change_favorite("http", f"Home Assistant ({event})", url)
- if not await self.async_webhook_is_registered(url):
+ if not await self.device.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
- return True
-
- async def async_webhook_is_registered(
- self, url: str, favs: dict[str, Any] | None = None
- ) -> bool:
- """Return whether the given URL is registered as a device favorite."""
- return await self.async_get_webhook_id(url, favs) is not None
- async def async_get_webhook_id(
- self, url: str, favs: dict[str, Any] | None = None
- ) -> str | None:
- """Return the device favorite ID for the given URL.
-
- The favorite must exist or there will be problems.
- """
- favs = favs if favs else await self.device.favorites()
- http_fav: dict[str, dict[str, Any]] = favs.get("http") or {}
- for fav_id, data in http_fav.items():
- if data["value"] == url:
- return fav_id
- return None
+ _LOGGER.info("Successfully registered URL for %s on %s", event, self.name)
+ return True
def get_event_data(self, event: str) -> dict[str, str | None]:
"""Get data to pass along with HA event."""
diff --git a/tests/components/doorbird/test_config_flow.py b/tests/components/doorbird/test_config_flow.py
index d77c5a81d96..107fd1454d3 100644
--- a/tests/components/doorbird/test_config_flow.py
+++ b/tests/components/doorbird/test_config_flow.py
@@ -8,7 +8,12 @@ import pytest
from homeassistant import config_entries
from homeassistant.components import zeroconf
-from homeassistant.components.doorbird.const import CONF_EVENTS, DOMAIN
+from homeassistant.components.doorbird.const import (
+ CONF_EVENTS,
+ DEFAULT_DOORBELL_EVENT,
+ DEFAULT_MOTION_EVENT,
+ DOMAIN,
+)
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@@ -82,6 +87,9 @@ async def test_user_form(hass: HomeAssistant) -> None:
"password": "password",
"username": "friend",
}
+ assert result2["options"] == {
+ CONF_EVENTS: [DEFAULT_DOORBELL_EVENT, DEFAULT_MOTION_EVENT]
+ }
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
--
GitLab