jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Additional doorbird cleanups to prepare for event entities (#98542)

Browse Source
This commit is contained in:
J. Nick Koston 2023-08-17 09:37:54 -05:00 committed by GitHub
parent 2d4decc9b1
commit 30a88e9e61
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 158 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
homeassistant/components/doorbird/__init__.py
View File

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

25
homeassistant/components/doorbird/button.py
View File

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

67
homeassistant/components/doorbird/camera.py
View File

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

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

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

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

@ -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),
}

24
homeassistant/components/doorbird/entity.py
View File

@ -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}",
)

26
homeassistant/components/doorbird/logbook.py
View File

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

26
homeassistant/components/doorbird/models.py Normal file
View File

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

49
homeassistant/components/doorbird/util.py
View File

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