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
|
from __future__ import annotations
import logging
import zigpy.device
import zigpy.endpoint
import zigpy.profiles.zgp
import zigpy.profiles.zha
import zigpy.profiles.zll
import zigpy.zdo.types as zdo_t
import bellows.types as t
LOGGER = logging.getLogger(__name__)
PROFILE_TO_DEVICE_TYPE = {
zigpy.profiles.zha.PROFILE_ID: zigpy.profiles.zha.DeviceType,
zigpy.profiles.zll.PROFILE_ID: zigpy.profiles.zll.DeviceType,
zigpy.profiles.zgp.PROFILE_ID: zigpy.profiles.zgp.DeviceType,
}
class EZSPEndpoint(zigpy.endpoint.Endpoint):
def __init__(
self,
device: zigpy.device.Device,
endpoint_id: int,
descriptor: zdo_t.SimpleDescriptor,
) -> None:
super().__init__(device, endpoint_id)
self.profile_id = descriptor.profile
if self.profile_id in PROFILE_TO_DEVICE_TYPE:
self.device_type = PROFILE_TO_DEVICE_TYPE[self.profile_id](
descriptor.device_type
)
else:
self.device_type = descriptor.device_type
for cluster in descriptor.input_clusters:
self.add_input_cluster(cluster)
for cluster in descriptor.output_clusters:
self.add_output_cluster(cluster)
self.status = zigpy.endpoint.Status.ZDO_INIT
@property
def manufacturer(self) -> str:
"""Manufacturer."""
return "Silicon Labs"
@property
def model(self) -> str:
"""Model."""
return "EZSP"
async def add_to_group(self, grp_id: int, name: str = None) -> None:
if grp_id in self.member_of:
return
app = self.device.application
status = await app.multicast.subscribe(grp_id)
if status != t.sl_Status.OK:
raise ValueError(f"Couldn't subscribe to 0x{grp_id:04x} group")
group = app.groups.add_group(grp_id, name)
group.add_member(self)
async def remove_from_group(self, grp_id: int) -> None:
if grp_id not in self.member_of:
return
app = self.device.application
status = await app.multicast.unsubscribe(grp_id)
if status != t.sl_Status.OK:
raise ValueError(f"Couldnt't unsubscribe from 0x{grp_id:04x} group")
app.groups[grp_id].remove_member(self)
|
