-
Notifications
You must be signed in to change notification settings - Fork 16
/
__init__.py
130 lines (103 loc) · 4.1 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
128
129
130
"""Illuminance Sensor."""
from __future__ import annotations
import asyncio
from collections.abc import Coroutine
from typing import Any
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_UNIQUE_ID,
EVENT_CORE_CONFIG_UPDATE,
SERVICE_RELOAD,
Platform,
)
from homeassistant.core import Event, HomeAssistant, ServiceCall
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.reload import async_integration_yaml_config
from homeassistant.helpers.service import async_register_admin_service
from homeassistant.helpers.sun import get_astral_location
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN
from .sensor import ILLUMINANCE_SCHEMA
_ILLUMINANCE_SCHEMA = vol.Schema(
ILLUMINANCE_SCHEMA
| {
vol.Required(CONF_UNIQUE_ID): cv.string,
}
)
CONFIG_SCHEMA = vol.Schema(
{
vol.Optional(DOMAIN): vol.All(
vol.All(
lambda config: config if config != {} else [],
cv.ensure_list,
),
[_ILLUMINANCE_SCHEMA],
)
},
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = [Platform.SENSOR]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up composite integration."""
async def async_get_loc_elev(event: Event | None = None) -> None:
"""Get HA Location object & elevation."""
if event is not None and not event.data:
return
def get_loc_elev() -> None:
"""Get HA Location object & elevation.
Then get Location's tzinfo to force pytz, which it calls indirectly, to do
its file I/O that it does when it sees a new time zone. This needs to be
done in an executor.
"""
loc, elv = get_astral_location(hass)
loc.tzinfo # noqa: B018
hass.data[DOMAIN] = loc, elv
await hass.async_add_executor_job(get_loc_elev)
async def process_config(
config: ConfigType | None, run_immediately: bool = True
) -> None:
"""Process illuminance config."""
if not config or not (configs := config.get(DOMAIN)):
configs = []
unique_ids = [config[CONF_UNIQUE_ID] for config in configs]
tasks: list[Coroutine[Any, Any, Any]] = []
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.source != SOURCE_IMPORT:
continue
if entry.unique_id not in unique_ids:
tasks.append(hass.config_entries.async_remove(entry.entry_id))
for conf in configs:
tasks.append(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=conf.copy()
)
)
if not tasks:
return
if run_immediately:
await asyncio.gather(*tasks)
else:
for task in tasks:
hass.async_create_task(task)
async def reload_config(call: ServiceCall | None = None) -> None:
"""Reload configuration."""
await process_config(await async_integration_yaml_config(hass, DOMAIN))
await async_get_loc_elev()
await process_config(config, run_immediately=False)
async_register_admin_service(hass, DOMAIN, SERVICE_RELOAD, reload_config)
hass.bus.async_listen(EVENT_CORE_CONFIG_UPDATE, async_get_loc_elev)
return True
async def entry_updated(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle config entry update."""
if not entry.state.recoverable:
return
await hass.config_entries.async_reload(entry.entry_id)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up config entry."""
entry.async_on_unload(entry.add_update_listener(entry_updated))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)