Add attribute-controlled siren entity for quirks v2

tests/test_siren.py

@@ -1,12 +1,17 @@
 """Test zha siren."""
 
 import asyncio
-from unittest.mock import patch
+from unittest.mock import call, patch
 
+from zhaquirks.builder import QuirkBuilder
+from zhaquirks.clusters import CustomCluster
 from zigpy.const import SIG_EP_PROFILE
 from zigpy.profiles import zha
+import zigpy.types as t
+from zigpy.typing import UNDEFINED
 from zigpy.zcl.clusters import general, security
 import zigpy.zcl.foundation as zcl_f
+from zigpy.zcl.foundation import BaseAttributeDefs, ZCLAttributeDef
 
 from tests.common import (
     SIG_EP_INPUT,
@@ -16,11 +21,13 @@
     get_entity,
     join_zigpy_device,
     mock_coro,
+    send_attributes_report,
+    update_attribute_cache,
 )
 from zha.application import Platform
 from zha.application.gateway import Gateway
-from zha.application.platforms.siren import SirenEntityFeature
-from zha.quirks import SIREN_BASIC
+from zha.application.platforms.siren import AttributeSiren, SirenEntityFeature
+from zha.quirks import QUIRK_REGISTRY_ENTRY_ATTR, SIREN_BASIC, DeviceRegistry
 from zha.zigbee.device import Device
 
 
@@ -236,3 +243,160 @@
 
     # test that the state has changed to off from the timer
     assert entity.state["state"] is False
+
+
+class _SmokeSirenEnum(t.enum8):
+    """Smoke siren type."""
+
+    Stop = 0
+    Smoke_siren = 1
+    CO_siren = 2
+
+
+class _SirenManufCluster(CustomCluster):
+    """Manufacturer-specific cluster with the attribute-controlled siren."""
+
+    cluster_id = 0xFC90
+    ep_attribute = "heiman_siren"
+
+    class AttributeDefs(BaseAttributeDefs):
+        """Attribute definitions."""
+
+        siren_for_automation = ZCLAttributeDef(
+            id=0x0012, type=_SmokeSirenEnum, manufacturer_code=0x120B
+        )
+
+
+async def attribute_siren_mock(
+    zha_gateway: Gateway,
+) -> tuple[Device, security.IasWd, CustomCluster]:
+    """Build a device whose quirk replaces the IAS WD siren with an AttributeSiren."""
+    zigpy_device = create_mock_zigpy_device(
+        zha_gateway,
+        {
+            1: {
+                SIG_EP_INPUT: [
+                    general.Basic.cluster_id,
+                    security.IasWd.cluster_id,
+                    _SirenManufCluster.cluster_id,
+                ],
+                SIG_EP_OUTPUT: [],
+                SIG_EP_TYPE: zha.DeviceType.IAS_ZONE,
+                SIG_EP_PROFILE: zha.PROFILE_ID,
+            }
+        },
+        manufacturer="HEIMAN",
+        model="attribute-siren-test",
+    )
+
+    registry = DeviceRegistry()
+    (
+        QuirkBuilder(zigpy_device.manufacturer, zigpy_device.model)
+        .replaces(_SirenManufCluster)
+        .prevent_default_entity_creation(
+            endpoint_id=1, cluster_id=security.IasWd.cluster_id
+        )
+        .siren(
+            _SirenManufCluster.AttributeDefs.siren_for_automation.name,
+            _SirenManufCluster.cluster_id,
+            available_tones={
+                _SmokeSirenEnum.Smoke_siren: "Smoke siren",
+                _SmokeSirenEnum.CO_siren: "CO siren",
+            },
+            off_value=_SmokeSirenEnum.Stop,
+            default_tone=_SmokeSirenEnum.Smoke_siren,
+            unique_id_suffix=str(security.IasWd.cluster_id),
+            translation_key="siren",
+            fallback_name="Siren",
+        )
+        .add_to_registry(registry)
+    )
+
+    zigpy_device = registry.resolve(zigpy_device)
+    assert getattr(zigpy_device, QUIRK_REGISTRY_ENTRY_ATTR, None) is not None
+
+    cluster = zigpy_device.endpoints[1].heiman_siren
+    cluster.PLUGGED_ATTR_READS = {"siren_for_automation": _SmokeSirenEnum.Stop}
+    update_attribute_cache(cluster)
+
+    zha_device = await join_zigpy_device(zha_gateway, zigpy_device)
+    return zha_device, zigpy_device.endpoints[1].ias_wd, cluster
+
+
+async def test_attribute_siren(zha_gateway: Gateway) -> None:
+    """Test the quirks v2 attribute-controlled siren entity."""
+    zha_device, ias_wd, cluster = await attribute_siren_mock(zha_gateway)
+
+    entity = get_entity(zha_device, platform=Platform.SIREN)
+
+    # the IAS WD siren was suppressed and replaced by the AttributeSiren, which
+    # reuses the IAS WD siren's unique_id so existing entities migrate
+    assert isinstance(entity, AttributeSiren)
+    assert entity.unique_id.endswith(f"-1-{int(security.IasWd.cluster_id)}")
+
+    assert entity.supported_features == (
+        SirenEntityFeature.TURN_ON
+        | SirenEntityFeature.TURN_OFF
+        | SirenEntityFeature.TONES
+    )
+    assert entity.available_tones == {
+        _SmokeSirenEnum.Smoke_siren: "Smoke siren",
+        _SmokeSirenEnum.CO_siren: "CO siren",
+    }
+
+    # seeded to Stop
+    assert entity.state["state"] is False
+
+    # device-driven report turns the siren on...
+    await send_attributes_report(
+        zha_gateway, cluster, {"siren_for_automation": _SmokeSirenEnum.Smoke_siren}
+    )
+    assert entity.state["state"] is True
+
+    # ...and the device's own reset back to Stop (e.g. the ~10 min timeout)
+    # updates the entity state without any user action
+    await send_attributes_report(
+        zha_gateway, cluster, {"siren_for_automation": _SmokeSirenEnum.Stop}
+    )
+    assert entity.state["state"] is False
+
+    # turn on from HA without a tone writes the default tone
+    with patch(
+        "zigpy.zcl.Cluster.write_attributes",
+        return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]],
+    ):
+        await entity.async_turn_on()
+        await zha_gateway.async_block_till_done()
+        assert cluster.write_attributes.mock_calls == [
+            call(
+                {"siren_for_automation": _SmokeSirenEnum.Smoke_siren},
+                manufacturer=UNDEFINED,
+            )
+        ]
+        cluster.write_attributes.reset_mock()
+
+    # turn on with an explicit tone writes that tone
+    with patch(
+        "zigpy.zcl.Cluster.write_attributes",
+        return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]],
+    ):
+        await entity.async_turn_on(tone=_SmokeSirenEnum.CO_siren)
+        await zha_gateway.async_block_till_done()
+        assert cluster.write_attributes.mock_calls == [
+            call(
+                {"siren_for_automation": _SmokeSirenEnum.CO_siren},
+                manufacturer=UNDEFINED,
+            )
+        ]
+        cluster.write_attributes.reset_mock()
+
+    # turn off writes the off value
+    with patch(
+        "zigpy.zcl.Cluster.write_attributes",
+        return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]],
+    ):
+        await entity.async_turn_off()
+        await zha_gateway.async_block_till_done()
+        assert cluster.write_attributes.mock_calls == [
+            call({"siren_for_automation": _SmokeSirenEnum.Stop}, manufacturer=UNDEFINED)
+        ]

zha/application/__init__.py

@@ -42,4 +42,5 @@ class EntityPlatform(StrEnum):
     NUMBER = "number"
     SENSOR = "sensor"
     SELECT = "select"
+    SIREN = "siren"
     SWITCH = "switch"

zha/application/platforms/siren.py

@@ -11,6 +11,13 @@
 from typing import TYPE_CHECKING, Any, Final
 
 from zigpy.profiles import zha
+import zigpy.zcl
+from zigpy.zcl import (
+    AttributeReadEvent,
+    AttributeReportedEvent,
+    AttributeUpdatedEvent,
+    AttributeWrittenEvent,
+)
 from zigpy.zcl.clusters.security import (
     IasWd,
     SirenLevel,
@@ -23,6 +30,7 @@
 )
 
 from zha.application import Platform
+from zha.application.helpers import write_attributes_safe
 from zha.application.platforms import (
     BaseEntityInfo,
     ClusterConfig,
@@ -121,17 +129,6 @@ async def async_turn_on(
     async def async_turn_off(self) -> None:
         """Turn off siren."""
 
-    # This method is a ZHA extension to the base HA siren entity
-    @abstractmethod
-    async def async_squawk(
-        self,
-        *,
-        mode: SquawkMode,
-        strobe: int,
-        squawk_level: int,
-    ) -> None:
-        """Issue a brief squawk pulse."""
-
 
 class BaseZclSiren(BaseSiren, ABC):
     """Base class for ZHA IAS WD siren entities with shared ZCL logic."""
@@ -387,3 +384,104 @@ async def async_turn_on(
         )
         self._tracked_handles.append(self._off_listener)
         self.maybe_emit_state_changed_event()
+
+
+class AttributeSiren(BaseSiren):
+    """Siren that is controlled by writing an enum attribute.
+
+    Unlike the IAS WD sirens, this entity does not issue ``start_warning``
+    commands. It turns on by writing a (tone) value to a manufacturer-specific
+    attribute and off by writing ``off_value``; the device keeps sounding until
+    it is turned off or the device itself reports the attribute back to
+    ``off_value``. State is derived from the cached attribute value, so a device
+    that resets the attribute on its own keeps the entity in sync.
+
+    This entity is only created from quirks v2 metadata (``.siren(...)``); it has
+    no default cluster match.
+    """
+
+    _attr_fallback_name: str = "Siren"
+
+    def __init__(
+        self,
+        endpoint: Endpoint,
+        device: Device,
+        *,
+        cluster: zigpy.zcl.Cluster,
+        attribute_name: str,
+        available_tones: dict[int, str] | None = None,
+        off_value: int = 0,
+        default_tone: int | None = None,
+        **kwargs: Any,
+    ) -> None:
+        """Init this attribute-controlled siren."""
+        self._attribute_name = attribute_name
+        self._off_value = off_value
+        self._attr_available_tones = dict(available_tones or {})
+        # Tone written when turned on without an explicit tone: the configured
+        # default, else the first available tone, else 1.
+        self._default_tone = (
+            default_tone
+            if default_tone is not None
+            else next(iter(self._attr_available_tones), 1)
+        )
+        super().__init__(endpoint=endpoint, device=device, cluster=cluster, **kwargs)
+        self._attr_supported_features = (
+            SirenEntityFeature.TURN_ON | SirenEntityFeature.TURN_OFF
+        )
+        if self._attr_available_tones:
+            self._attr_supported_features |= SirenEntityFeature.TONES
+
+    def on_add(self) -> None:
+        """Subscribe to attribute updates so device-driven changes update state."""
+        super().on_add()
+        for event_type in (
+            AttributeReadEvent,
+            AttributeReportedEvent,
+            AttributeUpdatedEvent,
+            AttributeWrittenEvent,
+        ):
+            self._on_remove_callbacks.append(
+                self._cluster.on_event(
+                    event_type.event_type, self.handle_attribute_updated
+                )
+            )
+
+    def handle_attribute_updated(
+        self,
+        event: AttributeReadEvent
+        | AttributeReportedEvent
+        | AttributeUpdatedEvent
+        | AttributeWrittenEvent,
+    ) -> None:
+        """Handle state update from the cluster."""
+        if event.attribute_name == self._attribute_name:
+            self.maybe_emit_state_changed_event()
+
+    @property
+    def is_on(self) -> bool:
+        """Return true if the siren is sounding."""
+        value = self._cluster.get(self._attribute_name)
+        return value is not None and value != self._off_value
+
+    async def async_turn_on(
+        self,
+        duration: int | None = None,
+        tone: int | None = None,
+        volume_level: int | None = None,
+        # These kwargs are ZHA extensions to the base HA entity signature
+        strobe: int | None = None,
+        strobe_duty_cycle: int | None = None,
+        strobe_intensity: int | None = None,
+    ) -> None:
+        """Turn on siren by writing the requested tone to the attribute."""
+        siren_tone = tone if tone is not None else self._default_tone
+        await write_attributes_safe(self._cluster, {self._attribute_name: siren_tone})
+        self.maybe_emit_state_changed_event()
+
+    async def async_turn_off(self) -> None:
+        """Turn off siren by writing the off value to the attribute."""
+        await write_attributes_safe(
+            self._cluster, {self._attribute_name: self._off_value}
+        )
+        self.maybe_emit_state_changed_event()