Add siren platform for tplink (#124934)

* Add siren platform for tplink

* Add tests

* Add alarm to features.json

* Update based on reviews

* Use alarm module instead of individual features

---------

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/tplink/const.py

@@ -36,6 +36,7 @@ PLATFORMS: Final = [
     Platform.NUMBER,
     Platform.SELECT,
     Platform.SENSOR,
+    Platform.SIREN,
     Platform.SWITCH,
 ]
 

homeassistant/components/tplink/entity.py

@@ -70,6 +70,8 @@ EXCLUDED_FEATURES = {
     "available_firmware_version",
     "update_available",
     "check_latest_firmware",
+    # siren
+    "alarm",
 }
 
 

homeassistant/components/tplink/siren.py

@@ -0,0 +1,61 @@
+"""Support for TPLink hub alarm."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from kasa import Device, Module
+from kasa.smart.modules.alarm import Alarm
+
+from homeassistant.components.siren import SirenEntity, SirenEntityFeature
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import TPLinkConfigEntry
+from .coordinator import TPLinkDataUpdateCoordinator
+from .entity import CoordinatedTPLinkEntity, async_refresh_after
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: TPLinkConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up siren entities."""
+    data = config_entry.runtime_data
+    parent_coordinator = data.parent_coordinator
+    device = parent_coordinator.device
+
+    if Module.Alarm in device.modules:
+        async_add_entities([TPLinkSirenEntity(device, parent_coordinator)])
+
+
+class TPLinkSirenEntity(CoordinatedTPLinkEntity, SirenEntity):
+    """Representation of a tplink hub alarm."""
+
+    _attr_name = None
+    _attr_supported_features = SirenEntityFeature.TURN_OFF | SirenEntityFeature.TURN_ON
+
+    def __init__(
+        self,
+        device: Device,
+        coordinator: TPLinkDataUpdateCoordinator,
+    ) -> None:
+        """Initialize the siren entity."""
+        self._alarm_module: Alarm = device.modules[Module.Alarm]
+        super().__init__(device, coordinator)
+
+    @async_refresh_after
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn the siren on."""
+        await self._alarm_module.play()
+
+    @async_refresh_after
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn the siren off."""
+        await self._alarm_module.stop()
+
+    @callback
+    def _async_update_attrs(self) -> None:
+        """Update the entity's attributes."""
+        self._attr_is_on = self._alarm_module.active

tests/components/tplink/__init__.py

@@ -18,6 +18,7 @@ from kasa import (
 )
 from kasa.interfaces import Fan, Light, LightEffect, LightState
 from kasa.protocol import BaseProtocol
+from kasa.smart.modules.alarm import Alarm
 from syrupy import SnapshotAssertion
 
 from homeassistant.components.tplink import (
@@ -387,6 +388,15 @@ def _mocked_fan_module(effect) -> Fan:
     return fan
 
 
+def _mocked_alarm_module(device):
+    alarm = MagicMock(auto_spec=Alarm, name="Mocked alarm")
+    alarm.active = False
+    alarm.play = AsyncMock()
+    alarm.stop = AsyncMock()
+
+    return alarm
+
+
 def _mocked_strip_children(features=None, alias=None) -> list[Device]:
     plug0 = _mocked_device(
         alias="Plug0" if alias is None else alias,
@@ -453,6 +463,7 @@ MODULE_TO_MOCK_GEN = {
     Module.Light: _mocked_light_module,
     Module.LightEffect: _mocked_light_effect_module,
     Module.Fan: _mocked_fan_module,
+    Module.Alarm: _mocked_alarm_module,
 }
 
 

tests/components/tplink/fixtures/features.json

@@ -200,6 +200,11 @@
     "type": "BinarySensor",
     "category": "Primary"
   },
+  "alarm": {
+    "value": false,
+    "type": "BinarySensor",
+    "category": "Info"
+  },
   "test_alarm": {
     "value": "<Action>",
     "type": "Action",

tests/components/tplink/snapshots/test_siren.ambr

@@ -0,0 +1,84 @@
+# serializer version: 1
+# name: test_states[hub-entry]
+  DeviceRegistryEntrySnapshot({
+    'area_id': None,
+    'config_entries': <ANY>,
+    'configuration_url': None,
+    'connections': set({
+      tuple(
+        'mac',
+        'aa:bb:cc:dd:ee:ff',
+      ),
+    }),
+    'disabled_by': None,
+    'entry_type': None,
+    'hw_version': '1.0.0',
+    'id': <ANY>,
+    'identifiers': set({
+      tuple(
+        'tplink',
+        '123456789ABCDEFGH',
+      ),
+    }),
+    'is_new': False,
+    'labels': set({
+    }),
+    'manufacturer': 'TP-Link',
+    'model': 'HS100',
+    'model_id': None,
+    'name': 'hub',
+    'name_by_user': None,
+    'primary_config_entry': <ANY>,
+    'serial_number': None,
+    'suggested_area': None,
+    'sw_version': '1.0.0',
+    'via_device_id': None,
+  })
+# ---
+# name: test_states[siren.hub-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'siren',
+    'entity_category': None,
+    'entity_id': 'siren.hub',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': None,
+    'original_icon': None,
+    'original_name': None,
+    'platform': 'tplink',
+    'previous_unique_id': None,
+    'supported_features': <SirenEntityFeature: 3>,
+    'translation_key': None,
+    'unique_id': '123456789ABCDEFGH',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_states[siren.hub-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'friendly_name': 'hub',
+      'supported_features': <SirenEntityFeature: 3>,
+    }),
+    'context': <ANY>,
+    'entity_id': 'siren.hub',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'off',
+  })
+# ---

tests/components/tplink/test_siren.py

@@ -0,0 +1,76 @@
+"""Tests for siren platform."""
+
+from __future__ import annotations
+
+from kasa import Device, Module
+import pytest
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.components.siren import (
+    DOMAIN as SIREN_DOMAIN,
+    SERVICE_TURN_OFF,
+    SERVICE_TURN_ON,
+)
+from homeassistant.const import ATTR_ENTITY_ID, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr, entity_registry as er
+
+from . import _mocked_device, setup_platform_for_device, snapshot_platform
+
+from tests.common import MockConfigEntry
+
+ENTITY_ID = "siren.hub"
+
+
+@pytest.fixture
+async def mocked_hub(hass: HomeAssistant) -> Device:
+    """Return mocked tplink hub with an alarm module."""
+
+    return _mocked_device(
+        alias="hub",
+        modules=[Module.Alarm],
+        device_type=Device.Type.Hub,
+    )
+
+
+async def test_states(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    entity_registry: er.EntityRegistry,
+    device_registry: dr.DeviceRegistry,
+    snapshot: SnapshotAssertion,
+    mocked_hub: Device,
+) -> None:
+    """Snapshot test."""
+    await setup_platform_for_device(hass, mock_config_entry, Platform.SIREN, mocked_hub)
+
+    await snapshot_platform(
+        hass, entity_registry, device_registry, snapshot, mock_config_entry.entry_id
+    )
+
+
+async def test_turn_on_and_off(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry, mocked_hub: Device
+) -> None:
+    """Test that turn_on and turn_off services work as expected."""
+    await setup_platform_for_device(hass, mock_config_entry, Platform.SIREN, mocked_hub)
+
+    alarm_module = mocked_hub.modules[Module.Alarm]
+
+    await hass.services.async_call(
+        SIREN_DOMAIN,
+        SERVICE_TURN_OFF,
+        {ATTR_ENTITY_ID: [ENTITY_ID]},
+        blocking=True,
+    )
+
+    alarm_module.stop.assert_called()
+
+    await hass.services.async_call(
+        SIREN_DOMAIN,
+        SERVICE_TURN_ON,
+        {ATTR_ENTITY_ID: [ENTITY_ID]},
+        blocking=True,
+    )
+
+    alarm_module.play.assert_called()