evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Zigbee group binding to ZHA (#30433)

Browse source 
* initial group binding work
* add group cluster binding
This commit is contained in:
David F. Mulcahey 2020-01-04 16:58:51 -05:00 • committed by Alexei Chetroi
parent 967fe89f6d
commit 6c89b6c5a2
3 changed files with 132 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

65
homeassistant/components/zha/api.py
View file

@ -1,7 +1,9 @@
"""Web socket API for Zigbee Home Automation devices."""
import asyncio
import collections
import logging
from typing import Any
import voluptuous as vol
from zigpy.types.named import EUI64
@ -31,6 +33,7 @@ from .core.const import (
ATTR_WARNING_DEVICE_STROBE,
ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE,
ATTR_WARNING_DEVICE_STROBE_INTENSITY,
BINDINGS,
CHANNEL_IAS_WD,
CLUSTER_COMMAND_SERVER,
CLUSTER_COMMANDS_CLIENT,
@ -163,6 +166,8 @@ SERVICE_SCHEMAS = {
),
}
ClusterBinding = collections.namedtuple("ClusterBinding", "id endpoint_id type name")
@websocket_api.require_admin
@websocket_api.async_response
@ -774,6 +779,64 @@ async def websocket_unbind_devices(hass, connection, msg):
)
def is_cluster_binding(value: Any) -> ClusterBinding:
"""Validate and transform a cluster binding."""
if not isinstance(value, collections.Mapping):
raise vol.Invalid("Not a cluster binding")
try:
cluster_binding = ClusterBinding(
name=value["name"],
type=value["type"],
id=value["id"],
endpoint_id=value["endpoint_id"],
)
except KeyError:
raise vol.Invalid("Not a cluster binding")
return cluster_binding
@websocket_api.require_admin
@websocket_api.async_response
@websocket_api.websocket_command(
{
vol.Required(TYPE): "zha/groups/bind",
vol.Required(ATTR_SOURCE_IEEE): EUI64.convert,
vol.Required(GROUP_ID): cv.positive_int,
vol.Required(BINDINGS): vol.All(cv.ensure_list, [is_cluster_binding]),
}
)
async def websocket_bind_group(hass, connection, msg):
"""Directly bind a device to a group."""
zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
source_ieee = msg[ATTR_SOURCE_IEEE]
group_id = msg[GROUP_ID]
bindings = msg[BINDINGS]
source_device = zha_gateway.get_device(source_ieee)
await source_device.async_bind_to_group(group_id, bindings)
@websocket_api.require_admin
@websocket_api.async_response
@websocket_api.websocket_command(
{
vol.Required(TYPE): "zha/groups/unbind",
vol.Required(ATTR_SOURCE_IEEE): EUI64.convert,
vol.Required(GROUP_ID): cv.positive_int,
vol.Required(BINDINGS): vol.All(cv.ensure_list, [is_cluster_binding]),
}
)
async def websocket_unbind_group(hass, connection, msg):
"""Unbind a device from a group."""
zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
source_ieee = msg[ATTR_SOURCE_IEEE]
group_id = msg[GROUP_ID]
bindings = msg[BINDINGS]
source_device = zha_gateway.get_device(source_ieee)
await source_device.async_unbind_from_group(group_id, bindings)
async def async_binding_operation(zha_gateway, source_ieee, target_ieee, operation):
"""Create or remove a direct zigbee binding between 2 devices."""
@ -1082,6 +1145,8 @@ def async_load_api(hass):
websocket_api.async_register_command(hass, websocket_remove_groups)
websocket_api.async_register_command(hass, websocket_add_group_members)
websocket_api.async_register_command(hass, websocket_remove_group_members)
websocket_api.async_register_command(hass, websocket_bind_group)
websocket_api.async_register_command(hass, websocket_unbind_group)
websocket_api.async_register_command(hass, websocket_reconfigure_node)
websocket_api.async_register_command(hass, websocket_device_clusters)
websocket_api.async_register_command(hass, websocket_device_cluster_attributes)