forked from home-assistant/core
-
Notifications
You must be signed in to change notification settings - Fork 0
/
group.py
233 lines (197 loc) · 7.76 KB
/
group.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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
"""Group for Zigbee Home Automation."""
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING, Any, NamedTuple
import zigpy.endpoint
import zigpy.exceptions
import zigpy.group
from zigpy.types.named import EUI64
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_registry import async_entries_for_device
from .helpers import LogMixin
if TYPE_CHECKING:
from .device import ZHADevice
from .gateway import ZHAGateway
_LOGGER = logging.getLogger(__name__)
class GroupMember(NamedTuple):
"""Describes a group member."""
ieee: EUI64
endpoint_id: int
class GroupEntityReference(NamedTuple):
"""Reference to a group entity."""
name: str
original_name: str
entity_id: int
class ZHAGroupMember(LogMixin):
"""Composite object that represents a device endpoint in a Zigbee group."""
def __init__(
self, zha_group: ZHAGroup, zha_device: ZHADevice, endpoint_id: int
) -> None:
"""Initialize the group member."""
self._zha_group = zha_group
self._zha_device = zha_device
self._endpoint_id = endpoint_id
@property
def group(self) -> ZHAGroup:
"""Return the group this member belongs to."""
return self._zha_group
@property
def endpoint_id(self) -> int:
"""Return the endpoint id for this group member."""
return self._endpoint_id
@property
def endpoint(self) -> zigpy.endpoint.Endpoint:
"""Return the endpoint for this group member."""
return self._zha_device.device.endpoints.get(self.endpoint_id)
@property
def device(self) -> ZHADevice:
"""Return the zha device for this group member."""
return self._zha_device
@property
def member_info(self) -> dict[str, Any]:
"""Get ZHA group info."""
member_info: dict[str, Any] = {}
member_info["endpoint_id"] = self.endpoint_id
member_info["device"] = self.device.zha_device_info
member_info["entities"] = self.associated_entities
return member_info
@property
def associated_entities(self) -> list[GroupEntityReference]:
"""Return the list of entities that were derived from this endpoint."""
ha_entity_registry = self.device.gateway.ha_entity_registry
zha_device_registry = self.device.gateway.device_registry
return [
GroupEntityReference(
ha_entity_registry.async_get(entity_ref.reference_id).name,
ha_entity_registry.async_get(entity_ref.reference_id).original_name,
entity_ref.reference_id,
)._asdict()
for entity_ref in zha_device_registry.get(self.device.ieee)
if list(entity_ref.cluster_channels.values())[
0
].cluster.endpoint.endpoint_id
== self.endpoint_id
]
async def async_remove_from_group(self) -> None:
"""Remove the device endpoint from the provided zigbee group."""
try:
await self._zha_device.device.endpoints[
self._endpoint_id
].remove_from_group(self._zha_group.group_id)
except (zigpy.exceptions.ZigbeeException, asyncio.TimeoutError) as ex:
self.debug(
"Failed to remove endpoint: %s for device '%s' from group: 0x%04x ex: %s",
self._endpoint_id,
self._zha_device.ieee,
self._zha_group.group_id,
str(ex),
)
def log(self, level: int, msg: str, *args: Any, **kwargs) -> None:
"""Log a message."""
msg = f"[%s](%s): {msg}"
args = (f"0x{self._zha_group.group_id:04x}", self.endpoint_id) + args
_LOGGER.log(level, msg, *args, **kwargs)
class ZHAGroup(LogMixin):
"""ZHA Zigbee group object."""
def __init__(
self,
hass: HomeAssistant,
zha_gateway: ZHAGateway,
zigpy_group: zigpy.group.Group,
) -> None:
"""Initialize the group."""
self.hass = hass
self._zha_gateway = zha_gateway
self._zigpy_group = zigpy_group
@property
def name(self) -> str:
"""Return group name."""
return self._zigpy_group.name
@property
def group_id(self) -> int:
"""Return group name."""
return self._zigpy_group.group_id
@property
def endpoint(self) -> zigpy.endpoint.Endpoint:
"""Return the endpoint for this group."""
return self._zigpy_group.endpoint
@property
def members(self) -> list[ZHAGroupMember]:
"""Return the ZHA devices that are members of this group."""
return [
ZHAGroupMember(
self, self._zha_gateway.devices.get(member_ieee), endpoint_id
)
for (member_ieee, endpoint_id) in self._zigpy_group.members.keys()
if member_ieee in self._zha_gateway.devices
]
async def async_add_members(self, members: list[GroupMember]) -> None:
"""Add members to this group."""
if len(members) > 1:
tasks = []
for member in members:
tasks.append(
self._zha_gateway.devices[member.ieee].async_add_endpoint_to_group(
member.endpoint_id, self.group_id
)
)
await asyncio.gather(*tasks)
else:
await self._zha_gateway.devices[
members[0].ieee
].async_add_endpoint_to_group(members[0].endpoint_id, self.group_id)
async def async_remove_members(self, members: list[GroupMember]) -> None:
"""Remove members from this group."""
if len(members) > 1:
tasks = []
for member in members:
tasks.append(
self._zha_gateway.devices[
member.ieee
].async_remove_endpoint_from_group(
member.endpoint_id, self.group_id
)
)
await asyncio.gather(*tasks)
else:
await self._zha_gateway.devices[
members[0].ieee
].async_remove_endpoint_from_group(members[0].endpoint_id, self.group_id)
@property
def member_entity_ids(self) -> list[str]:
"""Return the ZHA entity ids for all entities for the members of this group."""
all_entity_ids: list[str] = []
for member in self.members:
entity_references = member.associated_entities
for entity_reference in entity_references:
all_entity_ids.append(entity_reference["entity_id"])
return all_entity_ids
def get_domain_entity_ids(self, domain: str) -> list[str]:
"""Return entity ids from the entity domain for this group."""
domain_entity_ids: list[str] = []
for member in self.members:
if member.device.is_coordinator:
continue
entities = async_entries_for_device(
self._zha_gateway.ha_entity_registry,
member.device.device_id,
include_disabled_entities=True,
)
domain_entity_ids.extend(
[entity.entity_id for entity in entities if entity.domain == domain]
)
return domain_entity_ids
@property
def group_info(self) -> dict[str, Any]:
"""Get ZHA group info."""
group_info: dict[str, Any] = {}
group_info["group_id"] = self.group_id
group_info["name"] = self.name
group_info["members"] = [member.member_info for member in self.members]
return group_info
def log(self, level: int, msg: str, *args: Any, **kwargs) -> None:
"""Log a message."""
msg = f"[%s](%s): {msg}"
args = (self.name, self.group_id) + args
_LOGGER.log(level, msg, *args, **kwargs)