Add bitmap-bit mask support to ConfigurableAttributeSwitch

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

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