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
|
"""YoLink home manager."""
from __future__ import annotations
import logging
from typing import Any
from .auth_mgr import YoLinkAuthMgr
from .client import YoLinkClient
from .const import ATTR_DEVICE_WATER_DEPTH_SENSOR
from .device import YoLinkDevice, YoLinkDeviceMode
from .exception import YoLinkClientError, YoLinkUnSupportedMethodError
from .message_listener import MessageListener
from .model import BRDP
from .mqtt_client import YoLinkMqttClient
from .endpoint import Endpoint, Endpoints
_LOGGER = logging.getLogger(__name__)
has_external_data_devices = [ATTR_DEVICE_WATER_DEPTH_SENSOR]
class YoLinkHome:
"""YoLink home manager."""
def __init__(self) -> None:
"""Init YoLink Home Manager."""
self._home_devices: dict[str, YoLinkDevice] = {}
self._http_client: YoLinkClient
self._endpoints: dict[str, Endpoint] = {}
self._mqtt_clients: dict[str, YoLinkMqttClient] = {}
self._message_listener: MessageListener
async def async_setup(
self, auth_mgr: YoLinkAuthMgr, listener: MessageListener
) -> None:
"""Init YoLink home."""
if not auth_mgr:
raise YoLinkClientError("-1001", "setup failed, auth_mgr is required!")
if not listener:
raise YoLinkClientError(
"-1002", "setup failed, message listener is required!"
)
self._http_client = YoLinkClient(auth_mgr)
home_info: BRDP = await self.async_get_home_info()
# load home devices
await self.async_load_home_devices()
# setup yolink mqtt connection
self._message_listener = listener
# setup yolink mqtt clients
for endpoint in self._endpoints.values():
endpoint_mqtt_client = YoLinkMqttClient(
auth_manager=auth_mgr,
endpoint=endpoint,
broker_host=endpoint.mqtt_broker_host,
broker_port=endpoint.mqtt_broker_port,
devices=self._home_devices,
)
await endpoint_mqtt_client.connect(
f"yl-home/{home_info.data['id']}/+/report", self._message_listener
)
self._mqtt_clients[endpoint.name] = endpoint_mqtt_client
async def async_unload(self) -> None:
"""Unload YoLink home."""
self._home_devices = {}
self._http_client = None # type: ignore
for endpoint, client in self._mqtt_clients.items():
_LOGGER.info(
"[%s] shutting down yolink mqtt client.",
endpoint,
)
await client.disconnect()
_LOGGER.info(
"[%s] yolink mqtt client disconnected.",
endpoint,
)
self._message_listener = None # type: ignore
self._mqtt_clients = {}
async def async_get_home_info(self, **kwargs: Any) -> BRDP:
"""Get home general information."""
return await self._http_client.execute(
url=Endpoints.US.value.url, # type: ignore
bsdp={"method": "Home.getGeneralInfo"},
**kwargs,
)
async def async_load_home_devices(self, **kwargs: Any) -> dict[str, YoLinkDevice]:
"""Get home devices."""
# sync eu devices, will remove in future
eu_response: BRDP = await self._http_client.execute(
url=Endpoints.EU.value.url, # type: ignore
bsdp={"method": "Home.getDeviceList"},
**kwargs,
)
response: BRDP = await self._http_client.execute(
url=Endpoints.US.value.url, # type: ignore
bsdp={"method": "Home.getDeviceList"},
**kwargs,
)
eu_dev_tokens = {}
for eu_device in eu_response.data["devices"]:
eu_dev_tokens[eu_device["deviceId"]] = eu_device["token"]
for _device in response.data["devices"]:
_yl_device = YoLinkDevice(YoLinkDeviceMode(**_device), self._http_client)
if _yl_device.device_endpoint == Endpoints.EU.value: # type: ignore
# sync eu device token
_yl_device.device_token = eu_dev_tokens.get(_yl_device.device_id) # type: ignore
self._endpoints[_yl_device.device_endpoint.name] = (
_yl_device.device_endpoint
)
if _yl_device.device_type in has_external_data_devices:
try:
dev_external_data_resp = await _yl_device.get_external_data()
_yl_device.device_attrs = dev_external_data_resp.data.get("extData")
except YoLinkUnSupportedMethodError:
_LOGGER.debug(
"getExternalData is not supported for: %s",
_yl_device.device_type,
)
self._home_devices[_device["deviceId"]] = _yl_device
return self._home_devices
def get_devices(self) -> list[YoLinkDevice]:
"""Get home devices."""
return list(self._home_devices.values())
def get_device(self, device_id: str) -> YoLinkDevice | None:
"""Get home device via device id."""
return self._home_devices.get(device_id)
|
