home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-08-07 14:45:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

Pre-configure default doorbird events (#121692)

Browse Source
This commit is contained in:
J. Nick Koston
2024-07-11 04:31:29 -05:00
committed by GitHub
parent f40897b54c
commit ab7c4244d2
4 changed files with 151 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
homeassistant/components/doorbird/config_flow.py
View File

@@ -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, [])

13
homeassistant/components/doorbird/const.py
View File

@@ -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

163
homeassistant/components/doorbird/device.py
View File

@@ -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
if not self.door_station_events:
# User may not have permission to get the favorites
return
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
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(hass, prefer_external=False)
hass_url = get_url(self._hass, prefer_external=False)
if not self.door_station_events:
# User may not have permission to get the favorites
return
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()
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
)
return http_fav
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 _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,46 +204,38 @@ 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."""
"""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
# Register HA URL as webhook if not already, then get the ID
if await self.async_webhook_is_registered(url, favs=favs):
return True
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
_LOGGER.info("Successfully registered URL for %s on %s", event, self.name)
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
def get_event_data(self, event: str) -> dict[str, str | None]:
"""Get data to pass along with HA event."""
return {

10
tests/components/doorbird/test_config_flow.py
View File

@@ -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