diff --git a/tests/test_switch.py b/tests/test_switch.py index a468c2015..e08920abb 100644 --- a/tests/test_switch.py +++ b/tests/test_switch.py @@ -41,6 +41,7 @@ from zha.application import Platform from zha.application.gateway import Gateway from zha.application.platforms import GroupEntity, PlatformEntity +from zha.application.platforms.switch import ConfigurableAttributeSwitch from zha.exceptions import ZHAException from zha.quirks import QUIRK_REGISTRY_ENTRY_ATTR, DeviceRegistry from zha.zigbee.device import Device @@ -726,6 +727,80 @@ async def test_switch_configurable_custom_on_off_values_inverter_attribute( ] +async def test_switch_configurable_bitmap_mask(zha_gateway: Gateway) -> None: + """A masked configurable switch toggles a single bit, preserving the others. + + Uses ``DanfossAdaptationRunSettings`` (``_mask = 0x01``) as a real consumer of + the ``mask`` support on ``ConfigurableAttributeSwitch``. + """ + zigpy_device_ = await zigpy_device_from_json( + zha_gateway.application_controller, + "tests/data/devices/danfoss-etrv0103-0x00000014.json", + ) + zha_device = await join_zigpy_device(zha_gateway, zigpy_device_) + + entity = get_entity( + zha_device, platform=Platform.SWITCH, qualifier="adaptation_run_settings" + ) + cluster = zigpy_device_.endpoints[1].thermostat + attr_name = "adaptation_run_settings" + + # Another (higher) bit is set on the device; the masked bit (0x01) is clear. + await send_attributes_report(zha_gateway, cluster, {attr_name: 0b10}) + assert entity.is_on is False + + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + # Turn on: set bit 0x01 while preserving the existing 0b10 bit. + await entity.async_turn_on() + await zha_gateway.async_block_till_done() + assert cluster.write_attributes.mock_calls == [ + call({attr_name: 0b11}, manufacturer=UNDEFINED) + ] + cluster.write_attributes.reset_mock() + + # Both bits now set; turning off clears only the masked bit. + await send_attributes_report(zha_gateway, cluster, {attr_name: 0b11}) + assert entity.is_on is True + await entity.async_turn_off() + await zha_gateway.async_block_till_done() + assert cluster.write_attributes.mock_calls == [ + call({attr_name: 0b10}, manufacturer=UNDEFINED) + ] + + # The `mask=` constructor argument (the path quirks v2 discovery uses) is + # honored: build a switch on the same cluster targeting the higher bit. + masked = ConfigurableAttributeSwitch( + endpoint=entity._endpoint, + device=zha_device, + cluster=cluster, + from_quirk=True, + attribute_name=attr_name, + mask=0b10, + fallback_name="Masked bit", + unique_id_suffix="masked-bit", + ) + await send_attributes_report(zha_gateway, cluster, {attr_name: 0b10}) + assert masked.is_on is True + await send_attributes_report(zha_gateway, cluster, {attr_name: 0b01}) + assert masked.is_on is False + + # A zero mask is rejected (it would make the switch a permanent no-op). + with pytest.raises(ValueError, match="mask must be a non-zero bitmask"): + ConfigurableAttributeSwitch( + endpoint=entity._endpoint, + device=zha_device, + cluster=cluster, + from_quirk=True, + attribute_name=attr_name, + mask=0, + fallback_name="Zero mask", + unique_id_suffix="zero-mask", + ) + + WCAttrs = closures.WindowCovering.AttributeDefs WCT = closures.WindowCovering.WindowCoveringType WCCS = closures.WindowCovering.ConfigStatus diff --git a/zha/application/platforms/switch.py b/zha/application/platforms/switch.py index a6474a2ce..4a4b3c381 100644 --- a/zha/application/platforms/switch.py +++ b/zha/application/platforms/switch.py @@ -396,6 +396,7 @@ class ConfigurableAttributeSwitch(PlatformEntity): _force_inverted: bool = False _off_value: int = 0 _on_value: int = 1 + _mask: int | None = None def __init__( self, @@ -409,6 +410,7 @@ def __init__( force_inverted: bool = False, off_value: int = 0, on_value: int = 1, + mask: int | None = None, legacy_discovery_unique_id: str | None = None, **kwargs: Any, ) -> None: @@ -437,6 +439,13 @@ def __init__( self._force_inverted = force_inverted self._off_value = off_value self._on_value = on_value + if mask is not None: + if not mask: + raise ValueError( + "mask must be a non-zero bitmask; a zero mask would make the " + "switch a no-op that is always off" + ) + self._mask = mask super().__init__( endpoint=endpoint, @@ -507,7 +516,9 @@ def inverted(self) -> bool: @property def is_on(self) -> bool: """Return if the switch is on based on the statemachine.""" - if self._on_value != 1: + if self._mask is not None: + val = bool((self._cluster.get(self._attribute_name) or 0) & self._mask) + elif self._on_value != 1: val = self._cluster.get(self._attribute_name) val = val == self._on_value else: @@ -529,7 +540,17 @@ async def async_turn_on_off(self, state: bool) -> None: """Turn the entity on or off.""" if self.inverted: state = not state - if state: + if self._mask is not None: + # Toggle only the masked bit(s) of the bitmap attribute, preserving the + # other bits by reading the current (shared) cluster cache value first. + # Sibling switches on the same bitmap share this cluster cache, so their + # bits accumulate correctly across sequential writes. + current = self._cluster.get(self._attribute_name) or 0 + new_value = (current | self._mask) if state else (current & ~self._mask) + await write_attributes_safe( + self._cluster, {self._attribute_name: new_value} + ) + elif state: await write_attributes_safe( self._cluster, {self._attribute_name: self._on_value} ) @@ -1267,11 +1288,13 @@ class DanfossLoadBalancingEnable(ConfigurableAttributeSwitch): class DanfossAdaptationRunSettings(ConfigurableAttributeSwitch): """Danfoss proprietary attribute for enabling daily adaptation run. - Actually a bitmap, but only the first bit is used. + The attribute is a bitmap; only the first bit is used. ``_mask`` makes the + switch toggle just that bit, preserving any other bits the device may set. """ _unique_id_suffix = "adaptation_run_settings" _attribute_name: str = "adaptation_run_settings" + _mask: int = 0x01 _attr_translation_key: str = "adaptation_run_enabled" _cluster_id = Thermostat.cluster_id