-
Notifications
You must be signed in to change notification settings - Fork 5
/
groups.py
122 lines (99 loc) · 4.22 KB
/
groups.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import logging
LOGGER = logging.getLogger(__name__)
async def get_groups(app, listener, ieee, cmd, data, service):
LOGGER.debug("running 'get_groups' command: %s", service)
if ieee is None:
LOGGER.error("missing ieee")
return
src_dev = app.get_device(ieee=ieee)
for ep_id, ep in src_dev.endpoints.items():
if ep_id == 0:
continue
try:
name_support = await ep.groups.read_attributes(["name_support"])
LOGGER.debug("Group on 0x%04x name support: %s", src_dev.nwk, name_support)
all_groups = await ep.groups.get_membership([])
LOGGER.debug("Groups on 0x%04x : %s", src_dev.nwk, all_groups)
except AttributeError:
LOGGER.debug("0x%04x: no group cluster found", src_dev.nwk)
async def add_group(app, listener, ieee, cmd, data, service):
LOGGER.debug("running 'add group' command: %s", service)
if ieee is None or not data:
return
src_dev = app.get_device(ieee=ieee)
group_id = int(data, base=16)
for ep_id, ep in src_dev.endpoints.items():
if ep_id == 0:
continue
try:
res = await ep.groups.add(group_id, "group {}".format(group_id))
LOGGER.debug("0x%04x: Setting group 0x%04x: %s", src_dev.nwk, group_id, res)
except AttributeError:
LOGGER.debug("0x%04x: no group cluster found", src_dev.nwk)
async def remove_group(app, listener, ieee, cmd, data, service):
LOGGER.debug("running 'remove group' command: %s", service)
if ieee is None or not data:
LOGGER.error("missing ieee")
return
src_dev = app.get_device(ieee=ieee)
group_id = int(data, base=16)
for ep_id, ep in src_dev.endpoints.items():
if ep_id == 0:
continue
try:
res = await ep.groups.remove(group_id)
LOGGER.debug(
"0x%04x: Removing group 0x%04x: %s", src_dev.nwk, group_id, res
)
except AttributeError:
LOGGER.debug("0x%04x: no group cluster found", src_dev.nwk)
async def remove_all_groups(app, listener, ieee, cmd, data, service):
LOGGER.debug("running 'remove all group' command: %s", service)
if ieee is None:
return
src_dev = app.get_device(ieee=ieee)
for ep_id, ep in src_dev.endpoints.items():
if ep_id == 0:
continue
try:
res = await ep.groups.remove_all()
LOGGER.debug("0x%04x: Removing all groups: %s", src_dev.nwk, res)
except AttributeError:
LOGGER.debug("0x%04x: no group cluster on endpoint #%d", src_dev.nwk, ep_id)
async def add_to_group(app, listener, ieee, cmd, data, service):
if data is None or ieee is None:
LOGGER.error("invalid arguments for subscribe_group()")
return
dev = app.get_device(ieee)
grp_id = int(data, base=16)
LOGGER.debug("Subscribing EZSP to %s group: %s", grp_id, service)
res = await dev.add_to_group(grp_id, "Group {}".format(data))
LOGGER.info("Subscribed NCP to %s group: %s", grp_id, res)
async def remove_from_group(app, listener, ieee, cmd, data, service):
if data is None or ieee is None:
LOGGER.error("invalid arguments for unsubscribe_group()")
return
dev = app.get_device(ieee)
grp_id = int(data, base=16)
LOGGER.debug("Unsubscribing EZSP to %s group: %s", grp_id, service)
res = await dev.remove_from_group(grp_id)
LOGGER.info("Unsubscribed NCP to %s group: %s", grp_id, res)
async def get_zll_groups(app, listener, ieee, cmd, data, service):
from zigpy.zcl.clusters.lightlink import LightLink
if ieee is None:
LOGGER.error("missine ieee")
return
LOGGER.debug("Getting ZLL groups: %s", service)
dev = app.get_device(ieee=ieee)
clusters = [
ep.in_clusters[LightLink.cluster_id]
for epid, ep in dev.endpoints.items()
if epid and LightLink.cluster_id in ep.in_clusters
]
zll_cluster = next(iter(clusters))
if not zll_cluster:
LOGGER.warning("Couldn't find ZLL Commissioning cluster on %s", dev.ieee)
return
res = await zll_cluster.get_group_identifiers(0)
groups = res[2]
LOGGER.debug("Get group identifiers response: %s", [g.group_id for g in groups])