diff --git a/tests/test_siren.py b/tests/test_siren.py index 746a79926..048573eb6 100644 --- a/tests/test_siren.py +++ b/tests/test_siren.py @@ -1,10 +1,15 @@ """Test zha siren.""" import asyncio -from unittest.mock import patch +from unittest.mock import call, patch from zigpy.const import SIG_EP_PROFILE from zigpy.profiles import zha +from zigpy.quirks import DEVICE_REGISTRY +from zigpy.quirks.v2 import CustomDeviceV2, QuirkBuilder +from zigpy.quirks.v2.homeassistant import EntityPlatform +import zigpy.types as t +from zigpy.typing import UNDEFINED from zigpy.zcl.clusters import general, security import zigpy.zcl.foundation as zcl_f @@ -16,15 +21,22 @@ 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.application.platforms.siren import ( + ConfigurableAttributeSiren, + EnumSiren, + SirenEntityFeature, +) from zha.zigbee.device import Device async def siren_mock( zha_gateway: Gateway, + basic: bool = False, ) -> tuple[Device, security.IasWd]: """Siren fixture.""" @@ -40,6 +52,9 @@ async def siren_mock( }, ) + if basic: + zigpy_device.quirk_id = {"siren_basic"} + zha_device = await join_zigpy_device(zha_gateway, zigpy_device) return zha_device, zigpy_device.endpoints[1].ias_wd @@ -119,6 +134,79 @@ async def test_siren(zha_gateway: Gateway) -> None: assert entity.state["state"] is True +async def test_basic_siren(zha_gateway: Gateway) -> None: + """Test zha basic siren.""" + + zha_device, cluster = await siren_mock(zha_gateway, basic=True) + assert cluster is not None + + entity = get_entity(zha_device, platform=Platform.SIREN) + assert entity.supported_features == ( + SirenEntityFeature.TURN_ON + | SirenEntityFeature.TURN_OFF + | SirenEntityFeature.DURATION + ) + + assert entity.state["state"] is False + + # turn on from client + with patch( + "zigpy.zcl.Cluster.request", + return_value=[0x00, zcl_f.Status.SUCCESS], + ): + await entity.async_turn_on() + await zha_gateway.async_block_till_done() + 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] == 50 # 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 + cluster.request.reset_mock() + + # test that the state has changed to on + assert entity.state["state"] is True + + # turn off from client + with patch( + "zigpy.zcl.Cluster.request", + return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]), + ): + await entity.async_turn_off() + await zha_gateway.async_block_till_done() + 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 + cluster.request.reset_mock() + + # test that the state has changed to off + assert entity.state["state"] is False + + # turn on from client with duration option + with patch( + "zigpy.zcl.Cluster.request", + return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]), + ): + await entity.async_turn_on(duration=100) + await zha_gateway.async_block_till_done() + 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] == 50 # bitmask for specified args + assert cluster.request.call_args[0][4] == 100 # duration in seconds + assert cluster.request.call_args[0][5] == 0 + assert cluster.request.call_args[0][6] == 2 + cluster.request.reset_mock() + + # test that the state has changed to on + assert entity.state["state"] is True + + async def test_siren_timed_off(zha_gateway: Gateway) -> None: """Test zha siren platform.""" zha_device, cluster = await siren_mock(zha_gateway) @@ -151,3 +239,485 @@ 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 + + +async def test_siren_configurable_attribute(zha_gateway: Gateway) -> None: + """Test ZHA configurable attribute siren created from quirks v2 SwitchMetadata.""" + + zigpy_dev = create_mock_zigpy_device( + zha_gateway, + { + 1: { + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.IAS_WARNING_DEVICE, + SIG_EP_PROFILE: zha.PROFILE_ID, + } + }, + manufacturer="FakeSirenManufacturer", + model="FakeSirenModel", + ) + + ( + QuirkBuilder(zigpy_dev.manufacturer, zigpy_dev.model) + .switch( + general.Basic.AttributeDefs.power_source.name, + general.Basic.cluster_id, + on_value=1, + off_value=0, + entity_platform=EntityPlatform.SIREN, + translation_key="siren", + fallback_name="Siren", + ) + .add_to_registry() + ) + + zigpy_device_ = DEVICE_REGISTRY.get_device(zigpy_dev) + assert isinstance(zigpy_device_, CustomDeviceV2) + + cluster = zigpy_device_.endpoints[1].basic + cluster.PLUGGED_ATTR_READS = { + general.Basic.AttributeDefs.power_source.name: 0, + } + update_attribute_cache(cluster) + + zha_device = await join_zigpy_device(zha_gateway, zigpy_device_) + + entity = get_entity(zha_device, platform=Platform.SIREN) + assert isinstance(entity, ConfigurableAttributeSiren) + assert entity.supported_features == ( + SirenEntityFeature.TURN_ON | SirenEntityFeature.TURN_OFF + ) + assert entity.state["state"] is False + + # turn on via attribute report + await send_attributes_report( + zha_gateway, cluster, {general.Basic.AttributeDefs.power_source.name: 1} + ) + assert entity.state["state"] is True + + # turn off via attribute report + await send_attributes_report( + zha_gateway, cluster, {general.Basic.AttributeDefs.power_source.name: 0} + ) + assert entity.state["state"] is False + + # turn on from HA + 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( + {general.Basic.AttributeDefs.power_source.name: 1}, + manufacturer=UNDEFINED, + ) + ] + cluster.write_attributes.reset_mock() + + # turn off from HA + 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( + {general.Basic.AttributeDefs.power_source.name: 0}, + manufacturer=UNDEFINED, + ) + ] + + +async def test_siren_configurable_attribute_custom_on_off_values( + zha_gateway: Gateway, +) -> None: + """Test configurable attribute siren with custom on/off values.""" + + zigpy_dev = create_mock_zigpy_device( + zha_gateway, + { + 1: { + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.IAS_WARNING_DEVICE, + SIG_EP_PROFILE: zha.PROFILE_ID, + } + }, + manufacturer="FakeSirenManufacturer2", + model="FakeSirenModel2", + ) + + ( + QuirkBuilder(zigpy_dev.manufacturer, zigpy_dev.model) + .switch( + general.Basic.AttributeDefs.power_source.name, + general.Basic.cluster_id, + on_value=3, + off_value=5, + entity_platform=EntityPlatform.SIREN, + translation_key="siren", + fallback_name="Siren", + ) + .add_to_registry() + ) + + zigpy_device_ = DEVICE_REGISTRY.get_device(zigpy_dev) + assert isinstance(zigpy_device_, CustomDeviceV2) + + cluster = zigpy_device_.endpoints[1].basic + cluster.PLUGGED_ATTR_READS = { + general.Basic.AttributeDefs.power_source.name: 5, + } + update_attribute_cache(cluster) + + zha_device = await join_zigpy_device(zha_gateway, zigpy_device_) + entity = get_entity(zha_device, platform=Platform.SIREN) + assert isinstance(entity, ConfigurableAttributeSiren) + assert entity.state["state"] is False + + # turn on via attribute report + await send_attributes_report( + zha_gateway, cluster, {general.Basic.AttributeDefs.power_source.name: 3} + ) + assert entity.state["state"] is True + + # turn off via attribute report + await send_attributes_report( + zha_gateway, cluster, {general.Basic.AttributeDefs.power_source.name: 5} + ) + assert entity.state["state"] is False + + # turn on from HA + 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( + {general.Basic.AttributeDefs.power_source.name: 3}, + manufacturer=UNDEFINED, + ) + ] + cluster.write_attributes.reset_mock() + + # turn off from HA + 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( + {general.Basic.AttributeDefs.power_source.name: 5}, + manufacturer=UNDEFINED, + ) + ] + + +async def test_siren_configurable_attribute_force_inverted( + zha_gateway: Gateway, +) -> None: + """Test configurable attribute siren with force_inverted=True.""" + + zigpy_dev = create_mock_zigpy_device( + zha_gateway, + { + 1: { + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.IAS_WARNING_DEVICE, + SIG_EP_PROFILE: zha.PROFILE_ID, + } + }, + manufacturer="FakeSirenManufacturer3", + model="FakeSirenModel3", + ) + + ( + QuirkBuilder(zigpy_dev.manufacturer, zigpy_dev.model) + .switch( + general.Basic.AttributeDefs.power_source.name, + general.Basic.cluster_id, + on_value=3, + off_value=5, + force_inverted=True, + entity_platform=EntityPlatform.SIREN, + translation_key="siren", + fallback_name="Siren", + ) + .add_to_registry() + ) + + zigpy_device_ = DEVICE_REGISTRY.get_device(zigpy_dev) + assert isinstance(zigpy_device_, CustomDeviceV2) + + cluster = zigpy_device_.endpoints[1].basic + cluster.PLUGGED_ATTR_READS = { + general.Basic.AttributeDefs.power_source.name: 5, + } + update_attribute_cache(cluster) + + zha_device = await join_zigpy_device(zha_gateway, zigpy_device_) + entity = get_entity(zha_device, platform=Platform.SIREN) + assert isinstance(entity, ConfigurableAttributeSiren) + + # with force_inverted, off_value=5 reads as on + assert entity.state["state"] is True + + # attribute = on_value(3) -> inverted -> state is off + await send_attributes_report( + zha_gateway, cluster, {general.Basic.AttributeDefs.power_source.name: 3} + ) + assert entity.state["state"] is False + + # attribute = off_value(5) -> inverted -> state is on + await send_attributes_report( + zha_gateway, cluster, {general.Basic.AttributeDefs.power_source.name: 5} + ) + assert entity.state["state"] is True + + # turn on from HA: inverted, so writes off_value + 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( + {general.Basic.AttributeDefs.power_source.name: 5}, + manufacturer=UNDEFINED, + ) + ] + cluster.write_attributes.reset_mock() + + # turn off from HA: inverted, so writes on_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( + {general.Basic.AttributeDefs.power_source.name: 3}, + manufacturer=UNDEFINED, + ) + ] + + +async def test_siren_configurable_attribute_inverter_attribute( + zha_gateway: Gateway, +) -> None: + """Test configurable attribute siren with invert_attribute_name.""" + + zigpy_dev = create_mock_zigpy_device( + zha_gateway, + { + 1: { + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.IAS_WARNING_DEVICE, + SIG_EP_PROFILE: zha.PROFILE_ID, + } + }, + manufacturer="FakeSirenManufacturer4", + model="FakeSirenModel4", + ) + + ( + QuirkBuilder(zigpy_dev.manufacturer, zigpy_dev.model) + .switch( + general.Basic.AttributeDefs.power_source.name, + general.Basic.cluster_id, + on_value=3, + off_value=5, + invert_attribute_name=general.Basic.AttributeDefs.disable_local_config.name, + entity_platform=EntityPlatform.SIREN, + translation_key="siren", + fallback_name="Siren", + ) + .add_to_registry() + ) + + zigpy_device_ = DEVICE_REGISTRY.get_device(zigpy_dev) + assert isinstance(zigpy_device_, CustomDeviceV2) + + cluster = zigpy_device_.endpoints[1].basic + cluster.PLUGGED_ATTR_READS = { + general.Basic.AttributeDefs.power_source.name: 5, + general.Basic.AttributeDefs.disable_local_config.name: t.Bool(True), + } + update_attribute_cache(cluster) + + zha_device = await join_zigpy_device(zha_gateway, zigpy_device_) + entity = get_entity(zha_device, platform=Platform.SIREN) + assert isinstance(entity, ConfigurableAttributeSiren) + + # inverter_attribute is True, so off_value(5) reads as on + assert entity.state["state"] is True + + # attribute = on_value(3), inverter still True -> state is off + await send_attributes_report( + zha_gateway, cluster, {general.Basic.AttributeDefs.power_source.name: 3} + ) + assert entity.state["state"] is False + + # inverter attribute flipped to False -> off_value(5) with no inversion -> off + await send_attributes_report( + zha_gateway, + cluster, + {general.Basic.AttributeDefs.disable_local_config.name: t.Bool(False)}, + ) + await send_attributes_report( + zha_gateway, cluster, {general.Basic.AttributeDefs.power_source.name: 5} + ) + assert entity.state["state"] is False + + # turn on from HA (not inverted now): writes on_value + 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( + {general.Basic.AttributeDefs.power_source.name: 3}, + manufacturer=UNDEFINED, + ) + ] + cluster.write_attributes.reset_mock() + + # turn off from HA: writes 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( + {general.Basic.AttributeDefs.power_source.name: 5}, + manufacturer=UNDEFINED, + ) + ] + + +async def test_siren_enum(zha_gateway: Gateway) -> None: + """Test ZHA enum siren created from quirks v2 ZCLEnumMetadata. + + Uses a custom enum whose entries map to siren alert modes: + - entry 0 (Off) → off state + - entry 1 (Alert) → default on tone + - entry 2 (Alarm) → named tone + """ + + class SirenMode(t.enum8): + Off = 0x00 + Tone_1 = 0x01 + Tone_2 = 0x02 + + attr_name = general.Basic.AttributeDefs.disable_local_config.name + + zigpy_dev = create_mock_zigpy_device( + zha_gateway, + { + 1: { + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.ON_OFF_LIGHT, + SIG_EP_PROFILE: zha.PROFILE_ID, + } + }, + manufacturer="FakeSirenManufacturer5", + model="FakeSirenModel5", + ) + + ( + QuirkBuilder(zigpy_dev.manufacturer, zigpy_dev.model) + .enum( + attr_name, + SirenMode, + general.Basic.cluster_id, + entity_platform=EntityPlatform.SIREN, + translation_key="siren_mode", + fallback_name="Siren mode", + ) + .add_to_registry() + ) + + zigpy_device_ = DEVICE_REGISTRY.get_device(zigpy_dev) + assert isinstance(zigpy_device_, CustomDeviceV2) + + cluster = zigpy_device_.endpoints[1].basic + cluster.PLUGGED_ATTR_READS = {attr_name: SirenMode.Off} + update_attribute_cache(cluster) + + zha_device = await join_zigpy_device(zha_gateway, zigpy_device_) + entity = get_entity(zha_device, platform=Platform.SIREN) + assert isinstance(entity, EnumSiren) + + # Entry 0 (Off) = off state; entries 1+ are available tones + assert entity.available_tones == { + SirenMode.Tone_1.value: "Tone 1", + SirenMode.Tone_2.value: "Tone 2", + } + assert entity.supported_features == ( + SirenEntityFeature.TURN_ON + | SirenEntityFeature.TURN_OFF + | SirenEntityFeature.TONES + ) + + # Initial state: Off (0) -> off + assert entity.state["state"] is False + + # Attribute report: Alert -> on + await send_attributes_report(zha_gateway, cluster, {attr_name: SirenMode.Tone_1}) + assert entity.state["state"] is True + + # Attribute report: Off -> off + await send_attributes_report(zha_gateway, cluster, {attr_name: SirenMode.Off}) + assert entity.state["state"] is False + + # Turn on without tone: writes entry 1 (Alert) + 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({attr_name: SirenMode.Tone_1}, manufacturer=UNDEFINED) + ] + cluster.write_attributes.reset_mock() + + # Turn on with a specific tone (Alarm = value 2) + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + await entity.async_turn_on(tone=SirenMode.Tone_2.value) + await zha_gateway.async_block_till_done() + assert cluster.write_attributes.mock_calls == [ + call({attr_name: SirenMode.Tone_2}, manufacturer=UNDEFINED) + ] + cluster.write_attributes.reset_mock() + + # Turn off: writes entry 0 (Off) + 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({attr_name: SirenMode.Off}, manufacturer=UNDEFINED) + ] diff --git a/zha/application/discovery.py b/zha/application/discovery.py index 493b48be4..4b6fe5acc 100644 --- a/zha/application/discovery.py +++ b/zha/application/discovery.py @@ -107,6 +107,8 @@ (Platform.SENSOR, ZCLSensorMetadata): sensor.Sensor, (Platform.SELECT, ZCLEnumMetadata): select.ZCLEnumSelectEntity, (Platform.NUMBER, NumberMetadata): number.NumberConfigurationEntity, + (Platform.SIREN, SwitchMetadata): siren.ConfigurableAttributeSiren, + (Platform.SIREN, ZCLEnumMetadata): siren.EnumSiren, (Platform.SWITCH, SwitchMetadata): switch.ConfigurableAttributeSwitch, } diff --git a/zha/application/platforms/__init__.py b/zha/application/platforms/__init__.py index fae77701d..b0e440de5 100644 --- a/zha/application/platforms/__init__.py +++ b/zha/application/platforms/__init__.py @@ -69,6 +69,9 @@ class PlatformFeatureGroup(StrEnum): # Model-specific overrides for local temperature calibration LOCAL_TEMPERATURE_CALIBRATION = "local_temperature_calibration" + # IAS WD siren entity selection + SIREN = "siren" + @dataclasses.dataclass(frozen=True) class ClusterHandlerMatch: diff --git a/zha/application/platforms/siren.py b/zha/application/platforms/siren.py index 1e6933d22..7cd2d198b 100644 --- a/zha/application/platforms/siren.py +++ b/zha/application/platforms/siren.py @@ -2,14 +2,16 @@ from __future__ import annotations +from abc import ABC, abstractmethod import asyncio import contextlib from dataclasses import dataclass -from enum import IntFlag +from enum import Enum, IntFlag import functools from typing import TYPE_CHECKING, Any, Final, cast from zigpy.profiles import zha +from zigpy.quirks.v2 import SwitchMetadata, ZCLEnumMetadata from zigpy.zcl.clusters.security import IasWd from zha.application import Platform @@ -30,9 +32,14 @@ BaseEntityInfo, ClusterHandlerMatch, PlatformEntity, + PlatformFeatureGroup, register_entity, ) -from zha.zigbee.cluster_handlers.const import CLUSTER_HANDLER_IAS_WD +from zha.zigbee.cluster_handlers import ClusterAttributeUpdatedEvent +from zha.zigbee.cluster_handlers.const import ( + CLUSTER_HANDLER_ATTRIBUTE_UPDATED, + CLUSTER_HANDLER_IAS_WD, +) from zha.zigbee.cluster_handlers.security import IasWdClusterHandler if TYPE_CHECKING: @@ -66,16 +73,311 @@ class SirenEntityInfo(BaseEntityInfo): supported_features: SirenEntityFeature -@register_entity(IasWd.cluster_id) -class Siren(PlatformEntity): - """Representation of a ZHA siren.""" +class BaseSiren(PlatformEntity, ABC): + """Abstract base class for ZHA siren entities.""" PLATFORM = Platform.SIREN + + _attr_is_on: bool = False + _attr_available_tones: dict[int, str] + _attr_supported_features: SirenEntityFeature + + @property + def state(self) -> dict[str, Any]: + """Get the state of the siren.""" + response = super().state + response["state"] = self.is_on + return response + + @property + def is_on(self) -> bool: + """Return true if the entity is on.""" + return self._attr_is_on + + @property + def available_tones(self) -> dict[int, str]: + """Return available tones.""" + return self._attr_available_tones + + @property + def supported_features(self) -> SirenEntityFeature: + """Return supported features.""" + return self._attr_supported_features + + @functools.cached_property + def info_object(self) -> SirenEntityInfo: + """Return representation of the siren.""" + return SirenEntityInfo( + **super().info_object.__dict__, + available_tones=self.available_tones, + supported_features=self.supported_features, + ) + + @abstractmethod + async def async_turn_on( + self, + duration: int | None = None, + tone: int | None = None, + volume_level: int | None = None, + ) -> None: + """Turn on siren.""" + + @abstractmethod + async def async_turn_off(self) -> None: + """Turn off siren.""" + + +class BaseZclSiren(BaseSiren, ABC): + """Base class for ZHA IAS WD siren entities with shared ZCL logic.""" + + _cluster_handler: IasWdClusterHandler + _off_listener: asyncio.TimerHandle | None + + def __init__( + self, + cluster_handlers: list[ClusterHandler], + endpoint: Endpoint, + device: Device, + **kwargs: Any, + ) -> None: + """Init ZCL siren base.""" + self._cluster_handler = cast(IasWdClusterHandler, cluster_handlers[0]) + self._off_listener = None + super().__init__(cluster_handlers, endpoint, device, **kwargs) + + async def async_turn_off(self) -> None: + """Turn off siren.""" + await self._cluster_handler.issue_start_warning( + mode=WARNING_DEVICE_MODE_STOP, strobe=WARNING_DEVICE_STROBE_NO + ) + self._attr_is_on = False + self.maybe_emit_state_changed_event() + + 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.cancel() + + with contextlib.suppress(ValueError): + self._tracked_handles.remove(self._off_listener) + + self._off_listener = None + self.maybe_emit_state_changed_event() + + +class ConfigurableAttributeSiren(BaseSiren): + """Siren entity backed by a ZCL attribute, created from quirks v2 SwitchMetadata.""" + + _attribute_name: str + _inverter_attribute_name: str | None = None + _force_inverted: bool = False + _off_value: int = 0 + _on_value: int = 1 + + def __init__( + self, + cluster_handlers: list[ClusterHandler], + endpoint: Endpoint, + device: Device, + **kwargs: Any, + ) -> None: + """Init this configurable attribute siren.""" + self._cluster_handler: ClusterHandler = cluster_handlers[0] + super().__init__(cluster_handlers, endpoint, device, **kwargs) + self._attr_supported_features = ( + SirenEntityFeature.TURN_ON | SirenEntityFeature.TURN_OFF + ) + self._attr_available_tones: dict[int, str] = {} + self._cluster_handler.on_event( + CLUSTER_HANDLER_ATTRIBUTE_UPDATED, + self.handle_cluster_handler_attribute_updated, + ) + + def _init_from_quirks_metadata(self, entity_metadata: SwitchMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + self._attribute_name = entity_metadata.attribute_name + if entity_metadata.invert_attribute_name: + self._inverter_attribute_name = entity_metadata.invert_attribute_name + if entity_metadata.force_inverted: + self._force_inverted = entity_metadata.force_inverted + self._off_value = entity_metadata.off_value + self._on_value = entity_metadata.on_value + + @property + def inverted(self) -> bool: + """Return True if the siren is inverted.""" + if self._inverter_attribute_name: + return bool( + self._cluster_handler.cluster.get(self._inverter_attribute_name) + ) + return self._force_inverted + + @property + def is_on(self) -> bool: + """Return if the siren is on based on the cluster attribute.""" + if self._on_value != 1: + val = self._cluster_handler.cluster.get(self._attribute_name) + val = val == self._on_value + else: + val = bool(self._cluster_handler.cluster.get(self._attribute_name)) + return (not val) if self.inverted else val + + def handle_cluster_handler_attribute_updated( + self, + event: ClusterAttributeUpdatedEvent, + ) -> None: + """Handle state update from cluster handler.""" + if event.attribute_name == self._attribute_name: + self.maybe_emit_state_changed_event() + + async def async_turn_on( + self, + duration: int | None = None, + tone: int | None = None, + volume_level: int | None = None, + ) -> None: + """Turn on siren.""" + await self._cluster_handler.write_attributes_safe( + { + self._attribute_name: self._on_value + if not self.inverted + else self._off_value + } + ) + self.maybe_emit_state_changed_event() + + async def async_turn_off(self) -> None: + """Turn off siren.""" + await self._cluster_handler.write_attributes_safe( + { + self._attribute_name: self._off_value + if not self.inverted + else self._on_value + } + ) + self.maybe_emit_state_changed_event() + + async def async_update(self) -> None: + """Attempt to retrieve the state of the entity.""" + self.debug("Polling current state") + polling_attrs = [self._attribute_name] + if self._inverter_attribute_name: + polling_attrs.append(self._inverter_attribute_name) + results = await self._cluster_handler.get_attributes( + polling_attrs, from_cache=False, only_cache=False + ) + self.debug("read values=%s", results) + self.maybe_emit_state_changed_event() + + +class EnumSiren(BaseSiren): + """Siren entity backed by a ZCL enum attribute, created from quirks v2 ZCLEnumMetadata. + + Entry 0 of the enum is the off state. + Entry 1 is the default tone used when no specific tone is requested. + All remaining entries are exposed as additional tones. + """ + + _attribute_name: str + _enum: type[Enum] + + def __init__( + self, + cluster_handlers: list[ClusterHandler], + endpoint: Endpoint, + device: Device, + **kwargs: Any, + ) -> None: + """Init this enum siren.""" + self._cluster_handler: ClusterHandler = cluster_handlers[0] + self._attr_available_tones: dict[int, str] = {} + super().__init__(cluster_handlers, endpoint, device, **kwargs) + self._attr_supported_features = ( + SirenEntityFeature.TURN_ON + | SirenEntityFeature.TURN_OFF + | SirenEntityFeature.TONES + ) + self._cluster_handler.on_event( + CLUSTER_HANDLER_ATTRIBUTE_UPDATED, + self.handle_cluster_handler_attribute_updated, + ) + + def _init_from_quirks_metadata(self, entity_metadata: ZCLEnumMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + self._attribute_name = entity_metadata.attribute_name + self._enum = entity_metadata.enum + # All entries except index 0 (off) are exposed as tones + entries = list(self._enum) + self._attr_available_tones = { + entry.value: entry.name.replace("_", " ") for entry in entries[1:] + } + + @property + def is_on(self) -> bool: + """Return True if the current enum value is not the off entry (index 0).""" + value = self._cluster_handler.cluster.get(self._attribute_name) + if value is None: + return False + off_value = next(iter(self._enum)).value + return int(value) != off_value + + def handle_cluster_handler_attribute_updated( + self, + event: ClusterAttributeUpdatedEvent, + ) -> None: + """Handle state update from cluster handler.""" + if event.attribute_name == self._attribute_name: + self.maybe_emit_state_changed_event() + + async def async_turn_on( + self, + duration: int | None = None, + tone: int | None = None, + volume_level: int | None = None, + ) -> None: + """Turn on siren. Uses tone if provided, otherwise the second enum entry.""" + entries = list(self._enum) + if tone is not None and tone in self._attr_available_tones: + target = self._enum(tone) + else: + # Default: second entry (index 1) + target = entries[1] + await self._cluster_handler.write_attributes_safe( + {self._attribute_name: target} + ) + self.maybe_emit_state_changed_event() + + async def async_turn_off(self) -> None: + """Turn off siren by writing the first enum entry (index 0).""" + off_entry = next(iter(self._enum)) + await self._cluster_handler.write_attributes_safe( + {self._attribute_name: off_entry} + ) + self.maybe_emit_state_changed_event() + + async def async_update(self) -> None: + """Attempt to retrieve the state of the entity.""" + self.debug("Polling current state") + results = await self._cluster_handler.get_attributes( + [self._attribute_name], from_cache=False, only_cache=False + ) + self.debug("read values=%s", results) + self.maybe_emit_state_changed_event() + + +@register_entity(IasWd.cluster_id) +class AdvancedSiren(BaseZclSiren): + """Representation of a ZHA siren with full tone, level, and strobe support.""" + _attr_fallback_name: str = "Siren" _attr_primary_weight = 4 _cluster_handler_match = ClusterHandlerMatch( cluster_handlers=frozenset({CLUSTER_HANDLER_IAS_WD}), + feature_priority=(PlatformFeatureGroup.SIREN, 0), ) def __init__( @@ -86,10 +388,6 @@ def __init__( **kwargs: Any, ) -> None: """Init this siren.""" - self._cluster_handler: IasWdClusterHandler = cast( - IasWdClusterHandler, cluster_handlers[0] - ) - legacy_discovery_unique_id = ( f"{endpoint.device.ieee}-{endpoint.id}" if ( @@ -120,36 +418,13 @@ def __init__( WARNING_DEVICE_MODE_FIRE_PANIC: "Fire Panic", WARNING_DEVICE_MODE_EMERGENCY_PANIC: "Emergency Panic", } - self._attr_is_on: bool = False - self._off_listener: asyncio.TimerHandle | None = None - - @functools.cached_property - def info_object(self) -> SirenEntityInfo: - """Return representation of the siren.""" - return SirenEntityInfo( - **super().info_object.__dict__, - available_tones=self._attr_available_tones, - supported_features=self._attr_supported_features, - ) - @property - def state(self) -> dict[str, Any]: - """Get the state of the siren.""" - response = super().state - response["state"] = self.is_on - return response - - @property - def supported_features(self) -> SirenEntityFeature: - """Return supported features.""" - return self._attr_supported_features - - @property - def is_on(self) -> bool: - """Return true if the entity is on.""" - return self._attr_is_on - - async def async_turn_on(self, **kwargs: Any) -> None: + async def async_turn_on( + self, + duration: int | None = None, + tone: int | None = None, + volume_level: int | None = None, + ) -> None: """Turn on siren.""" if self._off_listener: self._off_listener.cancel() @@ -181,12 +456,12 @@ async def async_turn_on(self, **kwargs: Any) -> None: if strobe_level_cache is not None else WARNING_DEVICE_STROBE_HIGH ) - if (duration := kwargs.get(ATTR_DURATION)) is not None: + if duration is not None: siren_duration = duration - if (tone := kwargs.get(ATTR_TONE)) is not None: + if tone is not None: siren_tone = tone - if (level := kwargs.get(ATTR_VOLUME_LEVEL)) is not None: - siren_level = int(level) + if volume_level is not None: + siren_level = int(volume_level) await self._cluster_handler.issue_start_warning( mode=siren_tone, warning_duration=siren_duration, @@ -197,27 +472,63 @@ async def async_turn_on(self, **kwargs: Any) -> None: ) self._attr_is_on = True self._off_listener = asyncio.get_running_loop().call_later( - siren_duration, self.async_set_off + siren_duration, self._async_set_off ) self._tracked_handles.append(self._off_listener) self.maybe_emit_state_changed_event() - async def async_turn_off(self, **kwargs: Any) -> None: # pylint: disable=unused-argument - """Turn off siren.""" - await self._cluster_handler.issue_start_warning( - mode=WARNING_DEVICE_MODE_STOP, strobe=WARNING_DEVICE_STROBE_NO + +@register_entity(IasWd.cluster_id) +class BasicSiren(BaseZclSiren): + """Representation of a basic ZHA siren with fixed tone, level, and strobe.""" + + _attr_fallback_name: str = "Siren" + _attr_primary_weight = 4 + + _cluster_handler_match = ClusterHandlerMatch( + cluster_handlers=frozenset({CLUSTER_HANDLER_IAS_WD}), + exposed_features=frozenset({"siren_basic"}), + feature_priority=(PlatformFeatureGroup.SIREN, 1), + ) + + def __init__( + self, + cluster_handlers: list[ClusterHandler], + endpoint: Endpoint, + device: Device, + **kwargs: Any, + ) -> None: + """Init this basic siren.""" + super().__init__(cluster_handlers, endpoint, device, **kwargs) + self._attr_supported_features = ( + SirenEntityFeature.TURN_ON + | SirenEntityFeature.TURN_OFF + | SirenEntityFeature.DURATION ) - self._attr_is_on = False - self.maybe_emit_state_changed_event() + self._attr_available_tones: dict[int, str] = {} - def async_set_off(self) -> None: - """Set is_on to False and write HA state.""" - self._attr_is_on = False + async def async_turn_on( + self, + duration: int | None = None, + tone: int | None = None, + volume_level: int | None = None, + ) -> None: + """Turn on siren with fixed tone, level, and strobe.""" if self._off_listener: self._off_listener.cancel() - - with contextlib.suppress(ValueError): - self._tracked_handles.remove(self._off_listener) - self._off_listener = None + siren_duration = duration if duration is not None else DEFAULT_DURATION + await self._cluster_handler.issue_start_warning( + mode=WARNING_DEVICE_MODE_EMERGENCY, + warning_duration=siren_duration, + siren_level=WARNING_DEVICE_SOUND_HIGH, + strobe=WARNING_DEVICE_STROBE_NO, + strobe_duty_cycle=0, + strobe_intensity=WARNING_DEVICE_STROBE_HIGH, + ) + self._attr_is_on = True + self._off_listener = asyncio.get_running_loop().call_later( + siren_duration, self._async_set_off + ) + self._tracked_handles.append(self._off_listener) self.maybe_emit_state_changed_event()