-
-
Notifications
You must be signed in to change notification settings - Fork 15
/
__init__.py
127 lines (101 loc) · 3.79 KB
/
__init__.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
"""The SolarEdge Modbus Integration."""
import logging
from datetime import timedelta
from typing import Any
import async_timeout
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PORT,
CONF_SCAN_INTERVAL,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
CONF_DETECT_BATTERIES,
CONF_DETECT_METERS,
CONF_DEVICE_ID,
CONF_KEEP_MODBUS_OPEN,
CONF_NUMBER_INVERTERS,
CONF_SINGLE_DEVICE_ENTITY,
DEFAULT_DETECT_BATTERIES,
DEFAULT_DETECT_METERS,
DEFAULT_KEEP_MODBUS_OPEN,
DEFAULT_SCAN_INTERVAL,
DEFAULT_SINGLE_DEVICE_ENTITY,
DOMAIN,
)
from .hub import DataUpdateFailed, HubInitFailed, SolarEdgeModbusMultiHub
_LOGGER = logging.getLogger(__name__)
PLATFORMS: list[str] = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up SolarEdge Modbus from a config entry."""
entry_updates: dict[str, Any] = {}
if CONF_SCAN_INTERVAL in entry.data:
data = {**entry.data}
entry_updates["data"] = data
entry_updates["options"] = {
**entry.options,
CONF_SCAN_INTERVAL: data.pop(CONF_SCAN_INTERVAL),
}
if entry_updates:
hass.config_entries.async_update_entry(entry, **entry_updates)
solaredge_hub = SolarEdgeModbusMultiHub(
hass,
entry.data[CONF_NAME],
entry.data[CONF_HOST],
entry.data[CONF_PORT],
entry.data.get(CONF_NUMBER_INVERTERS, 1),
entry.data.get(CONF_DEVICE_ID, 1),
entry.options.get(CONF_DETECT_METERS, DEFAULT_DETECT_METERS),
entry.options.get(CONF_DETECT_BATTERIES, DEFAULT_DETECT_BATTERIES),
entry.options.get(CONF_SINGLE_DEVICE_ENTITY, DEFAULT_SINGLE_DEVICE_ENTITY),
entry.options.get(CONF_KEEP_MODBUS_OPEN, DEFAULT_KEEP_MODBUS_OPEN),
)
coordinator = SolarEdgeCoordinator(
hass,
solaredge_hub,
entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL),
)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {
"hub": solaredge_hub,
"coordinator": coordinator,
}
await coordinator.async_config_entry_first_refresh()
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
solaredge_hub = hass.data[DOMAIN][entry.entry_id]["hub"]
await solaredge_hub.shutdown()
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle an options update."""
await hass.config_entries.async_reload(entry.entry_id)
class SolarEdgeCoordinator(DataUpdateCoordinator):
def __init__(self, hass, hub, scan_interval):
super().__init__(
hass,
_LOGGER,
name="SolarEdge Coordinator",
update_interval=timedelta(seconds=scan_interval),
)
self._hub = hub
if scan_interval < 10:
_LOGGER.warning("Polling frequency < 10, requiring keep modbus open.")
self._hub.keep_modbus_open = True
async def _async_update_data(self):
try:
async with async_timeout.timeout(5):
return await self._hub.async_refresh_modbus_data()
except HubInitFailed as e:
raise UpdateFailed(f"{e}")
except DataUpdateFailed as e:
raise UpdateFailed(f"{e}")