Skip to content
76 changes: 75 additions & 1 deletion zha/application/platforms/switch.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from zha.zigbee.cluster_handlers.const import (
CLUSTER_HANDLER_ATTRIBUTE_UPDATED,
CLUSTER_HANDLER_BASIC,
CLUSTER_HANDLER_BINARY_OUTPUT,
CLUSTER_HANDLER_COVER,
CLUSTER_HANDLER_INOVELLI,
CLUSTER_HANDLER_ON_OFF,
Expand All @@ -36,7 +37,7 @@
from zha.zigbee.group import Group

if TYPE_CHECKING:
from zha.zigbee.cluster_handlers import ClusterHandler
from zha.zigbee.cluster_handlers import BinaryOutputClusterHandler, ClusterHandler
from zha.zigbee.device import Device
from zha.zigbee.endpoint import Endpoint

Expand Down Expand Up @@ -140,6 +141,79 @@ def handle_cluster_handler_attribute_updated(
self.maybe_emit_state_changed_event()


@STRICT_MATCH(cluster_handler_names=CLUSTER_HANDLER_BINARY_OUTPUT)
class BinaryOutputSwitch(PlatformEntity):
"""BinaryOutputCluster switch."""

PLATFORM = Platform.SWITCH
_attr_primary_weight = 10

def __init__(
self,
cluster_handlers: list[ClusterHandler],
endpoint: Endpoint,
device: Device,
**kwargs: Any,
) -> None:
"""Initialize the switch."""
super().__init__(cluster_handlers, endpoint, device, **kwargs)
self._binary_output_cluster_handler: BinaryOutputClusterHandler = (
self.cluster_handlers[CLUSTER_HANDLER_BINARY_OUTPUT]
)

def _is_supported(self) -> bool:
if self._cluster_handler.description is None:
Comment thread
puddly marked this conversation as resolved.
Outdated
return False

return super()._is_supported()

def recompute_capabilities(self) -> None:
"""Recompute capabilities."""
super().recompute_capabilities()
self._attr_fallback_name = self._binary_output_cluster_handler.description

def on_add(self) -> None:
"""Run when entity is added."""
super().on_add()
self._on_remove_callbacks.append(
self._binary_output_cluster_handler.on_event(
CLUSTER_HANDLER_ATTRIBUTE_UPDATED,
self.handle_cluster_handler_attribute_updated,
)
)

@property
def state(self) -> dict[str, Any]:
"""Return the state of the switch."""
response = super().state
response["state"] = self.is_on
return response

@property
def is_on(self) -> bool:
"""Return if the switch is on based on the statemachine."""
if self._binary_output_cluster_handler.present_value is None:
return False
return self._binary_output_cluster_handler.present_value

async def async_turn_on(self, **kwargs: Any) -> None: # pylint: disable=unused-argument
"""Turn the entity on."""
await self._binary_output_cluster_handler.async_set_present_value(True)
self.maybe_emit_state_changed_event()

async def async_turn_off(self, **kwargs: Any) -> None: # pylint: disable=unused-argument
"""Turn the entity off."""
await self._binary_output_cluster_handler.async_set_present_value(False)
self.maybe_emit_state_changed_event()

def handle_cluster_handler_attribute_updated(
self,
event: ClusterAttributeUpdatedEvent, # pylint: disable=unused-argument
) -> None:
"""Handle state update from cluster handler."""
self.maybe_emit_state_changed_event()


@GROUP_MATCH()
class SwitchGroup(GroupEntity, BaseSwitch):
"""Representation of a switch group."""
Expand Down
1 change: 1 addition & 0 deletions zha/application/registries.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
# a different dict that is keyed by manufacturer
zcl.clusters.general.AnalogOutput.cluster_id: Platform.NUMBER,
zcl.clusters.general.AnalogInput.cluster_id: Platform.SENSOR,
zcl.clusters.general.BinaryOutput.cluster_id: Platform.SWITCH,
zcl.clusters.general.MultistateInput.cluster_id: Platform.SENSOR,
zcl.clusters.general.OnOff.cluster_id: Platform.SWITCH,
zcl.clusters.hvac.Fan.cluster_id: Platform.FAN,
Expand Down
1 change: 1 addition & 0 deletions zha/zigbee/cluster_handlers/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@

CLUSTER_HANDLER_ACCELEROMETER: Final[str] = "accelerometer"
CLUSTER_HANDLER_BINARY_INPUT: Final[str] = "binary_input"
CLUSTER_HANDLER_BINARY_OUTPUT: Final[str] = "binary_output"
CLUSTER_HANDLER_ANALOG_INPUT: Final[str] = "analog_input"
CLUSTER_HANDLER_ANALOG_OUTPUT: Final[str] = "analog_output"
CLUSTER_HANDLER_ATTRIBUTE: Final[str] = "attribute"
Expand Down
26 changes: 26 additions & 0 deletions zha/zigbee/cluster_handlers/general.py
Original file line number Diff line number Diff line change
Expand Up @@ -325,6 +325,32 @@ class BinaryOutputClusterHandler(ClusterHandler):
),
)

ZCL_INIT_ATTRS = {
BinaryOutput.AttributeDefs.description.name: True,
}

@property
def description(self) -> str | None:
"""Return cached value of description."""
return self.cluster.get(BinaryOutput.AttributeDefs.description.name)

@property
def present_value(self) -> bool | None:
"""Return cached value of present_value."""
return self.cluster.get(BinaryOutput.AttributeDefs.present_value.name)

async def async_set_present_value(self, value: bool) -> None:
"""Update present_value."""
await self.write_attributes_safe(
{BinaryOutput.AttributeDefs.present_value.name: value}
)

async def async_update(self):
"""Update cluster value attribute."""
await self.get_attribute_value(
BinaryOutput.AttributeDefs.present_value.name, from_cache=False
)


@registries.CLUSTER_HANDLER_REGISTRY.register(BinaryValue.cluster_id)
class BinaryValueClusterHandler(ClusterHandler):
Expand Down
Loading