From 30a88e9e61201368dec3f41c238cad096a8f60d5 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Thu, 17 Aug 2023 09:37:54 -0500 Subject: [PATCH] Additional doorbird cleanups to prepare for event entities (#98542) --- homeassistant/components/doorbird/__init__.py | 113 +++++++----------- homeassistant/components/doorbird/button.py | 25 ++-- homeassistant/components/doorbird/camera.py | 67 ++++------- .../components/doorbird/config_flow.py | 11 +- homeassistant/components/doorbird/device.py | 26 ++-- homeassistant/components/doorbird/entity.py | 24 ++-- homeassistant/components/doorbird/logbook.py | 26 ++-- homeassistant/components/doorbird/models.py | 26 ++++ homeassistant/components/doorbird/util.py | 49 ++------ 9 files changed, 158 insertions(+), 209 deletions(-) create mode 100644 homeassistant/components/doorbird/models.py diff --git a/homeassistant/components/doorbird/__init__.py b/homeassistant/components/doorbird/__init__.py index 8651f7de6de..d1ad91bbb2c 100644 --- a/homeassistant/components/doorbird/__init__.py +++ b/homeassistant/components/doorbird/__init__.py @@ -14,30 +14,21 @@ from homeassistant.components import persistent_notification from homeassistant.components.http import HomeAssistantView from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( - ATTR_ENTITY_ID, CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME, ) -from homeassistant.core import HomeAssistant, callback +from homeassistant.core import Event, HomeAssistant, callback from homeassistant.exceptions import ConfigEntryNotReady import homeassistant.helpers.config_validation as cv from homeassistant.helpers.typing import ConfigType -from .const import ( - API_URL, - CONF_EVENTS, - DOMAIN, - DOOR_STATION, - DOOR_STATION_EVENT_ENTITY_IDS, - DOOR_STATION_INFO, - PLATFORMS, - UNDO_UPDATE_LISTENER, -) +from .const import API_URL, CONF_EVENTS, DOMAIN, PLATFORMS from .device import ConfiguredDoorBird -from .util import get_doorstation_by_token +from .models import DoorBirdData +from .util import get_door_station_by_token _LOGGER = logging.getLogger(__name__) @@ -64,26 +55,25 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up the DoorBird component.""" hass.data.setdefault(DOMAIN, {}) - # Provide an endpoint for the doorstations to call to trigger events + # Provide an endpoint for the door stations to call to trigger events hass.http.register_view(DoorBirdRequestView) - def _reset_device_favorites_handler(event): + def _reset_device_favorites_handler(event: Event) -> None: """Handle clearing favorites on device.""" if (token := event.data.get("token")) is None: return - doorstation = get_doorstation_by_token(hass, token) + door_station = get_door_station_by_token(hass, token) - if doorstation is None: + if door_station is None: _LOGGER.error("Device not found for provided token") return # Clear webhooks - favorites = doorstation.device.favorites() - - for favorite_type in favorites: - for favorite_id in favorites[favorite_type]: - doorstation.device.delete_favorite(favorite_type, favorite_id) + favorites: dict[str, list[str]] = door_station.device.favorites() + for favorite_type, favorite_ids in favorites.items(): + for favorite_id in favorite_ids: + door_station.device.delete_favorite(favorite_type, favorite_id) hass.bus.async_listen(RESET_DEVICE_FAVORITES, _reset_device_favorites_handler) @@ -95,17 +85,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: _async_import_options_from_data_if_missing(hass, entry) - doorstation_config = entry.data - doorstation_options = entry.options + door_station_config = entry.data config_entry_id = entry.entry_id - device_ip = doorstation_config[CONF_HOST] - username = doorstation_config[CONF_USERNAME] - password = doorstation_config[CONF_PASSWORD] + device_ip = door_station_config[CONF_HOST] + username = door_station_config[CONF_USERNAME] + password = door_station_config[CONF_PASSWORD] device = DoorBird(device_ip, username, password) try: - status, info = await hass.async_add_executor_job(_init_doorbird_device, device) + status, info = await hass.async_add_executor_job(_init_door_bird_device, device) except requests.exceptions.HTTPError as err: if err.response.status_code == HTTPStatus.UNAUTHORIZED: _LOGGER.error( @@ -126,50 +115,43 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: ) raise ConfigEntryNotReady - 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) + token: str = door_station_config.get(CONF_TOKEN, config_entry_id) + custom_url: str | None = door_station_config.get(CONF_CUSTOM_URL) + name: str | None = door_station_config.get(CONF_NAME) + events = entry.options.get(CONF_EVENTS, []) + event_entity_ids: dict[str, str] = {} + door_station = ConfiguredDoorBird(device, name, custom_url, token, event_entity_ids) + door_bird_data = DoorBirdData(door_station, info, event_entity_ids) + door_station.update_events(events) # Subscribe to doorbell or motion events - if not await _async_register_events(hass, doorstation): + if not await _async_register_events(hass, door_station): raise ConfigEntryNotReady - undo_listener = entry.add_update_listener(_update_listener) - - hass.data[DOMAIN][config_entry_id] = { - DOOR_STATION: doorstation, - DOOR_STATION_INFO: info, - UNDO_UPDATE_LISTENER: undo_listener, - } - + entry.async_on_unload(entry.add_update_listener(_update_listener)) + hass.data[DOMAIN][config_entry_id] = door_bird_data await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) return True -def _init_doorbird_device(device: DoorBird) -> tuple[tuple[bool, int], dict[str, Any]]: +def _init_door_bird_device(device: DoorBird) -> tuple[tuple[bool, int], dict[str, Any]]: + """Verify we can connect to the device and return the status.""" return device.ready(), device.info() async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" - - hass.data[DOMAIN][entry.entry_id][UNDO_UPDATE_LISTENER]() - - unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) - if unload_ok: - hass.data[DOMAIN].pop(entry.entry_id) - + data: dict[str, DoorBirdData] = hass.data[DOMAIN] + if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS): + data.pop(entry.entry_id) return unload_ok async def _async_register_events( - hass: HomeAssistant, doorstation: ConfiguredDoorBird + hass: HomeAssistant, door_station: ConfiguredDoorBird ) -> bool: try: - await hass.async_add_executor_job(doorstation.register_events, hass) + await hass.async_add_executor_job(door_station.register_events, hass) except requests.exceptions.HTTPError: persistent_notification.async_create( hass, @@ -190,10 +172,11 @@ async def _async_register_events( async def _update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: """Handle options update.""" config_entry_id = entry.entry_id - doorstation = hass.data[DOMAIN][config_entry_id][DOOR_STATION] - doorstation.update_events(entry.options[CONF_EVENTS]) + data: DoorBirdData = hass.data[DOMAIN][config_entry_id] + door_station = data.door_station + door_station.update_events(entry.options[CONF_EVENTS]) # Subscribe to doorbell or motion events - await _async_register_events(hass, doorstation) + await _async_register_events(hass, door_station) @callback @@ -217,21 +200,17 @@ class DoorBirdRequestView(HomeAssistantView): name = API_URL[1:].replace("/", ":") extra_urls = [API_URL + "/{event}"] - async def get(self, request, event): + async def get(self, request: web.Request, event: str) -> web.Response: """Respond to requests from the device.""" - hass = request.app["hass"] - - token = request.query.get("token") - - device = get_doorstation_by_token(hass, token) - - if device is None: + hass: HomeAssistant = request.app["hass"] + token: str | None = request.query.get("token") + if token is None or (device := get_door_station_by_token(hass, token)) is None: return web.Response( status=HTTPStatus.UNAUTHORIZED, text="Invalid token provided." ) if device: - event_data = device.get_event_data() + event_data = device.get_event_data(event) else: event_data = {} @@ -241,10 +220,6 @@ class DoorBirdRequestView(HomeAssistantView): message = f"HTTP Favorites cleared for {device.slug}" return web.Response(text=message) - event_data[ATTR_ENTITY_ID] = hass.data[DOMAIN][ - DOOR_STATION_EVENT_ENTITY_IDS - ].get(event) - hass.bus.async_fire(f"{DOMAIN}_{event}", event_data) return web.Response(text="OK") diff --git a/homeassistant/components/doorbird/button.py b/homeassistant/components/doorbird/button.py index fb13a6f5be3..1c69429d3c7 100644 --- a/homeassistant/components/doorbird/button.py +++ b/homeassistant/components/doorbird/button.py @@ -10,8 +10,9 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .const import DOMAIN, DOOR_STATION, DOOR_STATION_INFO +from .const import DOMAIN from .entity import DoorBirdEntity +from .models import DoorBirdData IR_RELAY = "__ir_light__" @@ -49,20 +50,14 @@ async def async_setup_entry( ) -> None: """Set up the DoorBird button platform.""" config_entry_id = config_entry.entry_id - - data = hass.data[DOMAIN][config_entry_id] - doorstation = data[DOOR_STATION] - doorstation_info = data[DOOR_STATION_INFO] - - relays = doorstation_info["RELAYS"] + door_bird_data: DoorBirdData = hass.data[DOMAIN][config_entry_id] + relays = door_bird_data.door_station_info["RELAYS"] entities = [ - DoorBirdButton(doorstation, doorstation_info, relay, RELAY_ENTITY_DESCRIPTION) + DoorBirdButton(door_bird_data, relay, RELAY_ENTITY_DESCRIPTION) for relay in relays ] - entities.append( - DoorBirdButton(doorstation, doorstation_info, IR_RELAY, IR_ENTITY_DESCRIPTION) - ) + entities.append(DoorBirdButton(door_bird_data, IR_RELAY, IR_ENTITY_DESCRIPTION)) async_add_entities(entities) @@ -74,16 +69,14 @@ class DoorBirdButton(DoorBirdEntity, ButtonEntity): def __init__( self, - doorstation: DoorBird, - doorstation_info, + door_bird_data: DoorBirdData, relay: str, entity_description: DoorbirdButtonEntityDescription, ) -> None: """Initialize a relay in a DoorBird device.""" - super().__init__(doorstation, doorstation_info) + super().__init__(door_bird_data) self._relay = relay self.entity_description = entity_description - if self._relay == IR_RELAY: self._attr_name = "IR" else: @@ -92,4 +85,4 @@ class DoorBirdButton(DoorBirdEntity, ButtonEntity): def press(self) -> None: """Power the relay.""" - self.entity_description.press_action(self._doorstation.device, self._relay) + self.entity_description.press_action(self._door_station.device, self._relay) diff --git a/homeassistant/components/doorbird/camera.py b/homeassistant/components/doorbird/camera.py index 63eb646972d..06bdb494463 100644 --- a/homeassistant/components/doorbird/camera.py +++ b/homeassistant/components/doorbird/camera.py @@ -14,13 +14,9 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.entity_platform import AddEntitiesCallback import homeassistant.util.dt as dt_util -from .const import ( - DOMAIN, - DOOR_STATION, - DOOR_STATION_EVENT_ENTITY_IDS, - DOOR_STATION_INFO, -) +from .const import DOMAIN from .entity import DoorBirdEntity +from .models import DoorBirdData _LAST_VISITOR_INTERVAL = datetime.timedelta(minutes=2) _LAST_MOTION_INTERVAL = datetime.timedelta(seconds=30) @@ -36,39 +32,31 @@ async def async_setup_entry( ) -> None: """Set up the DoorBird camera platform.""" config_entry_id = config_entry.entry_id - config_data = hass.data[DOMAIN][config_entry_id] - doorstation = config_data[DOOR_STATION] - doorstation_info = config_data[DOOR_STATION_INFO] - device = doorstation.device + door_bird_data: DoorBirdData = hass.data[DOMAIN][config_entry_id] + device = door_bird_data.door_station.device async_add_entities( [ DoorBirdCamera( - doorstation, - doorstation_info, + door_bird_data, device.live_image_url, "live", "live", - doorstation.doorstation_events, _LIVE_INTERVAL, device.rtsp_live_video_url, ), DoorBirdCamera( - doorstation, - doorstation_info, + door_bird_data, device.history_image_url(1, "doorbell"), "last_ring", "last_ring", - [], _LAST_VISITOR_INTERVAL, ), DoorBirdCamera( - doorstation, - doorstation_info, + door_bird_data, device.history_image_url(1, "motionsensor"), "last_motion", "last_motion", - [], _LAST_MOTION_INTERVAL, ), ] @@ -80,17 +68,15 @@ class DoorBirdCamera(DoorBirdEntity, Camera): def __init__( self, - doorstation, - doorstation_info, - url, - camera_id, - translation_key, - doorstation_events, - interval, - stream_url=None, + door_bird_data: DoorBirdData, + url: str, + camera_id: str, + translation_key: str, + interval: datetime.timedelta, + stream_url: str | None = None, ) -> None: """Initialize the camera on a DoorBird device.""" - super().__init__(doorstation, doorstation_info) + super().__init__(door_bird_data) self._url = url self._stream_url = stream_url self._attr_translation_key = translation_key @@ -100,7 +86,6 @@ class DoorBirdCamera(DoorBirdEntity, Camera): self._interval = interval self._last_update = datetime.datetime.min self._attr_unique_id = f"{self._mac_addr}_{camera_id}" - self._doorstation_events = doorstation_events async def stream_source(self): """Return the stream source.""" @@ -133,19 +118,15 @@ class DoorBirdCamera(DoorBirdEntity, Camera): return self._last_image async def async_added_to_hass(self) -> None: - """Add callback after being added to hass. - - Registers entity_id map for the logbook - """ - event_to_entity_id = self.hass.data[DOMAIN].setdefault( - DOOR_STATION_EVENT_ENTITY_IDS, {} - ) - for event in self._doorstation_events: + """Subscribe to events.""" + await super().async_added_to_hass() + event_to_entity_id = self._door_bird_data.event_entity_ids + for event in self._door_station.events: event_to_entity_id[event] = self.entity_id - async def will_remove_from_hass(self): - """Unregister entity_id map for the logbook.""" - event_to_entity_id = self.hass.data[DOMAIN][DOOR_STATION_EVENT_ENTITY_IDS] - for event in self._doorstation_events: - if event in event_to_entity_id: - del event_to_entity_id[event] + async def async_will_remove_from_hass(self) -> None: + """Unsubscribe from events.""" + event_to_entity_id = self._door_bird_data.event_entity_ids + for event in self._door_station.events: + del event_to_entity_id[event] + await super().async_will_remove_from_hass() diff --git a/homeassistant/components/doorbird/config_flow.py b/homeassistant/components/doorbird/config_flow.py index 4ad5e24247e..d2197de93c9 100644 --- a/homeassistant/components/doorbird/config_flow.py +++ b/homeassistant/components/doorbird/config_flow.py @@ -4,6 +4,7 @@ from __future__ import annotations from http import HTTPStatus from ipaddress import ip_address import logging +from typing import Any from doorbirdpy import DoorBird import requests @@ -12,12 +13,12 @@ import voluptuous as vol from homeassistant import config_entries, core, exceptions from homeassistant.components import zeroconf from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME -from homeassistant.core import callback +from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResult from homeassistant.util.network import is_ipv4_address, is_link_local from .const import CONF_EVENTS, DOMAIN, DOORBIRD_OUI -from .util import get_mac_address_from_doorstation_info +from .util import get_mac_address_from_door_station_info _LOGGER = logging.getLogger(__name__) @@ -33,7 +34,7 @@ def _schema_with_defaults(host=None, name=None): ) -def _check_device(device): +def _check_device(device: DoorBird) -> tuple[tuple[bool, int], dict[str, Any]]: """Verify we can connect to the device and return the status.""" return device.ready(), device.info() @@ -53,13 +54,13 @@ async def validate_input(hass: core.HomeAssistant, data): if not status[0]: raise CannotConnect - mac_addr = get_mac_address_from_doorstation_info(info) + mac_addr = get_mac_address_from_door_station_info(info) # Return info that you want to store in the config entry. return {"title": data[CONF_HOST], "mac_addr": mac_addr} -async def async_verify_supported_device(hass, host): +async def async_verify_supported_device(hass: HomeAssistant, host: str) -> bool: """Verify the doorbell state endpoint returns a 401.""" device = DoorBird(host, "", "") try: diff --git a/homeassistant/components/doorbird/device.py b/homeassistant/components/doorbird/device.py index 1c787feb934..aced0d8723f 100644 --- a/homeassistant/components/doorbird/device.py +++ b/homeassistant/components/doorbird/device.py @@ -6,6 +6,7 @@ from typing import Any from doorbirdpy import DoorBird +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 @@ -19,20 +20,28 @@ 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 + self, + device: DoorBird, + name: str | None, + custom_url: str | None, + token: str, + event_entity_ids: dict[str, 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 + self._event_entity_ids = event_entity_ids + self.events: list[str] = [] + self.door_station_events: list[str] = [] - def update_events(self, events): + def update_events(self, events: list[str]) -> None: """Update the doorbird events.""" self.events = events - self.doorstation_events = [self._get_event_name(event) for event in self.events] + self.door_station_events = [ + self._get_event_name(event) for event in self.events + ] @property def name(self) -> str | None: @@ -63,12 +72,12 @@ class ConfiguredDoorBird: if self.custom_url is not None: hass_url = self.custom_url - if not self.doorstation_events: + if not self.door_station_events: # User may not have permission to get the favorites return favorites = self.device.favorites() - for event in self.doorstation_events: + for event in self.door_station_events: if self._register_event(hass_url, event, favs=favorites): _LOGGER.info( "Successfully registered URL for %s on %s", event, self.name @@ -126,7 +135,7 @@ class ConfiguredDoorBird: return None - def get_event_data(self) -> dict[str, str]: + 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(), @@ -134,4 +143,5 @@ class ConfiguredDoorBird: "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, + ATTR_ENTITY_ID: self._event_entity_ids.get(event), } diff --git a/homeassistant/components/doorbird/entity.py b/homeassistant/components/doorbird/entity.py index 32c9cfff784..4360a8ff490 100644 --- a/homeassistant/components/doorbird/entity.py +++ b/homeassistant/components/doorbird/entity.py @@ -1,6 +1,5 @@ """The DoorBird integration base entity.""" -from typing import Any from homeassistant.helpers import device_registry as dr from homeassistant.helpers.device_registry import DeviceInfo @@ -12,8 +11,8 @@ from .const import ( DOORBIRD_INFO_KEY_FIRMWARE, MANUFACTURER, ) -from .device import ConfiguredDoorBird -from .util import get_mac_address_from_doorstation_info +from .models import DoorBirdData +from .util import get_mac_address_from_door_station_info class DoorBirdEntity(Entity): @@ -21,21 +20,20 @@ class DoorBirdEntity(Entity): _attr_has_entity_name = True - def __init__( - self, doorstation: ConfiguredDoorBird, doorstation_info: dict[str, Any] - ) -> None: + def __init__(self, door_bird_data: DoorBirdData) -> None: """Initialize the entity.""" super().__init__() - self._doorstation = doorstation - self._mac_addr = get_mac_address_from_doorstation_info(doorstation_info) - - firmware = doorstation_info[DOORBIRD_INFO_KEY_FIRMWARE] - firmware_build = doorstation_info[DOORBIRD_INFO_KEY_BUILD_NUMBER] + self._door_bird_data = door_bird_data + self._door_station = door_bird_data.door_station + door_station_info = door_bird_data.door_station_info + self._mac_addr = get_mac_address_from_door_station_info(door_station_info) + firmware = door_station_info[DOORBIRD_INFO_KEY_FIRMWARE] + firmware_build = door_station_info[DOORBIRD_INFO_KEY_BUILD_NUMBER] self._attr_device_info = DeviceInfo( configuration_url="https://webadmin.doorbird.com/", connections={(dr.CONNECTION_NETWORK_MAC, self._mac_addr)}, manufacturer=MANUFACTURER, - model=doorstation_info[DOORBIRD_INFO_KEY_DEVICE_TYPE], - name=self._doorstation.name, + model=door_station_info[DOORBIRD_INFO_KEY_DEVICE_TYPE], + name=self._door_station.name, sw_version=f"{firmware} {firmware_build}", ) diff --git a/homeassistant/components/doorbird/logbook.py b/homeassistant/components/doorbird/logbook.py index f3beebe6971..7c8e3cd3c51 100644 --- a/homeassistant/components/doorbird/logbook.py +++ b/homeassistant/components/doorbird/logbook.py @@ -1,43 +1,35 @@ """Describe logbook events.""" from __future__ import annotations -from typing import Any - from homeassistant.components.logbook import ( LOGBOOK_ENTRY_ENTITY_ID, LOGBOOK_ENTRY_MESSAGE, LOGBOOK_ENTRY_NAME, ) from homeassistant.const import ATTR_ENTITY_ID -from homeassistant.core import callback +from homeassistant.core import Event, HomeAssistant, callback -from .const import DOMAIN, DOOR_STATION, DOOR_STATION_EVENT_ENTITY_IDS +from .const import DOMAIN +from .models import DoorBirdData @callback -def async_describe_events(hass, async_describe_event): +def async_describe_events(hass: HomeAssistant, async_describe_event): """Describe logbook events.""" @callback - def async_describe_logbook_event(event): + def async_describe_logbook_event(event: Event): """Describe a logbook event.""" - doorbird_event = event.event_type.split("_", 1)[1] - return { LOGBOOK_ENTRY_NAME: "Doorbird", LOGBOOK_ENTRY_MESSAGE: f"Event {event.event_type} was fired", - LOGBOOK_ENTRY_ENTITY_ID: hass.data[DOMAIN][ - DOOR_STATION_EVENT_ENTITY_IDS - ].get(doorbird_event, event.data.get(ATTR_ENTITY_ID)), + # Database entries before Jun 25th 2020 will not have an entity ID + LOGBOOK_ENTRY_ENTITY_ID: event.data.get(ATTR_ENTITY_ID), } - domain_data: dict[str, Any] = hass.data[DOMAIN] - + domain_data: dict[str, DoorBirdData] = hass.data[DOMAIN] for data in domain_data.values(): - if DOOR_STATION not in data: - # We need to skip door_station_event_entity_ids - continue - for event in data[DOOR_STATION].doorstation_events: + for event in data.door_station.door_station_events: async_describe_event( DOMAIN, f"{DOMAIN}_{event}", async_describe_logbook_event ) diff --git a/homeassistant/components/doorbird/models.py b/homeassistant/components/doorbird/models.py new file mode 100644 index 00000000000..f8fb8687e59 --- /dev/null +++ b/homeassistant/components/doorbird/models.py @@ -0,0 +1,26 @@ +"""The doorbird integration models.""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .device import ConfiguredDoorBird + + +@dataclass +class DoorBirdData: + """Data for the doorbird integration.""" + + door_station: ConfiguredDoorBird + door_station_info: dict[str, Any] + + # + # This integration uses a different event for + # each entity id. It would be a major breaking + # change to change this to a single event at this + # point. + # + # Do not copy this pattern in the future + # for any new integrations. + # + event_entity_ids: dict[str, str] diff --git a/homeassistant/components/doorbird/util.py b/homeassistant/components/doorbird/util.py index 7b406bc07fa..52c1417a67c 100644 --- a/homeassistant/components/doorbird/util.py +++ b/homeassistant/components/doorbird/util.py @@ -2,50 +2,23 @@ from homeassistant.core import HomeAssistant -from .const import DOMAIN, DOOR_STATION +from .const import DOMAIN from .device import ConfiguredDoorBird +from .models import DoorBirdData -def get_mac_address_from_doorstation_info(doorstation_info): +def get_mac_address_from_door_station_info(door_station_info): """Get the mac address depending on the device type.""" - if "PRIMARY_MAC_ADDR" in doorstation_info: - return doorstation_info["PRIMARY_MAC_ADDR"] - return doorstation_info["WIFI_MAC_ADDR"] + return door_station_info.get("PRIMARY_MAC_ADDR", door_station_info["WIFI_MAC_ADDR"]) -def get_doorstation_by_token( +def get_door_station_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: HomeAssistant, slug: str -) -> ConfiguredDoorBird | None: - """Get doorstation by slug.""" - return _get_doorstation_by_attr(hass, "slug", slug) - - -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 - - doorstation = entry[DOOR_STATION] - - if getattr(doorstation, attr) == val: - return doorstation - + """Get door station by token.""" + domain_data: dict[str, DoorBirdData] = hass.data[DOMAIN] + for data in domain_data.values(): + door_station = data.door_station + if door_station.token == token: + return door_station return None - - -def get_all_doorstations(hass: HomeAssistant) -> list[ConfiguredDoorBird]: - """Get all doorstations.""" - return [ - entry[DOOR_STATION] - for entry in hass.data[DOMAIN].values() - if DOOR_STATION in entry - ]