diff --git a/tests/test_siren.py b/tests/test_siren.py index 997f28d41..b6f5ca998 100644 --- a/tests/test_siren.py +++ b/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 @@ async def test_siren_timed_off(zha_gateway: Gateway) -> None: # 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) + ] diff --git a/zha/application/__init__.py b/zha/application/__init__.py index d8c01161c..3bffe1e73 100644 --- a/zha/application/__init__.py +++ b/zha/application/__init__.py @@ -42,4 +42,5 @@ class EntityPlatform(StrEnum): NUMBER = "number" SENSOR = "sensor" SELECT = "select" + SIREN = "siren" SWITCH = "switch" diff --git a/zha/application/platforms/siren.py b/zha/application/platforms/siren.py index 4a3688681..38eede295 100644 --- a/zha/application/platforms/siren.py +++ b/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()