Add some typing to doorbird (#98483)
J. Nick Koston
GitHub
parent
b680bca5e9
commit
b083f5bf89
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
||||
|
|
|
|||
|
|
@ -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}"
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
Loading…
Reference in New Issue