mirror of
https://github.com/home-assistant/core.git
synced 2026-05-30 12:14:20 +01:00
Tom Wilkie
107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
"""Support for the Hive devices and services."""
|
|
|
|
from collections.abc import Awaitable, Callable, Coroutine
|
|
from functools import wraps
|
|
import logging
|
|
from typing import Any, Concatenate
|
|
|
|
from aiohttp.web_exceptions import HTTPException
|
|
from apyhiveapi import Auth, Hive
|
|
from apyhiveapi.helper.hive_exceptions import HiveReauthRequired
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import CONF_SCAN_INTERVAL
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
|
|
from homeassistant.helpers import aiohttp_client, device_registry as dr
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
|
|
|
from .const import DOMAIN, PLATFORM_LOOKUP, PLATFORMS
|
|
from .entity import HiveEntity
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
type HiveConfigEntry = ConfigEntry[Hive]
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: HiveConfigEntry) -> bool:
|
|
"""Set up Hive from a config entry."""
|
|
web_session = aiohttp_client.async_get_clientsession(hass)
|
|
hive_config = dict(entry.data)
|
|
hive = Hive(web_session)
|
|
|
|
hive_config["options"] = {}
|
|
hive_config["options"].update(
|
|
{CONF_SCAN_INTERVAL: dict(entry.options).get(CONF_SCAN_INTERVAL, 120)}
|
|
)
|
|
entry.runtime_data = hive
|
|
|
|
try:
|
|
devices = await hive.session.startSession(hive_config)
|
|
except HTTPException as error:
|
|
_LOGGER.error("Could not connect to the internet: %s", error)
|
|
raise ConfigEntryNotReady from error
|
|
except HiveReauthRequired as err:
|
|
raise ConfigEntryAuthFailed from err
|
|
|
|
hub_data = devices["parent"][0]
|
|
connections: set[tuple[str, str]] = set()
|
|
if mac := hub_data.get("macAddress"):
|
|
connections.add((dr.CONNECTION_NETWORK_MAC, dr.format_mac(mac)))
|
|
|
|
device_registry = dr.async_get(hass)
|
|
device_registry.async_get_or_create(
|
|
config_entry_id=entry.entry_id,
|
|
identifiers={(DOMAIN, hub_data["device_id"])},
|
|
connections=connections,
|
|
name=hub_data["hiveName"],
|
|
model=hub_data["deviceData"]["model"],
|
|
sw_version=hub_data["deviceData"]["version"],
|
|
manufacturer=hub_data["deviceData"]["manufacturer"],
|
|
)
|
|
|
|
await hass.config_entries.async_forward_entry_setups(
|
|
entry,
|
|
[
|
|
ha_type
|
|
for ha_type, hive_type in PLATFORM_LOOKUP.items()
|
|
if devices.get(hive_type)
|
|
],
|
|
)
|
|
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: HiveConfigEntry) -> bool:
|
|
"""Unload a config entry."""
|
|
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
|
|
|
|
|
async def async_remove_entry(hass: HomeAssistant, entry: HiveConfigEntry) -> None:
|
|
"""Remove a config entry."""
|
|
hive = Auth(entry.data["username"], entry.data["password"])
|
|
await hive.forget_device(
|
|
entry.data["tokens"]["AuthenticationResult"]["AccessToken"],
|
|
entry.data["device_data"][1],
|
|
)
|
|
|
|
|
|
async def async_remove_config_entry_device(
|
|
hass: HomeAssistant, config_entry: HiveConfigEntry, device_entry: dr.DeviceEntry
|
|
) -> bool:
|
|
"""Remove a config entry from a device."""
|
|
return True
|
|
|
|
|
|
def refresh_system[_HiveEntityT: HiveEntity, **_P](
|
|
func: Callable[Concatenate[_HiveEntityT, _P], Awaitable[Any]],
|
|
) -> Callable[Concatenate[_HiveEntityT, _P], Coroutine[Any, Any, None]]:
|
|
"""Force update all entities after state change."""
|
|
|
|
@wraps(func)
|
|
async def wrapper(self: _HiveEntityT, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
|
await func(self, *args, **kwargs)
|
|
async_dispatcher_send(self.hass, DOMAIN)
|
|
|
|
return wrapper
|