diff --git a/.coveragerc b/.coveragerc index 9930cbaf0b5..4dd8dea258d 100644 --- a/.coveragerc +++ b/.coveragerc @@ -212,8 +212,9 @@ omit = homeassistant/components/dominos/* homeassistant/components/doods/* homeassistant/components/doorbird/__init__.py - homeassistant/components/doorbird/button.py homeassistant/components/doorbird/camera.py + homeassistant/components/doorbird/button.py + homeassistant/components/doorbird/device.py homeassistant/components/doorbird/entity.py homeassistant/components/doorbird/util.py homeassistant/components/dormakaba_dkey/__init__.py diff --git a/homeassistant/components/doorbird/__init__.py b/homeassistant/components/doorbird/__init__.py index deb37c1bfe3..8651f7de6de 100644 --- a/homeassistant/components/doorbird/__init__.py +++ b/homeassistant/components/doorbird/__init__.py @@ -24,11 +24,10 @@ from homeassistant.const import ( from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import ConfigEntryNotReady import homeassistant.helpers.config_validation as cv -from homeassistant.helpers.network import get_url from homeassistant.helpers.typing import ConfigType -from homeassistant.util import dt as dt_util, slugify from .const import ( + API_URL, CONF_EVENTS, DOMAIN, DOOR_STATION, @@ -37,12 +36,11 @@ from .const import ( PLATFORMS, UNDO_UPDATE_LISTENER, ) +from .device import ConfiguredDoorBird from .util import get_doorstation_by_token _LOGGER = logging.getLogger(__name__) -API_URL = f"/api/{DOMAIN}" - CONF_CUSTOM_URL = "hass_url_override" RESET_DEVICE_FAVORITES = "doorbird_reset_favorites" @@ -128,9 +126,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: ) raise ConfigEntryNotReady - token = doorstation_config.get(CONF_TOKEN, config_entry_id) - custom_url = doorstation_config.get(CONF_CUSTOM_URL) - name = doorstation_config.get(CONF_NAME) + token: str = doorstation_config.get(CONF_TOKEN, config_entry_id) + custom_url: str | None = doorstation_config.get(CONF_CUSTOM_URL) + name: str | None = doorstation_config.get(CONF_NAME) events = doorstation_options.get(CONF_EVENTS, []) doorstation = ConfiguredDoorBird(device, name, custom_url, token) doorstation.update_events(events) @@ -151,7 +149,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return True -def _init_doorbird_device(device): +def _init_doorbird_device(device: DoorBird) -> tuple[tuple[bool, int], dict[str, Any]]: return device.ready(), device.info() @@ -211,122 +209,6 @@ def _async_import_options_from_data_if_missing(hass: HomeAssistant, entry: Confi hass.config_entries.async_update_entry(entry, options=options) -class ConfiguredDoorBird: - """Attach additional information to pass along with configured device.""" - - def __init__(self, device, name, custom_url, token): - """Initialize configured device.""" - self._name = name - self._device = device - self._custom_url = custom_url - self.events = None - self.doorstation_events = None - self._token = token - - def update_events(self, events): - """Update the doorbird events.""" - self.events = events - self.doorstation_events = [self._get_event_name(event) for event in self.events] - - @property - def name(self): - """Get custom device name.""" - return self._name - - @property - def device(self): - """Get the configured device.""" - return self._device - - @property - def custom_url(self): - """Get custom url for device.""" - return self._custom_url - - @property - def token(self): - """Get token for device.""" - return self._token - - def register_events(self, hass: HomeAssistant) -> None: - """Register events on device.""" - # Get the URL of this server - hass_url = get_url(hass, prefer_external=False) - - # Override url if another is specified in the configuration - if self.custom_url is not None: - hass_url = self.custom_url - - if not self.doorstation_events: - # User may not have permission to get the favorites - return - - favorites = self.device.favorites() - for event in self.doorstation_events: - if self._register_event(hass_url, event, favs=favorites): - _LOGGER.info( - "Successfully registered URL for %s on %s", event, self.name - ) - - @property - def slug(self): - """Get device slug.""" - return slugify(self._name) - - def _get_event_name(self, event): - return f"{self.slug}_{event}" - - def _register_event( - self, hass_url: str, event: str, favs: dict[str, Any] | None = None - ) -> bool: - """Add a schedule entry in the device for a sensor.""" - url = f"{hass_url}{API_URL}/{event}?token={self._token}" - - # Register HA URL as webhook if not already, then get the ID - if self.webhook_is_registered(url, favs=favs): - return True - - self.device.change_favorite("http", f"Home Assistant ({event})", url) - if not self.webhook_is_registered(url): - _LOGGER.warning( - 'Unable to set favorite URL "%s". Event "%s" will not fire', - url, - event, - ) - return False - return True - - def webhook_is_registered(self, url, favs=None) -> bool: - """Return whether the given URL is registered as a device favorite.""" - return self.get_webhook_id(url, favs) is not None - - def get_webhook_id(self, url, favs=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 self.device.favorites() - - if "http" not in favs: - return None - - for fav_id in favs["http"]: - if favs["http"][fav_id]["value"] == url: - return fav_id - - return None - - def get_event_data(self): - """Get data to pass along with HA event.""" - return { - "timestamp": dt_util.utcnow().isoformat(), - "live_video_url": self._device.live_video_url, - "live_image_url": self._device.live_image_url, - "rtsp_live_video_url": self._device.rtsp_live_video_url, - "html5_viewer_url": self._device.html5_viewer_url, - } - - class DoorBirdRequestView(HomeAssistantView): """Provide a page for the device to call.""" diff --git a/homeassistant/components/doorbird/const.py b/homeassistant/components/doorbird/const.py index 767366af734..416603a312c 100644 --- a/homeassistant/components/doorbird/const.py +++ b/homeassistant/components/doorbird/const.py @@ -19,3 +19,5 @@ DOORBIRD_INFO_KEY_PRIMARY_MAC_ADDR = "PRIMARY_MAC_ADDR" DOORBIRD_INFO_KEY_WIFI_MAC_ADDR = "WIFI_MAC_ADDR" UNDO_UPDATE_LISTENER = "undo_update_listener" + +API_URL = f"/api/{DOMAIN}" diff --git a/homeassistant/components/doorbird/device.py b/homeassistant/components/doorbird/device.py new file mode 100644 index 00000000000..1c787feb934 --- /dev/null +++ b/homeassistant/components/doorbird/device.py @@ -0,0 +1,137 @@ +"""Support for DoorBird devices.""" +from __future__ import annotations + +import logging +from typing import Any + +from doorbirdpy import DoorBird + +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 + +_LOGGER = logging.getLogger(__name__) + + +class ConfiguredDoorBird: + """Attach additional information to pass along with configured device.""" + + def __init__( + self, device: DoorBird, name: str | None, custom_url: str | None, token: str + ) -> None: + """Initialize configured device.""" + self._name = name + self._device = device + self._custom_url = custom_url + self.events = None + self.doorstation_events = None + self._token = token + + def update_events(self, events): + """Update the doorbird events.""" + self.events = events + self.doorstation_events = [self._get_event_name(event) for event in self.events] + + @property + def name(self) -> str | None: + """Get custom device name.""" + return self._name + + @property + def device(self) -> DoorBird: + """Get the configured device.""" + return self._device + + @property + def custom_url(self) -> str | None: + """Get custom url for device.""" + return self._custom_url + + @property + def token(self) -> str: + """Get token for device.""" + return self._token + + def register_events(self, hass: HomeAssistant) -> None: + """Register events on device.""" + # Get the URL of this server + hass_url = get_url(hass, prefer_external=False) + + # Override url if another is specified in the configuration + if self.custom_url is not None: + hass_url = self.custom_url + + if not self.doorstation_events: + # User may not have permission to get the favorites + return + + favorites = self.device.favorites() + for event in self.doorstation_events: + if self._register_event(hass_url, event, favs=favorites): + _LOGGER.info( + "Successfully registered URL for %s on %s", event, self.name + ) + + @property + def slug(self) -> str: + """Get device slug.""" + return slugify(self._name) + + def _get_event_name(self, event: str) -> str: + return f"{self.slug}_{event}" + + def _register_event( + self, hass_url: str, event: str, favs: dict[str, Any] | None = None + ) -> bool: + """Add a schedule entry in the device for a sensor.""" + url = f"{hass_url}{API_URL}/{event}?token={self._token}" + + # Register HA URL as webhook if not already, then get the ID + if self.webhook_is_registered(url, favs=favs): + return True + + self.device.change_favorite("http", f"Home Assistant ({event})", url) + if not self.webhook_is_registered(url): + _LOGGER.warning( + 'Unable to set favorite URL "%s". Event "%s" will not fire', + url, + event, + ) + return False + return True + + def 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 self.get_webhook_id(url, favs) is not None + + def 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 self.device.favorites() + + if "http" not in favs: + return None + + for fav_id in favs["http"]: + if favs["http"][fav_id]["value"] == url: + return fav_id + + return None + + def get_event_data(self) -> dict[str, str]: + """Get data to pass along with HA event.""" + return { + "timestamp": dt_util.utcnow().isoformat(), + "live_video_url": self._device.live_video_url, + "live_image_url": self._device.live_image_url, + "rtsp_live_video_url": self._device.rtsp_live_video_url, + "html5_viewer_url": self._device.html5_viewer_url, + } diff --git a/homeassistant/components/doorbird/entity.py b/homeassistant/components/doorbird/entity.py index 65431e38be1..32c9cfff784 100644 --- a/homeassistant/components/doorbird/entity.py +++ b/homeassistant/components/doorbird/entity.py @@ -1,5 +1,7 @@ """The DoorBird integration base entity.""" +from typing import Any + from homeassistant.helpers import device_registry as dr from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity import Entity @@ -10,6 +12,7 @@ from .const import ( DOORBIRD_INFO_KEY_FIRMWARE, MANUFACTURER, ) +from .device import ConfiguredDoorBird from .util import get_mac_address_from_doorstation_info @@ -18,7 +21,9 @@ class DoorBirdEntity(Entity): _attr_has_entity_name = True - def __init__(self, doorstation, doorstation_info): + def __init__( + self, doorstation: ConfiguredDoorBird, doorstation_info: dict[str, Any] + ) -> None: """Initialize the entity.""" super().__init__() self._doorstation = doorstation diff --git a/homeassistant/components/doorbird/util.py b/homeassistant/components/doorbird/util.py index 55974bc1866..7b406bc07fa 100644 --- a/homeassistant/components/doorbird/util.py +++ b/homeassistant/components/doorbird/util.py @@ -1,6 +1,9 @@ """DoorBird integration utils.""" +from homeassistant.core import HomeAssistant + from .const import DOMAIN, DOOR_STATION +from .device import ConfiguredDoorBird def get_mac_address_from_doorstation_info(doorstation_info): @@ -10,17 +13,23 @@ def get_mac_address_from_doorstation_info(doorstation_info): return doorstation_info["WIFI_MAC_ADDR"] -def get_doorstation_by_token(hass, token): +def get_doorstation_by_token( + hass: HomeAssistant, token: str +) -> ConfiguredDoorBird | None: """Get doorstation by token.""" return _get_doorstation_by_attr(hass, "token", token) -def get_doorstation_by_slug(hass, slug): +def get_doorstation_by_slug( + hass: HomeAssistant, slug: str +) -> ConfiguredDoorBird | None: """Get doorstation by slug.""" return _get_doorstation_by_attr(hass, "slug", slug) -def _get_doorstation_by_attr(hass, attr, val): +def _get_doorstation_by_attr( + hass: HomeAssistant, attr: str, val: str +) -> ConfiguredDoorBird | None: for entry in hass.data[DOMAIN].values(): if DOOR_STATION not in entry: continue @@ -33,7 +42,7 @@ def _get_doorstation_by_attr(hass, attr, val): return None -def get_all_doorstations(hass): +def get_all_doorstations(hass: HomeAssistant) -> list[ConfiguredDoorBird]: """Get all doorstations.""" return [ entry[DOOR_STATION]