Skip to content
Draft
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions tests/test_switch.py
Original file line number Diff line number Diff line change
Expand Up @@ -726,6 +726,50 @@ 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)
]


WCAttrs = closures.WindowCovering.AttributeDefs
WCT = closures.WindowCovering.WindowCoveringType
WCCS = closures.WindowCovering.ConfigStatus
Expand Down
24 changes: 21 additions & 3 deletions zha/application/platforms/switch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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:
Expand Down Expand Up @@ -437,6 +439,8 @@ def __init__(
self._force_inverted = force_inverted
self._off_value = off_value
self._on_value = on_value
if mask is not None:
self._mask = mask
Comment on lines +442 to +448

super().__init__(
endpoint=endpoint,
Expand Down Expand Up @@ -507,7 +511,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:
Expand All @@ -529,7 +535,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}
)
Expand Down Expand Up @@ -1267,11 +1283,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
Expand Down
Loading