Skip to content

Commit

Permalink
Add strict typing to ring integration (home-assistant#115276)
Browse files Browse the repository at this point in the history
  • Loading branch information
sdb9696 authored Apr 11, 2024
1 parent 3546ca3 commit 6954fcc
Show file tree
Hide file tree
Showing 16 changed files with 391 additions and 366 deletions.
1 change: 1 addition & 0 deletions .strict-typing
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,7 @@ homeassistant.components.rest_command.*
homeassistant.components.rfxtrx.*
homeassistant.components.rhasspy.*
homeassistant.components.ridwell.*
homeassistant.components.ring.*
homeassistant.components.rituals_perfume_genie.*
homeassistant.components.roku.*
homeassistant.components.romy.*
Expand Down
47 changes: 27 additions & 20 deletions homeassistant/components/ring/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,34 +2,39 @@

from __future__ import annotations

from dataclasses import dataclass
from functools import partial
import logging
from typing import Any, cast

from ring_doorbell import Auth, Ring
from ring_doorbell import Auth, Ring, RingDevices

from homeassistant.config_entries import ConfigEntry
from homeassistant.const import APPLICATION_NAME, CONF_TOKEN, __version__
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue

from .const import (
DOMAIN,
PLATFORMS,
RING_API,
RING_DEVICES,
RING_DEVICES_COORDINATOR,
RING_NOTIFICATIONS_COORDINATOR,
)
from .const import DOMAIN, PLATFORMS
from .coordinator import RingDataCoordinator, RingNotificationsCoordinator

_LOGGER = logging.getLogger(__name__)


@dataclass
class RingData:
"""Class to support type hinting of ring data collection."""

api: Ring
devices: RingDevices
devices_coordinator: RingDataCoordinator
notifications_coordinator: RingNotificationsCoordinator


async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""

def token_updater(token):
def token_updater(token: dict[str, Any]) -> None:
"""Handle from sync context when token is updated."""
hass.loop.call_soon_threadsafe(
partial(
Expand All @@ -51,12 +56,12 @@ def token_updater(token):
await devices_coordinator.async_config_entry_first_refresh()
await notifications_coordinator.async_config_entry_first_refresh()

hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
RING_API: ring,
RING_DEVICES: ring.devices(),
RING_DEVICES_COORDINATOR: devices_coordinator,
RING_NOTIFICATIONS_COORDINATOR: notifications_coordinator,
}
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = RingData(
api=ring,
devices=ring.devices(),
devices_coordinator=devices_coordinator,
notifications_coordinator=notifications_coordinator,
)

await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

Expand All @@ -83,8 +88,9 @@ async def async_refresh_all(_: ServiceCall) -> None:
)

for info in hass.data[DOMAIN].values():
await info[RING_DEVICES_COORDINATOR].async_refresh()
await info[RING_NOTIFICATIONS_COORDINATOR].async_refresh()
ring_data = cast(RingData, info)
await ring_data.devices_coordinator.async_refresh()
await ring_data.notifications_coordinator.async_refresh()

# register service
hass.services.async_register(DOMAIN, "update", async_refresh_all)
Expand Down Expand Up @@ -121,8 +127,9 @@ async def _migrate_old_unique_ids(hass: HomeAssistant, entry_id: str) -> None:
@callback
def _async_migrator(entity_entry: er.RegistryEntry) -> dict[str, str] | None:
# Old format for camera and light was int
if isinstance(entity_entry.unique_id, int):
new_unique_id = str(entity_entry.unique_id)
unique_id = cast(str | int, entity_entry.unique_id)
if isinstance(unique_id, int):
new_unique_id = str(unique_id)
if existing_entity_id := entity_registry.async_get_entity_id(
entity_entry.domain, entity_entry.platform, new_unique_id
):
Expand Down
62 changes: 36 additions & 26 deletions homeassistant/components/ring/binary_sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,13 @@

from __future__ import annotations

from collections.abc import Callable, Mapping
from dataclasses import dataclass
from datetime import datetime
from typing import Any

from ring_doorbell import Ring, RingEvent, RingGeneric

from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
Expand All @@ -15,29 +18,32 @@
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from .const import DOMAIN, RING_API, RING_DEVICES, RING_NOTIFICATIONS_COORDINATOR
from . import RingData
from .const import DOMAIN
from .coordinator import RingNotificationsCoordinator
from .entity import RingEntity
from .entity import RingBaseEntity


@dataclass(frozen=True, kw_only=True)
class RingBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Describes Ring binary sensor entity."""

category: list[str]
exists_fn: Callable[[RingGeneric], bool]


BINARY_SENSOR_TYPES: tuple[RingBinarySensorEntityDescription, ...] = (
RingBinarySensorEntityDescription(
key="ding",
translation_key="ding",
category=["doorbots", "authorized_doorbots", "other"],
device_class=BinarySensorDeviceClass.OCCUPANCY,
exists_fn=lambda device: device.family
in {"doorbots", "authorized_doorbots", "other"},
),
RingBinarySensorEntityDescription(
key="motion",
category=["doorbots", "authorized_doorbots", "stickup_cams"],
device_class=BinarySensorDeviceClass.MOTION,
exists_fn=lambda device: device.family
in {"doorbots", "authorized_doorbots", "stickup_cams"},
),
)

Expand All @@ -48,34 +54,36 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Ring binary sensors from a config entry."""
ring = hass.data[DOMAIN][config_entry.entry_id][RING_API]
devices = hass.data[DOMAIN][config_entry.entry_id][RING_DEVICES]
notifications_coordinator: RingNotificationsCoordinator = hass.data[DOMAIN][
config_entry.entry_id
][RING_NOTIFICATIONS_COORDINATOR]
ring_data: RingData = hass.data[DOMAIN][config_entry.entry_id]

entities = [
RingBinarySensor(ring, device, notifications_coordinator, description)
for device_type in ("doorbots", "authorized_doorbots", "stickup_cams", "other")
RingBinarySensor(
ring_data.api,
device,
ring_data.notifications_coordinator,
description,
)
for description in BINARY_SENSOR_TYPES
if device_type in description.category
for device in devices[device_type]
for device in ring_data.devices.all_devices
if description.exists_fn(device)
]

async_add_entities(entities)


class RingBinarySensor(RingEntity, BinarySensorEntity):
class RingBinarySensor(
RingBaseEntity[RingNotificationsCoordinator], BinarySensorEntity
):
"""A binary sensor implementation for Ring device."""

_active_alert: dict[str, Any] | None = None
_active_alert: RingEvent | None = None
entity_description: RingBinarySensorEntityDescription

def __init__(
self,
ring,
device,
coordinator,
ring: Ring,
device: RingGeneric,
coordinator: RingNotificationsCoordinator,
description: RingBinarySensorEntityDescription,
) -> None:
"""Initialize a sensor for Ring device."""
Expand All @@ -89,13 +97,13 @@ def __init__(
self._update_alert()

@callback
def _handle_coordinator_update(self, _=None):
def _handle_coordinator_update(self, _: Any = None) -> None:
"""Call update method."""
self._update_alert()
super()._handle_coordinator_update()

@callback
def _update_alert(self):
def _update_alert(self) -> None:
"""Update active alert."""
self._active_alert = next(
(
Expand All @@ -108,21 +116,23 @@ def _update_alert(self):
)

@property
def is_on(self):
def is_on(self) -> bool:
"""Return True if the binary sensor is on."""
return self._active_alert is not None

@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return the state attributes."""
attrs = super().extra_state_attributes

if self._active_alert is None:
return attrs

assert isinstance(attrs, dict)
attrs["state"] = self._active_alert["state"]
attrs["expires_at"] = datetime.fromtimestamp(
self._active_alert.get("now") + self._active_alert.get("expires_in")
).isoformat()
now = self._active_alert.get("now")
expires_in = self._active_alert.get("expires_in")
assert now and expires_in
attrs["expires_at"] = datetime.fromtimestamp(now + expires_in).isoformat()

return attrs
21 changes: 12 additions & 9 deletions homeassistant/components/ring/button.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,15 @@

from __future__ import annotations

from ring_doorbell import RingOther

from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from .const import DOMAIN, RING_DEVICES, RING_DEVICES_COORDINATOR
from . import RingData
from .const import DOMAIN
from .coordinator import RingDataCoordinator
from .entity import RingEntity, exception_wrap

Expand All @@ -22,25 +25,25 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Create the buttons for the Ring devices."""
devices = hass.data[DOMAIN][config_entry.entry_id][RING_DEVICES]
devices_coordinator: RingDataCoordinator = hass.data[DOMAIN][config_entry.entry_id][
RING_DEVICES_COORDINATOR
]
ring_data: RingData = hass.data[DOMAIN][config_entry.entry_id]
devices_coordinator = ring_data.devices_coordinator

async_add_entities(
RingDoorButton(device, devices_coordinator, BUTTON_DESCRIPTION)
for device in devices["other"]
for device in ring_data.devices.other
if device.has_capability("open")
)


class RingDoorButton(RingEntity, ButtonEntity):
"""Creates a button to open the ring intercom door."""

_device: RingOther

def __init__(
self,
device,
coordinator,
device: RingOther,
coordinator: RingDataCoordinator,
description: ButtonEntityDescription,
) -> None:
"""Initialize the button."""
Expand All @@ -52,6 +55,6 @@ def __init__(
self._attr_unique_id = f"{device.id}-{description.key}"

@exception_wrap
def press(self):
def press(self) -> None:
"""Open the door."""
self._device.open_door()
Loading

0 comments on commit 6954fcc

Please sign in to comment.