mirror of https://github.com/home-assistant/core
152 lines
5.1 KiB
Python
152 lines
5.1 KiB
Python
"""Init the tedee component."""
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
from http import HTTPStatus
|
|
import logging
|
|
from typing import Any
|
|
|
|
from aiohttp.hdrs import METH_POST
|
|
from aiohttp.web import Request, Response
|
|
from aiotedee.exception import TedeeDataUpdateException, TedeeWebhookException
|
|
|
|
from homeassistant.components.http import HomeAssistantView
|
|
from homeassistant.components.webhook import (
|
|
async_generate_id as webhook_generate_id,
|
|
async_generate_url as webhook_generate_url,
|
|
async_register as webhook_register,
|
|
async_unregister as webhook_unregister,
|
|
)
|
|
from homeassistant.const import CONF_WEBHOOK_ID, EVENT_HOMEASSISTANT_STOP, Platform
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import device_registry as dr
|
|
from homeassistant.helpers.network import get_url
|
|
|
|
from .const import DOMAIN, NAME
|
|
from .coordinator import TedeeApiCoordinator, TedeeConfigEntry
|
|
|
|
PLATFORMS = [
|
|
Platform.BINARY_SENSOR,
|
|
Platform.LOCK,
|
|
Platform.SENSOR,
|
|
]
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: TedeeConfigEntry) -> bool:
|
|
"""Integration setup."""
|
|
|
|
coordinator = TedeeApiCoordinator(hass, entry)
|
|
|
|
await coordinator.async_config_entry_first_refresh()
|
|
|
|
device_registry = dr.async_get(hass)
|
|
device_registry.async_get_or_create(
|
|
config_entry_id=entry.entry_id,
|
|
identifiers={(DOMAIN, coordinator.bridge.serial)},
|
|
manufacturer="Tedee",
|
|
name=coordinator.bridge.name,
|
|
model="Bridge",
|
|
serial_number=coordinator.bridge.serial,
|
|
)
|
|
|
|
entry.runtime_data = coordinator
|
|
|
|
async def unregister_webhook(_: Any) -> None:
|
|
await coordinator.async_unregister_webhook()
|
|
webhook_unregister(hass, entry.data[CONF_WEBHOOK_ID])
|
|
|
|
async def register_webhook() -> None:
|
|
instance_url = get_url(hass, allow_ip=True, allow_external=False)
|
|
# first make sure we don't have leftover callbacks to the same instance
|
|
try:
|
|
await coordinator.tedee_client.cleanup_webhooks_by_host(instance_url)
|
|
except (TedeeDataUpdateException, TedeeWebhookException) as ex:
|
|
_LOGGER.warning("Failed to cleanup Tedee webhooks by host: %s", ex)
|
|
|
|
webhook_url = webhook_generate_url(
|
|
hass, entry.data[CONF_WEBHOOK_ID], allow_external=False, allow_ip=True
|
|
)
|
|
webhook_name = "Tedee"
|
|
if entry.title != NAME:
|
|
webhook_name = f"{NAME} {entry.title}"
|
|
|
|
webhook_register(
|
|
hass,
|
|
DOMAIN,
|
|
webhook_name,
|
|
entry.data[CONF_WEBHOOK_ID],
|
|
get_webhook_handler(coordinator),
|
|
allowed_methods=[METH_POST],
|
|
)
|
|
_LOGGER.debug("Registered Tedee webhook at hass: %s", webhook_url)
|
|
|
|
try:
|
|
await coordinator.async_register_webhook(webhook_url)
|
|
except TedeeWebhookException:
|
|
_LOGGER.exception("Failed to register Tedee webhook from bridge")
|
|
else:
|
|
entry.async_on_unload(
|
|
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, unregister_webhook)
|
|
)
|
|
|
|
entry.async_create_background_task(
|
|
hass, register_webhook(), "tedee_register_webhook"
|
|
)
|
|
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
|
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: TedeeConfigEntry) -> bool:
|
|
"""Unload a config entry."""
|
|
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
|
|
|
|
|
def get_webhook_handler(
|
|
coordinator: TedeeApiCoordinator,
|
|
) -> Callable[[HomeAssistant, str, Request], Awaitable[Response | None]]:
|
|
"""Return webhook handler."""
|
|
|
|
async def async_webhook_handler(
|
|
hass: HomeAssistant, webhook_id: str, request: Request
|
|
) -> Response | None:
|
|
# Handle http post calls to the path.
|
|
if not request.body_exists:
|
|
return HomeAssistantView.json(
|
|
result="No Body", status_code=HTTPStatus.BAD_REQUEST
|
|
)
|
|
|
|
body = await request.json()
|
|
try:
|
|
coordinator.webhook_received(body)
|
|
except TedeeWebhookException as ex:
|
|
return HomeAssistantView.json(
|
|
result=str(ex), status_code=HTTPStatus.BAD_REQUEST
|
|
)
|
|
|
|
return HomeAssistantView.json(result="OK", status_code=HTTPStatus.OK)
|
|
|
|
return async_webhook_handler
|
|
|
|
|
|
async def async_migrate_entry(
|
|
hass: HomeAssistant, config_entry: TedeeConfigEntry
|
|
) -> bool:
|
|
"""Migrate old entry."""
|
|
if config_entry.version > 1:
|
|
# This means the user has downgraded from a future version
|
|
return False
|
|
|
|
version = config_entry.version
|
|
minor_version = config_entry.minor_version
|
|
|
|
if version == 1 and minor_version == 1:
|
|
_LOGGER.debug(
|
|
"Migrating Tedee config entry from version %s.%s", version, minor_version
|
|
)
|
|
data = {**config_entry.data, CONF_WEBHOOK_ID: webhook_generate_id()}
|
|
hass.config_entries.async_update_entry(config_entry, data=data, minor_version=2)
|
|
_LOGGER.debug("Migration to version 1.2 successful")
|
|
return True
|