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
|
import logging
from pathlib import Path
from types import TracebackType
from typing import List, Optional, Dict, Any, Type, Callable
from aiohttp import ClientSession
from typing_extensions import Self
from pyhon.appliance import HonAppliance
from pyhon.connection.api import HonAPI
from pyhon.connection.api import TestAPI
from pyhon.connection.mqtt import MQTTClient
from pyhon.exceptions import NoAuthenticationException
_LOGGER = logging.getLogger(__name__)
# pylint: disable=too-many-instance-attributes
class Hon:
def __init__(
self,
email: Optional[str] = "",
password: Optional[str] = "",
session: Optional[ClientSession] = None,
mobile_id: str = "",
refresh_token: str = "",
test_data_path: Optional[Path] = None,
):
self._email: Optional[str] = email
self._password: Optional[str] = password
self._session: ClientSession | None = session
self._appliances: List[HonAppliance] = []
self._api: Optional[HonAPI] = None
self._test_data_path: Path = test_data_path or Path().cwd()
self._mobile_id: str = mobile_id
self._refresh_token: str = refresh_token
self._mqtt_client: MQTTClient | None = None
self._notify_function: Optional[Callable[[Any], None]] = None
async def __aenter__(self) -> Self:
return await self.create()
async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
await self.close()
@property
def api(self) -> HonAPI:
if self._api is None:
raise NoAuthenticationException
return self._api
@property
def email(self) -> str:
if not self._email:
raise ValueError("Missing email")
return self._email
@property
def password(self) -> str:
if not self._password:
raise ValueError("Missing password")
return self._password
async def create(self) -> Self:
self._api = await HonAPI(
self.email,
self.password,
session=self._session,
mobile_id=self._mobile_id,
refresh_token=self._refresh_token,
).create()
await self.setup()
return self
@property
def appliances(self) -> List[HonAppliance]:
return self._appliances
@appliances.setter
def appliances(self, appliances: List[HonAppliance]) -> None:
self._appliances = appliances
async def _create_appliance(
self, appliance_data: Dict[str, Any], api: HonAPI, zone: int = 0
) -> None:
appliance = HonAppliance(api, appliance_data, zone=zone)
if appliance.mac_address == "":
return
try:
await appliance.load_commands()
await appliance.load_attributes()
await appliance.load_statistics()
except (KeyError, ValueError, IndexError) as error:
_LOGGER.exception(error)
_LOGGER.error("Device data - %s", appliance_data)
self._appliances.append(appliance)
async def setup(self) -> None:
appliances = await self.api.load_appliances()
for appliance in appliances:
if (zones := int(appliance.get("zone", "0"))) > 1:
for zone in range(zones):
await self._create_appliance(
appliance.copy(), self.api, zone=zone + 1
)
await self._create_appliance(appliance, self.api)
if (
self._test_data_path
and (
test_data := self._test_data_path / "hon-test-data" / "test_data"
).exists()
or (test_data := test_data / "..").exists()
):
api = TestAPI(test_data)
for appliance in await api.load_appliances():
await self._create_appliance(appliance, api)
if not self._mqtt_client:
self._mqtt_client = await MQTTClient(self, self._mobile_id).create()
def subscribe_updates(self, notify_function: Callable[[Any], None]) -> None:
self._notify_function = notify_function
def notify(self) -> None:
if self._notify_function:
self._notify_function(None)
async def close(self) -> None:
await self.api.close()
|
