-
-
Notifications
You must be signed in to change notification settings - Fork 101
/
coordinators.py
100 lines (81 loc) · 3.11 KB
/
coordinators.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
import logging
from abc import ABC
from dataclasses import dataclass
from datetime import timedelta
from typing import Dict
from typing import List
from typing import Type
from typing import TypeVar
import aiohttp
import async_timeout
from homeassistant.core import CALLBACK_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.helpers.update_coordinator import UpdateFailed
from plugp100.new.tapodevice import TapoDevice
from plugp100.new.tapohub import TapoHub
from plugp100.responses.child_device_list import PowerStripChild
from plugp100.responses.tapo_exception import TapoException
from custom_components.tapo.const import DOMAIN
from custom_components.tapo.helpers import _raise_from_tapo_exception
_LOGGER = logging.getLogger(__name__)
DEBOUNCER_COOLDOWN = 2
@dataclass
class HassTapoDeviceData:
device: TapoDevice
coordinator: "TapoDataCoordinator"
config_entry_update_unsub: CALLBACK_TYPE
child_coordinators: List["TapoDataCoordinator"]
# def create_tapo_device(model: str, client: TapoClient) -> Optional[TapoDevice]:
# if model in SUPPORTED_DEVICE_AS_SWITCH:
# return PlugDevice(client)
# if model in SUPPORTED_DEVICE_AS_LED_STRIP:
# return LedStripDevice(client)
# if model in SUPPORTED_DEVICE_AS_LIGHT:
# return LightDevice(client)
# if model in SUPPORTED_POWER_STRIP_DEVICE_MODEL:
# return PowerStripDevice(client)
# if model in SUPPORTED_HUB_DEVICE_MODEL:
# return HubDevice(client, subscription_polling_interval_millis=30_000)
# return None
T = TypeVar("T")
StateMap = Dict[Type[T], T]
class TapoDataCoordinator(ABC, DataUpdateCoordinator[StateMap]):
def __init__(
self,
hass: HomeAssistant,
device: TapoDevice,
polling_interval: timedelta,
):
self._device = device
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=polling_interval,
request_refresh_debouncer=Debouncer(
hass, _LOGGER, cooldown=DEBOUNCER_COOLDOWN, immediate=True
),
)
self._states: StateMap = {}
# TODO: expose .state from self.device as raw json
@property
def device(self) -> TapoDevice:
return self._device
@property
def is_hub(self) -> bool:
return isinstance(self.device, TapoHub)
async def _async_update_data(self) -> StateMap:
try:
async with async_timeout.timeout(10):
return await self.poll_update()
except TapoException as error:
_raise_from_tapo_exception(error, _LOGGER)
except aiohttp.ClientError as error:
raise UpdateFailed(f"Error communication with API: {str(error)}") from error
except Exception as exception:
raise UpdateFailed(f"Unexpected exception: {str(exception)}") from exception
async def poll_update(self):
return await self.device.update()
PowerStripChildrenState = dict[str, PowerStripChild]