From be8fecb01f97be6eb8eb16632361b79bbbabcbbb Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 05:05:26 +0100 Subject: [PATCH 01/10] Migrate siren platform to abstract base classes (from #662) Co-authored-by: puddly <32534428+puddly@users.noreply.github.com> --- zha/application/platforms/siren.py | 102 +++++++++++++++++++---------- 1 file changed, 67 insertions(+), 35 deletions(-) diff --git a/zha/application/platforms/siren.py b/zha/application/platforms/siren.py index 1e6933d22..fca20acb7 100644 --- a/zha/application/platforms/siren.py +++ b/zha/application/platforms/siren.py @@ -2,6 +2,7 @@ from __future__ import annotations +from abc import ABC, abstractmethod import asyncio import contextlib from dataclasses import dataclass @@ -66,11 +67,64 @@ class SirenEntityInfo(BaseEntityInfo): supported_features: SirenEntityFeature +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.""" + + @register_entity(IasWd.cluster_id) -class Siren(PlatformEntity): +class Siren(BaseSiren): """Representation of a ZHA siren.""" - PLATFORM = Platform.SIREN _attr_fallback_name: str = "Siren" _attr_primary_weight = 4 @@ -120,36 +174,14 @@ 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 +213,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, @@ -202,7 +234,7 @@ async def async_turn_on(self, **kwargs: Any) -> None: 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 + 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 From d6da8a8a07b6b99bc63f5ed0bae49bf4004d905d Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 05:19:55 +0100 Subject: [PATCH 02/10] Add `BasicSiren` class --- zha/application/platforms/__init__.py | 3 + zha/application/platforms/siren.py | 82 +++++++++++++++++++++++++++ 2 files changed, 85 insertions(+) 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 fca20acb7..ce75e34bb 100644 --- a/zha/application/platforms/siren.py +++ b/zha/application/platforms/siren.py @@ -31,6 +31,7 @@ BaseEntityInfo, ClusterHandlerMatch, PlatformEntity, + PlatformFeatureGroup, register_entity, ) from zha.zigbee.cluster_handlers.const import CLUSTER_HANDLER_IAS_WD @@ -130,6 +131,7 @@ class Siren(BaseSiren): _cluster_handler_match = ClusterHandlerMatch( cluster_handlers=frozenset({CLUSTER_HANDLER_IAS_WD}), + feature_priority=(PlatformFeatureGroup.SIREN, 0), ) def __init__( @@ -253,3 +255,83 @@ def async_set_off(self) -> None: self._off_listener = None self.maybe_emit_state_changed_event() + + +@register_entity(IasWd.cluster_id) +class BasicSiren(BaseSiren): + """Representation of a basic ZHA siren with no 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.""" + self._cluster_handler: IasWdClusterHandler = cast( + IasWdClusterHandler, cluster_handlers[0] + ) + super().__init__(cluster_handlers, endpoint, device, **kwargs) + self._attr_supported_features = ( + SirenEntityFeature.TURN_ON + | SirenEntityFeature.TURN_OFF + | SirenEntityFeature.DURATION + ) + self._attr_available_tones: dict[int, str] = {} + self._off_listener: asyncio.TimerHandle | None = None + + 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() + 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() + + 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() From e1689981d352842700b847e7d26122c494673ea7 Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 05:20:04 +0100 Subject: [PATCH 03/10] Add test --- tests/test_siren.py | 77 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/tests/test_siren.py b/tests/test_siren.py index 746a79926..2a361c884 100644 --- a/tests/test_siren.py +++ b/tests/test_siren.py @@ -25,6 +25,7 @@ async def siren_mock( zha_gateway: Gateway, + basic: bool = False, ) -> tuple[Device, security.IasWd]: """Siren fixture.""" @@ -40,6 +41,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 +123,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) From 466eb6bd320206c91a02cd553cdd6cb2a01613ad Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 05:35:57 +0100 Subject: [PATCH 04/10] Also add `BaseZclSiren` class --- zha/application/platforms/siren.py | 98 ++++++++++++++---------------- 1 file changed, 44 insertions(+), 54 deletions(-) diff --git a/zha/application/platforms/siren.py b/zha/application/platforms/siren.py index ce75e34bb..a86fb867e 100644 --- a/zha/application/platforms/siren.py +++ b/zha/application/platforms/siren.py @@ -122,9 +122,48 @@ 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() + + @register_entity(IasWd.cluster_id) -class Siren(BaseSiren): - """Representation of a ZHA siren.""" +class AdvancedSiren(BaseZclSiren): + """Representation of a ZHA siren with full tone, level, and strobe support.""" _attr_fallback_name: str = "Siren" _attr_primary_weight = 4 @@ -142,10 +181,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 ( @@ -176,7 +211,6 @@ def __init__( WARNING_DEVICE_MODE_FIRE_PANIC: "Fire Panic", WARNING_DEVICE_MODE_EMERGENCY_PANIC: "Emergency Panic", } - self._off_listener: asyncio.TimerHandle | None = None async def async_turn_on( self, @@ -231,35 +265,15 @@ async def async_turn_on( ) 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) -> 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() - @register_entity(IasWd.cluster_id) -class BasicSiren(BaseSiren): - """Representation of a basic ZHA siren with no tone, level, and strobe.""" +class BasicSiren(BaseZclSiren): + """Representation of a basic ZHA siren with fixed tone, level, and strobe.""" _attr_fallback_name: str = "Siren" _attr_primary_weight = 4 @@ -278,9 +292,6 @@ def __init__( **kwargs: Any, ) -> None: """Init this basic siren.""" - self._cluster_handler: IasWdClusterHandler = cast( - IasWdClusterHandler, cluster_handlers[0] - ) super().__init__(cluster_handlers, endpoint, device, **kwargs) self._attr_supported_features = ( SirenEntityFeature.TURN_ON @@ -288,7 +299,6 @@ def __init__( | SirenEntityFeature.DURATION ) self._attr_available_tones: dict[int, str] = {} - self._off_listener: asyncio.TimerHandle | None = None async def async_turn_on( self, @@ -315,23 +325,3 @@ async def async_turn_on( ) self._tracked_handles.append(self._off_listener) self.maybe_emit_state_changed_event() - - 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() From e4560d76321c7f619bffba62d33c92ba435d96ce Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 06:20:02 +0100 Subject: [PATCH 05/10] Add siren class for quirks v2 switch metadata --- zha/application/discovery.py | 1 + zha/application/platforms/siren.py | 113 ++++++++++++++++++++++++++++- 2 files changed, 113 insertions(+), 1 deletion(-) diff --git a/zha/application/discovery.py b/zha/application/discovery.py index 493b48be4..ec923b364 100644 --- a/zha/application/discovery.py +++ b/zha/application/discovery.py @@ -107,6 +107,7 @@ (Platform.SENSOR, ZCLSensorMetadata): sensor.Sensor, (Platform.SELECT, ZCLEnumMetadata): select.ZCLEnumSelectEntity, (Platform.NUMBER, NumberMetadata): number.NumberConfigurationEntity, + (Platform.SIREN, SwitchMetadata): siren.ConfigurableAttributeSiren, (Platform.SWITCH, SwitchMetadata): switch.ConfigurableAttributeSwitch, } diff --git a/zha/application/platforms/siren.py b/zha/application/platforms/siren.py index a86fb867e..9e41d8421 100644 --- a/zha/application/platforms/siren.py +++ b/zha/application/platforms/siren.py @@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any, Final, cast from zigpy.profiles import zha +from zigpy.quirks.v2 import SwitchMetadata from zigpy.zcl.clusters.security import IasWd from zha.application import Platform @@ -34,7 +35,11 @@ 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: @@ -161,6 +166,112 @@ def _async_set_off(self) -> 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() + + @register_entity(IasWd.cluster_id) class AdvancedSiren(BaseZclSiren): """Representation of a ZHA siren with full tone, level, and strobe support.""" From 4c1b9f33274dc62ead44e2eb9519628bbb0b917c Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 06:20:56 +0100 Subject: [PATCH 06/10] Add test --- tests/test_siren.py | 104 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 102 insertions(+), 2 deletions(-) diff --git a/tests/test_siren.py b/tests/test_siren.py index 2a361c884..ffde8c9f8 100644 --- a/tests/test_siren.py +++ b/tests/test_siren.py @@ -1,10 +1,14 @@ """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 +from zigpy.typing import UNDEFINED from zigpy.zcl.clusters import general, security import zigpy.zcl.foundation as zcl_f @@ -16,10 +20,15 @@ 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, + SirenEntityFeature, +) from zha.zigbee.device import Device @@ -228,3 +237,94 @@ 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, + ) + ] From 290213c42591f49f00d83935fe265769021daee6 Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 06:23:32 +0100 Subject: [PATCH 07/10] Add other tests from `test_switch.py` --- tests/test_siren.py | 280 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 280 insertions(+) diff --git a/tests/test_siren.py b/tests/test_siren.py index ffde8c9f8..b33b9c032 100644 --- a/tests/test_siren.py +++ b/tests/test_siren.py @@ -8,6 +8,7 @@ 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 @@ -328,3 +329,282 @@ async def test_siren_configurable_attribute(zha_gateway: Gateway) -> None: 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, + ) + ] From 84452df481a288bfa962bf24f07ea98ca2a84a78 Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 06:34:23 +0100 Subject: [PATCH 08/10] Add quirks v2 siren based on `EnumMetadata` --- zha/application/discovery.py | 1 + zha/application/platforms/siren.py | 100 ++++++++++++++++++++++++++++- 2 files changed, 99 insertions(+), 2 deletions(-) diff --git a/zha/application/discovery.py b/zha/application/discovery.py index ec923b364..4b6fe5acc 100644 --- a/zha/application/discovery.py +++ b/zha/application/discovery.py @@ -108,6 +108,7 @@ (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/siren.py b/zha/application/platforms/siren.py index 9e41d8421..7cd2d198b 100644 --- a/zha/application/platforms/siren.py +++ b/zha/application/platforms/siren.py @@ -6,12 +6,12 @@ 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 +from zigpy.quirks.v2 import SwitchMetadata, ZCLEnumMetadata from zigpy.zcl.clusters.security import IasWd from zha.application import Platform @@ -272,6 +272,102 @@ async def async_update(self) -> None: 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.""" From 84ef03fe690c0c3d70e5c05e9dde2b5f18ce73f8 Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 06:34:35 +0100 Subject: [PATCH 09/10] WIP: Add test --- tests/test_siren.py | 113 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 113 insertions(+) diff --git a/tests/test_siren.py b/tests/test_siren.py index b33b9c032..51c806c90 100644 --- a/tests/test_siren.py +++ b/tests/test_siren.py @@ -28,6 +28,7 @@ from zha.application.gateway import Gateway from zha.application.platforms.siren import ( ConfigurableAttributeSiren, + EnumSiren, SirenEntityFeature, ) from zha.zigbee.device import Device @@ -608,3 +609,115 @@ async def test_siren_configurable_attribute_inverter_attribute( manufacturer=UNDEFINED, ) ] + + +async def test_siren_enum(zha_gateway: Gateway) -> None: + """Test ZHA enum siren created from quirks v2 ZCLEnumMetadata. + + Verifies that: + - enum entry 0 (Unknown) is the off state + - enum entry 1 (Mains_single_phase) is the default on tone + - remaining entries are exposed as available tones + - specific tone can be requested via async_turn_on(tone=...) + """ + + # Use Basic cluster because it has a real ZCL attribute (power_source) + # with an associated enum type (PowerSource). + PowerSource = general.Basic.PowerSource + attr_name = general.Basic.AttributeDefs.power_source.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, + PowerSource, + general.Basic.cluster_id, + entity_platform=EntityPlatform.SIREN, + translation_key="power_mode", + fallback_name="Power 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: PowerSource.Unknown} + 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 (Unknown) = off; remaining entries are available tones + entries = list(PowerSource) + expected_tones = {e.value: e.name.replace("_", " ") for e in entries[1:]} + assert entity.available_tones == expected_tones + assert entity.supported_features == ( + SirenEntityFeature.TURN_ON + | SirenEntityFeature.TURN_OFF + | SirenEntityFeature.TONES + ) + + # Initial state: Unknown (0) -> off + assert entity.state["state"] is False + + # Attribute report: Mains_single_phase -> on + await send_attributes_report( + zha_gateway, cluster, {attr_name: PowerSource.Mains_single_phase} + ) + assert entity.state["state"] is True + + # Attribute report: Unknown -> off + await send_attributes_report(zha_gateway, cluster, {attr_name: PowerSource.Unknown}) + assert entity.state["state"] is False + + # Turn on without tone: writes entry 1 (Mains_single_phase) + 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: PowerSource.Mains_single_phase}, manufacturer=UNDEFINED) + ] + cluster.write_attributes.reset_mock() + + # Turn on with a specific tone (Battery = value 3) + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + await entity.async_turn_on(tone=PowerSource.Battery.value) + await zha_gateway.async_block_till_done() + assert cluster.write_attributes.mock_calls == [ + call({attr_name: PowerSource.Battery}, manufacturer=UNDEFINED) + ] + cluster.write_attributes.reset_mock() + + # Turn off: writes entry 0 (Unknown) + 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: PowerSource.Unknown}, manufacturer=UNDEFINED) + ] From cd079d877d367a8f8c145584e1d7d12b0377dd93 Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sat, 21 Feb 2026 06:37:27 +0100 Subject: [PATCH 10/10] Change test to use custom enum --- tests/test_siren.py | 62 ++++++++++++++++++++++----------------------- 1 file changed, 31 insertions(+), 31 deletions(-) diff --git a/tests/test_siren.py b/tests/test_siren.py index 51c806c90..048573eb6 100644 --- a/tests/test_siren.py +++ b/tests/test_siren.py @@ -614,17 +614,18 @@ async def test_siren_configurable_attribute_inverter_attribute( async def test_siren_enum(zha_gateway: Gateway) -> None: """Test ZHA enum siren created from quirks v2 ZCLEnumMetadata. - Verifies that: - - enum entry 0 (Unknown) is the off state - - enum entry 1 (Mains_single_phase) is the default on tone - - remaining entries are exposed as available tones - - specific tone can be requested via async_turn_on(tone=...) + 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 """ - # Use Basic cluster because it has a real ZCL attribute (power_source) - # with an associated enum type (PowerSource). - PowerSource = general.Basic.PowerSource - attr_name = general.Basic.AttributeDefs.power_source.name + 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, @@ -644,11 +645,11 @@ async def test_siren_enum(zha_gateway: Gateway) -> None: QuirkBuilder(zigpy_dev.manufacturer, zigpy_dev.model) .enum( attr_name, - PowerSource, + SirenMode, general.Basic.cluster_id, entity_platform=EntityPlatform.SIREN, - translation_key="power_mode", - fallback_name="Power mode", + translation_key="siren_mode", + fallback_name="Siren mode", ) .add_to_registry() ) @@ -657,37 +658,36 @@ async def test_siren_enum(zha_gateway: Gateway) -> None: assert isinstance(zigpy_device_, CustomDeviceV2) cluster = zigpy_device_.endpoints[1].basic - cluster.PLUGGED_ATTR_READS = {attr_name: PowerSource.Unknown} + 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 (Unknown) = off; remaining entries are available tones - entries = list(PowerSource) - expected_tones = {e.value: e.name.replace("_", " ") for e in entries[1:]} - assert entity.available_tones == expected_tones + # 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: Unknown (0) -> off + # Initial state: Off (0) -> off assert entity.state["state"] is False - # Attribute report: Mains_single_phase -> on - await send_attributes_report( - zha_gateway, cluster, {attr_name: PowerSource.Mains_single_phase} - ) + # Attribute report: Alert -> on + await send_attributes_report(zha_gateway, cluster, {attr_name: SirenMode.Tone_1}) assert entity.state["state"] is True - # Attribute report: Unknown -> off - await send_attributes_report(zha_gateway, cluster, {attr_name: PowerSource.Unknown}) + # 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 (Mains_single_phase) + # Turn on without tone: writes entry 1 (Alert) with patch( "zigpy.zcl.Cluster.write_attributes", return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], @@ -695,23 +695,23 @@ async def test_siren_enum(zha_gateway: Gateway) -> None: await entity.async_turn_on() await zha_gateway.async_block_till_done() assert cluster.write_attributes.mock_calls == [ - call({attr_name: PowerSource.Mains_single_phase}, manufacturer=UNDEFINED) + call({attr_name: SirenMode.Tone_1}, manufacturer=UNDEFINED) ] cluster.write_attributes.reset_mock() - # Turn on with a specific tone (Battery = value 3) + # 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=PowerSource.Battery.value) + 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: PowerSource.Battery}, manufacturer=UNDEFINED) + call({attr_name: SirenMode.Tone_2}, manufacturer=UNDEFINED) ] cluster.write_attributes.reset_mock() - # Turn off: writes entry 0 (Unknown) + # Turn off: writes entry 0 (Off) with patch( "zigpy.zcl.Cluster.write_attributes", return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], @@ -719,5 +719,5 @@ async def test_siren_enum(zha_gateway: Gateway) -> None: await entity.async_turn_off() await zha_gateway.async_block_till_done() assert cluster.write_attributes.mock_calls == [ - call({attr_name: PowerSource.Unknown}, manufacturer=UNDEFINED) + call({attr_name: SirenMode.Off}, manufacturer=UNDEFINED) ]