Migrate ring siren and switch platforms to entity descriptions (#125775)

homeassistant/components/ring/entity.py

@@ -1,6 +1,6 @@
 """Base class for Ring entity."""
 
-from collections.abc import Callable, Coroutine
+from collections.abc import Awaitable, Callable, Coroutine
 from dataclasses import dataclass
 from typing import Any, Concatenate, Generic, cast
 
@@ -76,6 +76,19 @@ def exception_wrap[_RingBaseEntityT: RingBaseEntity[Any, Any], **_P, _R](
     return _wrap
 
 
+def refresh_after[_RingEntityT: RingEntity[Any], **_P](
+    func: Callable[Concatenate[_RingEntityT, _P], Awaitable[None]],
+) -> Callable[Concatenate[_RingEntityT, _P], Coroutine[Any, Any, None]]:
+    """Define a wrapper to handle api call errors or refresh after success."""
+
+    @exception_wrap
+    async def _wrap(self: _RingEntityT, *args: _P.args, **kwargs: _P.kwargs) -> None:
+        await func(self, *args, **kwargs)
+        await self.coordinator.async_request_refresh()
+
+    return _wrap
+
+
 def async_check_create_deprecated(
     hass: HomeAssistant,
     platform: Platform,

homeassistant/components/ring/siren.py

@@ -1,21 +1,69 @@
 """Component providing HA Siren support for Ring Chimes."""
 
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
 import logging
-from typing import Any
+from typing import Any, Generic, cast
 
-from ring_doorbell import RingChime, RingEventKind
+from ring_doorbell import RingChime, RingEventKind, RingGeneric
 
-from homeassistant.components.siren import ATTR_TONE, SirenEntity, SirenEntityFeature
+from homeassistant.components.siren import (
-from homeassistant.core import HomeAssistant
+    ATTR_TONE,
+    SirenEntity,
+    SirenEntityDescription,
+    SirenEntityFeature,
+    SirenTurnOnServiceParameters,
+)
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from . import RingConfigEntry
 from .coordinator import RingDataCoordinator
-from .entity import RingEntity, exception_wrap
+from .entity import (
+    RingDeviceT,
+    RingEntity,
+    RingEntityDescription,
+    async_check_create_deprecated,
+    refresh_after,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True, kw_only=True)
+class RingSirenEntityDescription(
+    SirenEntityDescription, RingEntityDescription, Generic[RingDeviceT]
+):
+    """Describes a Ring siren entity."""
+
+    exists_fn: Callable[[RingGeneric], bool]
+    unique_id_fn: Callable[[RingDeviceT], str] = lambda device: str(
+        device.device_api_id
+    )
+    is_on_fn: Callable[[RingDeviceT], bool] | None = None
+    turn_on_fn: (
+        Callable[[RingDeviceT, SirenTurnOnServiceParameters], Coroutine[Any, Any, Any]]
+        | None
+    ) = None
+    turn_off_fn: Callable[[RingDeviceT], Coroutine[Any, Any, None]] | None = None
+
+
+SIRENS: tuple[RingSirenEntityDescription[Any], ...] = (
+    RingSirenEntityDescription[RingChime](
+        key="siren",
+        translation_key="siren",
+        available_tones=[RingEventKind.DING.value, RingEventKind.MOTION.value],
+        # Historically the chime siren entity has appended `siren` to the unique id
+        unique_id_fn=lambda device: f"{device.device_api_id}-siren",
+        exists_fn=lambda device: isinstance(device, RingChime),
+        turn_on_fn=lambda device, kwargs: device.async_test_sound(
+            kind=str(kwargs.get(ATTR_TONE) or "") or RingEventKind.DING.value
+        ),
+    ),
+)
+
+
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: RingConfigEntry,
@@ -26,27 +74,74 @@ async def async_setup_entry(
     devices_coordinator = ring_data.devices_coordinator
 
     async_add_entities(
-        RingChimeSiren(device, devices_coordinator)
+        RingSiren(device, devices_coordinator, description)
-        for device in ring_data.devices.chimes
+        for device in ring_data.devices.all_devices
+        for description in SIRENS
+        if description.exists_fn(device)
+        and async_check_create_deprecated(
+            hass,
+            Platform.SIREN,
+            description.unique_id_fn(device),
+            description,
+        )
     )
 
 
-class RingChimeSiren(RingEntity[RingChime], SirenEntity):
+class RingSiren(RingEntity[RingDeviceT], SirenEntity):
     """Creates a siren to play the test chimes of a Chime device."""
 
-    _attr_available_tones = [RingEventKind.DING.value, RingEventKind.MOTION.value]
+    entity_description: RingSirenEntityDescription[RingDeviceT]
-    _attr_supported_features = SirenEntityFeature.TURN_ON | SirenEntityFeature.TONES
-    _attr_translation_key = "siren"
 
-    def __init__(self, device: RingChime, coordinator: RingDataCoordinator) -> None:
+    def __init__(
+        self,
+        device: RingDeviceT,
+        coordinator: RingDataCoordinator,
+        description: RingSirenEntityDescription[RingDeviceT],
+    ) -> None:
         """Initialize a Ring Chime siren."""
         super().__init__(device, coordinator)
-        # Entity class attributes
+        self.entity_description = description
-        self._attr_unique_id = f"{self._device.id}-siren"
+        self._attr_unique_id = description.unique_id_fn(device)
+        if description.is_on_fn:
+            self._attr_is_on = description.is_on_fn(self._device)
+        features = SirenEntityFeature(0)
+        if description.turn_on_fn:
+            features = features | SirenEntityFeature.TURN_ON
+        if description.turn_off_fn:
+            features = features | SirenEntityFeature.TURN_OFF
+        if description.available_tones:
+            features = features | SirenEntityFeature.TONES
+        self._attr_supported_features = features
 
-    @exception_wrap
+    async def _async_set_siren(self, siren_on: bool, **kwargs: Any) -> None:
+        if siren_on and self.entity_description.turn_on_fn:
+            turn_on_params = cast(SirenTurnOnServiceParameters, kwargs)
+            await self.entity_description.turn_on_fn(self._device, turn_on_params)
+        elif not siren_on and self.entity_description.turn_off_fn:
+            await self.entity_description.turn_off_fn(self._device)
+
+        if self.entity_description.is_on_fn:
+            self._attr_is_on = siren_on
+            self.async_write_ha_state()
+
+    @refresh_after
     async def async_turn_on(self, **kwargs: Any) -> None:
-        """Play the test sound on a Ring Chime device."""
+        """Turn on the siren."""
-        tone = kwargs.get(ATTR_TONE) or RingEventKind.DING.value
+        await self._async_set_siren(True, **kwargs)
 
-        await self._device.async_test_sound(kind=tone)
+    @refresh_after
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn off the siren."""
+        await self._async_set_siren(False)
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Call update method."""
+        if not self.entity_description.is_on_fn:
+            return
+        self._device = cast(
+            RingDeviceT,
+            self._get_coordinator_data().get_device(self._device.device_api_id),
+        )
+        self._attr_is_on = self.entity_description.is_on_fn(self._device)
+        super()._handle_coordinator_update()

homeassistant/components/ring/switch.py

@@ -1,29 +1,56 @@
 """Component providing HA switch support for Ring Door Bell/Chimes."""
 
-from datetime import timedelta
+from collections.abc import Callable, Coroutine, Sequence
+from dataclasses import dataclass
 import logging
-from typing import Any
+from typing import Any, Generic, Self, cast
 
-from ring_doorbell import RingStickUpCam
+from ring_doorbell import RingCapability, RingStickUpCam
 
-from homeassistant.components.switch import SwitchEntity
+from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
+from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 import homeassistant.util.dt as dt_util
 
 from . import RingConfigEntry
 from .coordinator import RingDataCoordinator
-from .entity import RingEntity, exception_wrap
+from .entity import (
+    RingDeviceT,
+    RingEntity,
+    RingEntityDescription,
+    async_check_create_deprecated,
+    refresh_after,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
 
-# It takes a few seconds for the API to correctly return an update indicating
+@dataclass(frozen=True, kw_only=True)
-# that the changes have been made. Once we request a change (i.e. a light
+class RingSwitchEntityDescription(
-# being turned on) we simply wait for this time delta before we allow
+    SwitchEntityDescription, RingEntityDescription, Generic[RingDeviceT]
-# updates to take place.
+):
+    """Describes a Ring switch entity."""
 
-SKIP_UPDATES_DELAY = timedelta(seconds=5)
+    exists_fn: Callable[[RingDeviceT], bool]
+    unique_id_fn: Callable[[Self, RingDeviceT], str] = (
+        lambda self, device: f"{device.device_api_id}-{self.key}"
+    )
+    is_on_fn: Callable[[RingDeviceT], bool]
+    turn_on_fn: Callable[[RingDeviceT], Coroutine[Any, Any, None]]
+    turn_off_fn: Callable[[RingDeviceT], Coroutine[Any, Any, None]]
+
+
+SWITCHES: Sequence[RingSwitchEntityDescription[Any]] = (
+    RingSwitchEntityDescription[RingStickUpCam](
+        key="siren",
+        translation_key="siren",
+        exists_fn=lambda device: device.has_capability(RingCapability.SIREN),
+        is_on_fn=lambda device: device.siren > 0,
+        turn_on_fn=lambda device: device.async_set_siren(1),
+        turn_off_fn=lambda device: device.async_set_siren(0),
+    ),
+)
 
 
 async def async_setup_entry(
@@ -36,61 +63,62 @@ async def async_setup_entry(
     devices_coordinator = ring_data.devices_coordinator
 
     async_add_entities(
-        SirenSwitch(device, devices_coordinator)
+        RingSwitch(device, devices_coordinator, description)
-        for device in ring_data.devices.stickup_cams
+        for description in SWITCHES
-        if device.has_capability("siren")
+        for device in ring_data.devices.all_devices
+        if description.exists_fn(device)
+        and async_check_create_deprecated(
+            hass,
+            Platform.SWITCH,
+            description.unique_id_fn(description, device),
+            description,
+        )
     )
 
 
-class BaseRingSwitch(RingEntity[RingStickUpCam], SwitchEntity):
+class RingSwitch(RingEntity[RingDeviceT], SwitchEntity):
     """Represents a switch for controlling an aspect of a ring device."""
 
+    entity_description: RingSwitchEntityDescription[RingDeviceT]
+
     def __init__(
-        self, device: RingStickUpCam, coordinator: RingDataCoordinator, device_type: str
+        self,
+        device: RingDeviceT,
+        coordinator: RingDataCoordinator,
+        description: RingSwitchEntityDescription[RingDeviceT],
     ) -> None:
         """Initialize the switch."""
         super().__init__(device, coordinator)
-        self._device_type = device_type
+        self.entity_description = description
-        self._attr_unique_id = f"{self._device.id}-{self._device_type}"
-
-
-class SirenSwitch(BaseRingSwitch):
-    """Creates a switch to turn the ring cameras siren on and off."""
-
-    _attr_translation_key = "siren"
-
-    def __init__(
-        self, device: RingStickUpCam, coordinator: RingDataCoordinator
-    ) -> None:
-        """Initialize the switch for a device with a siren."""
-        super().__init__(device, coordinator, "siren")
         self._no_updates_until = dt_util.utcnow()
-        self._attr_is_on = device.siren > 0
+        self._attr_unique_id = description.unique_id_fn(description, device)
+        self._attr_is_on = description.is_on_fn(device)
 
     @callback
     def _handle_coordinator_update(self) -> None:
         """Call update method."""
-        if self._no_updates_until > dt_util.utcnow():
+        self._device = cast(
-            return
+            RingDeviceT,
-        device = self._get_coordinator_data().get_stickup_cam(
+            self._get_coordinator_data().get_device(self._device.device_api_id),
-            self._device.device_api_id
         )
-        self._attr_is_on = device.siren > 0
+        self._attr_is_on = self.entity_description.is_on_fn(self._device)
         super()._handle_coordinator_update()
 
-    @exception_wrap
+    @refresh_after
-    async def _async_set_switch(self, new_state: int) -> None:
+    async def _async_set_switch(self, switch_on: bool) -> None:
         """Update switch state, and causes Home Assistant to correctly update."""
-        await self._device.async_set_siren(new_state)
+        if switch_on:
+            await self.entity_description.turn_on_fn(self._device)
+        else:
+            await self.entity_description.turn_off_fn(self._device)
 
-        self._attr_is_on = new_state > 0
+        self._attr_is_on = switch_on
-        self._no_updates_until = dt_util.utcnow() + SKIP_UPDATES_DELAY
         self.async_write_ha_state()
 
     async def async_turn_on(self, **kwargs: Any) -> None:
         """Turn the siren on for 30 seconds."""
-        await self._async_set_switch(1)
+        await self._async_set_switch(True)
 
     async def async_turn_off(self, **kwargs: Any) -> None:
         """Turn the siren off."""
-        await self._async_set_switch(0)
+        await self._async_set_switch(False)

tests/components/ring/device_mocks.py

@@ -158,6 +158,9 @@ def _mocked_ring_device(device_dict, device_family, device_class, capabilities):
         mock_device.configure_mock(
             siren=device_dict["siren_status"].get("seconds_remaining")
         )
+        mock_device.async_set_siren.side_effect = lambda i: mock_device.configure_mock(
+            siren=i
+        )
 
     if has_capability(RingCapability.BATTERY):
         mock_device.configure_mock(