Add support for siren entities in ZHA (#60920)

homeassistant/components/zha/core/const.py

@@ -22,6 +22,7 @@ from homeassistant.components.light import DOMAIN as LIGHT
 from homeassistant.components.lock import DOMAIN as LOCK
 from homeassistant.components.number import DOMAIN as NUMBER
 from homeassistant.components.sensor import DOMAIN as SENSOR
+from homeassistant.components.siren import DOMAIN as SIREN
 from homeassistant.components.switch import DOMAIN as SWITCH
 import homeassistant.helpers.config_validation as cv
 
@@ -120,6 +121,7 @@ PLATFORMS = (
     LOCK,
     NUMBER,
     SENSOR,
+    SIREN,
     SWITCH,
 )
 

homeassistant/components/zha/core/discovery.py

@@ -25,6 +25,7 @@ from .. import (  # noqa: F401 pylint: disable=unused-import,
     lock,
     number,
     sensor,
+    siren,
     switch,
 )
 from .channels import base

homeassistant/components/zha/core/registries.py

@@ -21,6 +21,7 @@ from homeassistant.components.light import DOMAIN as LIGHT
 from homeassistant.components.lock import DOMAIN as LOCK
 from homeassistant.components.number import DOMAIN as NUMBER
 from homeassistant.components.sensor import DOMAIN as SENSOR
+from homeassistant.components.siren import DOMAIN as SIREN
 from homeassistant.components.switch import DOMAIN as SWITCH
 
 # importing channels updates registries
@@ -113,6 +114,7 @@ DEVICE_CLASS = {
         zigpy.profiles.zha.DeviceType.SHADE: COVER,
         zigpy.profiles.zha.DeviceType.SMART_PLUG: SWITCH,
         zigpy.profiles.zha.DeviceType.IAS_ANCILLARY_CONTROL: ALARM,
+        zigpy.profiles.zha.DeviceType.IAS_WARNING_DEVICE: SIREN,
     },
     zigpy.profiles.zll.PROFILE_ID: {
         zigpy.profiles.zll.DeviceType.COLOR_LIGHT: LIGHT,

homeassistant/components/zha/siren.py

@@ -0,0 +1,143 @@
+"""Support for ZHA sirens."""
+
+from __future__ import annotations
+
+import functools
+from typing import Any
+
+from homeassistant.components.siren import (
+    ATTR_DURATION,
+    DOMAIN,
+    SUPPORT_DURATION,
+    SUPPORT_TURN_OFF,
+    SUPPORT_TURN_ON,
+    SirenEntity,
+)
+from homeassistant.components.siren.const import (
+    ATTR_TONE,
+    ATTR_VOLUME_LEVEL,
+    SUPPORT_TONES,
+    SUPPORT_VOLUME_SET,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.event import async_call_later
+
+from .core import discovery
+from .core.channels.security import IasWd
+from .core.const import (
+    CHANNEL_IAS_WD,
+    DATA_ZHA,
+    SIGNAL_ADD_ENTITIES,
+    WARNING_DEVICE_MODE_BURGLAR,
+    WARNING_DEVICE_MODE_EMERGENCY,
+    WARNING_DEVICE_MODE_EMERGENCY_PANIC,
+    WARNING_DEVICE_MODE_FIRE,
+    WARNING_DEVICE_MODE_FIRE_PANIC,
+    WARNING_DEVICE_MODE_POLICE_PANIC,
+    WARNING_DEVICE_MODE_STOP,
+    WARNING_DEVICE_SOUND_HIGH,
+    WARNING_DEVICE_STROBE_NO,
+)
+from .core.registries import ZHA_ENTITIES
+from .core.typing import ChannelType, ZhaDeviceType
+from .entity import ZhaEntity
+
+STRICT_MATCH = functools.partial(ZHA_ENTITIES.strict_match, DOMAIN)
+DEFAULT_DURATION = 5  # seconds
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the Zigbee Home Automation siren from config entry."""
+    entities_to_create = hass.data[DATA_ZHA][DOMAIN]
+
+    unsub = async_dispatcher_connect(
+        hass,
+        SIGNAL_ADD_ENTITIES,
+        functools.partial(
+            discovery.async_add_entities,
+            async_add_entities,
+            entities_to_create,
+            update_before_add=False,
+        ),
+    )
+    config_entry.async_on_unload(unsub)
+
+
+@STRICT_MATCH(channel_names=CHANNEL_IAS_WD)
+class ZHASiren(ZhaEntity, SirenEntity):
+    """Representation of a ZHA siren."""
+
+    def __init__(
+        self,
+        unique_id: str,
+        zha_device: ZhaDeviceType,
+        channels: list[ChannelType],
+        **kwargs,
+    ) -> None:
+        """Init this siren."""
+        self._attr_supported_features = (
+            SUPPORT_TURN_ON
+            | SUPPORT_TURN_OFF
+            | SUPPORT_DURATION
+            | SUPPORT_VOLUME_SET
+            | SUPPORT_TONES
+        )
+        self._attr_available_tones: list[int | str] | dict[int, str] | None = {
+            WARNING_DEVICE_MODE_BURGLAR: "Burglar",
+            WARNING_DEVICE_MODE_FIRE: "Fire",
+            WARNING_DEVICE_MODE_EMERGENCY: "Emergency",
+            WARNING_DEVICE_MODE_POLICE_PANIC: "Police Panic",
+            WARNING_DEVICE_MODE_FIRE_PANIC: "Fire Panic",
+            WARNING_DEVICE_MODE_EMERGENCY_PANIC: "Emergency Panic",
+        }
+        super().__init__(unique_id, zha_device, channels, **kwargs)
+        self._channel: IasWd = channels[0]
+        self._attr_is_on: bool = False
+        self._off_listener = None
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn on siren."""
+        if self._off_listener:
+            self._off_listener()
+            self._off_listener = None
+        siren_tone = WARNING_DEVICE_MODE_EMERGENCY
+        siren_duration = DEFAULT_DURATION
+        siren_level = WARNING_DEVICE_SOUND_HIGH
+        if (duration := kwargs.get(ATTR_DURATION)) is not None:
+            siren_duration = duration
+        if (tone := kwargs.get(ATTR_TONE)) is not None:
+            siren_tone = tone
+        if (level := kwargs.get(ATTR_VOLUME_LEVEL)) is not None:
+            siren_level = int(level)
+        await self._channel.issue_start_warning(
+            mode=siren_tone, warning_duration=siren_duration, siren_level=siren_level
+        )
+        self._attr_is_on = True
+        self._off_listener = async_call_later(
+            self._zha_device.hass, siren_duration, self.async_set_off
+        )
+        self.async_write_ha_state()
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn off siren."""
+        await self._channel.issue_start_warning(
+            mode=WARNING_DEVICE_MODE_STOP, strobe=WARNING_DEVICE_STROBE_NO
+        )
+        self._attr_is_on = False
+        self.async_write_ha_state()
+
+    @callback
+    def async_set_off(self, _) -> None:
+        """Set is_on to False and write HA state."""
+        self._attr_is_on = False
+        if self._off_listener:
+            self._off_listener()
+            self._off_listener = None
+        self.async_write_ha_state()

tests/components/zha/test_siren.py

@@ -0,0 +1,139 @@
+"""Test zha siren."""
+from datetime import timedelta
+from unittest.mock import patch
+
+import pytest
+from zigpy.const import SIG_EP_PROFILE
+import zigpy.profiles.zha as zha
+import zigpy.zcl.clusters.general as general
+import zigpy.zcl.clusters.security as security
+import zigpy.zcl.foundation as zcl_f
+
+from homeassistant.components.siren import DOMAIN
+from homeassistant.components.siren.const import (
+    ATTR_DURATION,
+    ATTR_TONE,
+    ATTR_VOLUME_LEVEL,
+)
+from homeassistant.components.zha.core.const import (
+    WARNING_DEVICE_MODE_EMERGENCY_PANIC,
+    WARNING_DEVICE_SOUND_MEDIUM,
+)
+from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
+import homeassistant.util.dt as dt_util
+
+from .common import async_enable_traffic, find_entity_id
+from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_TYPE
+
+from tests.common import async_fire_time_changed, mock_coro
+
+
+@pytest.fixture
+async def siren(hass, zigpy_device_mock, zha_device_joined_restored):
+    """Siren fixture."""
+
+    zigpy_device = zigpy_device_mock(
+        {
+            1: {
+                SIG_EP_INPUT: [general.Basic.cluster_id, security.IasWd.cluster_id],
+                SIG_EP_OUTPUT: [],
+                SIG_EP_TYPE: zha.DeviceType.IAS_WARNING_DEVICE,
+                SIG_EP_PROFILE: zha.PROFILE_ID,
+            }
+        },
+    )
+
+    zha_device = await zha_device_joined_restored(zigpy_device)
+    return zha_device, zigpy_device.endpoints[1].ias_wd
+
+
+async def test_siren(hass, siren):
+    """Test zha siren platform."""
+
+    zha_device, cluster = siren
+    assert cluster is not None
+    entity_id = await find_entity_id(DOMAIN, zha_device, hass)
+    assert entity_id is not None
+
+    assert hass.states.get(entity_id).state == STATE_OFF
+    await async_enable_traffic(hass, [zha_device], enabled=False)
+    # test that the switch was created and that its state is unavailable
+    assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
+
+    # allow traffic to flow through the gateway and device
+    await async_enable_traffic(hass, [zha_device])
+
+    # test that the state has changed from unavailable to off
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    # turn on from HA
+    with patch(
+        "zigpy.zcl.Cluster.request",
+        return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]),
+    ):
+        # turn on via UI
+        await hass.services.async_call(
+            DOMAIN, "turn_on", {"entity_id": entity_id}, blocking=True
+        )
+        assert len(cluster.request.mock_calls) == 1
+        assert cluster.request.call_args[0][0] is False
+        assert cluster.request.call_args[0][1] == 0
+        assert cluster.request.call_args[0][3] == 54  # bitmask for default args
+        assert cluster.request.call_args[0][4] == 5  # duration in seconds
+        assert cluster.request.call_args[0][5] == 0
+        assert cluster.request.call_args[0][6] == 2
+
+    # test that the state has changed to on
+    assert hass.states.get(entity_id).state == STATE_ON
+
+    # turn off from HA
+    with patch(
+        "zigpy.zcl.Cluster.request",
+        return_value=mock_coro([0x01, zcl_f.Status.SUCCESS]),
+    ):
+        # turn off via UI
+        await hass.services.async_call(
+            DOMAIN, "turn_off", {"entity_id": entity_id}, blocking=True
+        )
+        assert len(cluster.request.mock_calls) == 1
+        assert cluster.request.call_args[0][0] is False
+        assert cluster.request.call_args[0][1] == 0
+        assert cluster.request.call_args[0][3] == 2  # bitmask for default args
+        assert cluster.request.call_args[0][4] == 5  # duration in seconds
+        assert cluster.request.call_args[0][5] == 0
+        assert cluster.request.call_args[0][6] == 2
+
+    # test that the state has changed to off
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    # turn on from HA
+    with patch(
+        "zigpy.zcl.Cluster.request",
+        return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]),
+    ):
+        # turn on via UI
+        await hass.services.async_call(
+            DOMAIN,
+            "turn_on",
+            {
+                "entity_id": entity_id,
+                ATTR_DURATION: 10,
+                ATTR_TONE: WARNING_DEVICE_MODE_EMERGENCY_PANIC,
+                ATTR_VOLUME_LEVEL: WARNING_DEVICE_SOUND_MEDIUM,
+            },
+            blocking=True,
+        )
+        assert len(cluster.request.mock_calls) == 1
+        assert cluster.request.call_args[0][0] is False
+        assert cluster.request.call_args[0][1] == 0
+        assert cluster.request.call_args[0][3] == 101  # bitmask for passed args
+        assert cluster.request.call_args[0][4] == 10  # duration in seconds
+        assert cluster.request.call_args[0][5] == 0
+        assert cluster.request.call_args[0][6] == 2
+
+        # test that the state has changed to on
+    assert hass.states.get(entity_id).state == STATE_ON
+
+    now = dt_util.utcnow() + timedelta(seconds=15)
+    async_fire_time_changed(hass, now)
+    await hass.async_block_till_done()

tests/components/zha/zha_devices_list.py

@@ -593,13 +593,21 @@ DEVICES = [
                 SIG_EP_PROFILE: 260,
             }
         },
-        DEV_SIG_ENTITIES: ["binary_sensor.heiman_warningdevice_77665544_ias_zone"],
+        DEV_SIG_ENTITIES: [
+            "binary_sensor.heiman_warningdevice_77665544_ias_zone",
+            "siren.heiman_warningdevice_77665544_ias_wd",
+        ],
         DEV_SIG_ENT_MAP: {
             ("binary_sensor", "00:11:22:33:44:55:66:77-1-1280"): {
                 DEV_SIG_CHANNELS: ["ias_zone"],
                 DEV_SIG_ENT_MAP_CLASS: "IASZone",
                 DEV_SIG_ENT_MAP_ID: "binary_sensor.heiman_warningdevice_77665544_ias_zone",
-            }
+            },
+            ("siren", "00:11:22:33:44:55:66:77-1"): {
+                DEV_SIG_CHANNELS: ["ias_wd"],
+                DEV_SIG_ENT_MAP_CLASS: "ZHASiren",
+                DEV_SIG_ENT_MAP_ID: "siren.heiman_warningdevice_77665544_ias_wd",
+            },
         },
         DEV_SIG_EVT_CHANNELS: ["1:0x0019"],
         SIG_MANUFACTURER: "Heiman",