__init__.py

Go to the documentation of this file.

"""Support for Minut Point."""
 
import asyncio
from dataclasses import dataclass
from http import HTTPStatus
import logging
 
from aiohttp import ClientError, ClientResponseError, web
from pypoint import PointSession
import voluptuous as vol
 
from homeassistant.components import webhook
from homeassistant.components.application_credentials import (
    ClientCredential,
    async_import_client_credential,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
    CONF_CLIENT_ID,
    CONF_CLIENT_SECRET,
    CONF_WEBHOOK_ID,
    Platform,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import (
    aiohttp_client,
    config_entry_oauth2_flow,
    config_validation as cv,
)
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType
 
from . import api
from .const import (
    CONF_WEBHOOK_URL,
    DOMAIN,
    EVENT_RECEIVED,
    POINT_DISCOVERY_NEW,
    SCAN_INTERVAL,
    SIGNAL_UPDATE_ENTITY,
    SIGNAL_WEBHOOK,
)
 
_LOGGER = logging.getLogger(__name__)
 
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
 
type PointConfigEntry = ConfigEntry[PointData]
 
CONFIG_SCHEMA = vol.Schema(
    {
        DOMAIN: vol.Schema(
            {
                vol.Required(CONF_CLIENT_ID): cv.string,
                vol.Required(CONF_CLIENT_SECRET): cv.string,
            }
        )
    },
    extra=vol.ALLOW_EXTRA,
)
 
 
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
    """Set up the Minut Point component."""
    if DOMAIN not in config:
        return True
 
    conf = config[DOMAIN]
 
    async_create_issue(
        hass,
        HOMEASSISTANT_DOMAIN,
        f"deprecated_yaml_{DOMAIN}",
        breaks_in_ha_version="2025.4.0",
        is_fixable=False,
        issue_domain=DOMAIN,
        severity=IssueSeverity.WARNING,
        translation_key="deprecated_yaml",
        translation_placeholders={
            "domain": DOMAIN,
            "integration_title": "Point",
        },
    )
 
    if not hass.config_entries.async_entries(DOMAIN):
        await async_import_client_credential(
            hass,
            DOMAIN,
            ClientCredential(
                conf[CONF_CLIENT_ID],
                conf[CONF_CLIENT_SECRET],
            ),
        )
 
        hass.async_create_task(
            hass.config_entries.flow.async_init(
                DOMAIN, context={"source": SOURCE_IMPORT}, data=conf
            )
        )
 
    return True
 
 
async def async_setup_entry(hass: HomeAssistant, entry: PointConfigEntry) -> bool:
    """Set up Minut Point from a config entry."""
 
    if "auth_implementation" not in entry.data:
        raise ConfigEntryAuthFailed("Authentication failed. Please re-authenticate.")
 
    implementation = (
        await config_entry_oauth2_flow.async_get_config_entry_implementation(
            hass, entry
        )
    )
    session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
    auth = api.AsyncConfigEntryAuth(
        aiohttp_client.async_get_clientsession(hass), session
    )
 
    try:
        await auth.async_get_access_token()
    except ClientResponseError as err:
        if err.status in {HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN}:
            raise ConfigEntryAuthFailed from err
        raise ConfigEntryNotReady from err
    except ClientError as err:
        raise ConfigEntryNotReady from err
 
    point_session = PointSession(auth)
 
    client = MinutPointClient(hass, entry, point_session)
    hass.async_create_task(client.update())
    entry.runtime_data = PointData(client)
 
    await async_setup_webhook(hass, entry, point_session)
    await hass.config_entries.async_forward_entry_setups(
        entry, [*PLATFORMS, Platform.ALARM_CONTROL_PANEL]
    )
 
    return True
 
 
async def async_setup_webhook(
    hass: HomeAssistant, entry: PointConfigEntry, session: PointSession
) -> None:
    """Set up a webhook to handle binary sensor events."""
    if CONF_WEBHOOK_ID not in entry.data:
        webhook_id = webhook.async_generate_id()
        webhook_url = webhook.async_generate_url(hass, webhook_id)
        _LOGGER.debug("Registering new webhook at: %s", webhook_url)
 
        hass.config_entries.async_update_entry(
            entry,
            data={
                **entry.data,
                CONF_WEBHOOK_ID: webhook_id,
                CONF_WEBHOOK_URL: webhook_url,
            },
        )
 
    await session.update_webhook(
        webhook.async_generate_url(hass, entry.data[CONF_WEBHOOK_ID]),
        entry.data[CONF_WEBHOOK_ID],
        ["*"],
    )
    webhook.async_register(
        hass, DOMAIN, "Point", entry.data[CONF_WEBHOOK_ID], handle_webhook
    )
 
 
async def async_unload_entry(hass: HomeAssistant, entry: PointConfigEntry) -> bool:
    """Unload a config entry."""
    if unload_ok := await hass.config_entries.async_unload_platforms(
        entry, [*PLATFORMS, Platform.ALARM_CONTROL_PANEL]
    ):
        session: PointSession = entry.runtime_data.client
        if CONF_WEBHOOK_ID in entry.data:
            webhook.async_unregister(hass, entry.data[CONF_WEBHOOK_ID])
            await session.remove_webhook()
    return unload_ok
 
 
async def handle_webhook(
    hass: HomeAssistant, webhook_id: str, request: web.Request
) -> None:
    """Handle webhook callback."""
    try:
        data = await request.json()
        _LOGGER.debug("Webhook %s: %s", webhook_id, data)
    except ValueError:
        return
 
    if isinstance(data, dict):
        data["webhook_id"] = webhook_id
        async_dispatcher_send(hass, SIGNAL_WEBHOOK, data, data.get("hook_id"))
    hass.bus.async_fire(EVENT_RECEIVED, data)
 
 
class MinutPointClient:
    """Get the latest data and update the states."""
 
    def __init__(
        self, hass: HomeAssistant, config_entry: ConfigEntry, session: PointSession
    ) -> None:
        """Initialize the Minut data object."""
        self._known_devices: set[str] = set()
        self._known_homes: set[str] = set()
        self._hass_hass = hass
        self._config_entry_config_entry = config_entry
        self._is_available_is_available = True
        self._client_client = session
 
        async_track_time_interval(self._hass_hass, self.updateupdate, SCAN_INTERVAL)
 
    async def update(self, *args):
        """Periodically poll the cloud for current state."""
        await self._sync_sync()
 
    async def _sync(self):
        """Update local list of devices."""
        if not await self._client_client.update():
            self._is_available_is_available = False
            _LOGGER.warning("Device is unavailable")
            async_dispatcher_send(self._hass_hass, SIGNAL_UPDATE_ENTITY)
            return
 
        self._is_available_is_available = True
        for home_id in self._client_client.homes:
            if home_id not in self._known_homes:
                async_dispatcher_send(
                    self._hass_hass,
                    POINT_DISCOVERY_NEW.format(Platform.ALARM_CONTROL_PANEL),
                    home_id,
                )
                self._known_homes.add(home_id)
        for device in self._client_client.devices:
            if device.device_id not in self._known_devices:
                for platform in PLATFORMS:
                    async_dispatcher_send(
                        self._hass_hass,
                        POINT_DISCOVERY_NEW.format(platform),
                        device.device_id,
                    )
                self._known_devices.add(device.device_id)
        async_dispatcher_send(self._hass_hass, SIGNAL_UPDATE_ENTITY)
 
    def device(self, device_id):
        """Return device representation."""
        return self._client_client.device(device_id)
 
    def is_available(self, device_id):
        """Return device availability."""
        if not self._is_available_is_available:
            return False
        return device_id in self._client_client.device_ids
 
    async def remove_webhook(self):
        """Remove the session webhook."""
        return await self._client_client.remove_webhook()
 
    @property
    def homes(self):
        """Return known homes."""
        return self._client_client.homes
 
    async def async_alarm_disarm(self, home_id):
        """Send alarm disarm command."""
        return await self._client_client.alarm_disarm(home_id)
 
    async def async_alarm_arm(self, home_id):
        """Send alarm arm command."""
        return await self._client_client.alarm_arm(home_id)
 
 
@dataclass
class PointData:
    """Point Data."""
 
    client: MinutPointClient
    entry_lock: asyncio.Lock = asyncio.Lock()