Loqed integration

CODEOWNERS

@@ -590,6 +590,8 @@ build.json @home-assistant/supervisor
 /tests/components/logi_circle/ @evanjd
 /homeassistant/components/lookin/ @ANMalko @bdraco
 /tests/components/lookin/ @ANMalko @bdraco
+/homeassistant/components/loqed/ @cpolhout
+/tests/components/loqed/ @cpolhout
 /homeassistant/components/lovelace/ @home-assistant/frontend
 /tests/components/lovelace/ @home-assistant/frontend
 /homeassistant/components/luci/ @mzdrale

homeassistant/components/loqed/__init__.py

@@ -0,0 +1,133 @@
+"""The loqed integration."""
+from __future__ import annotations
+
+import logging
+
+from aiohttp.web import Request
+import async_timeout
+from loqedAPI import loqed
+
+from homeassistant.components import webhook
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_WEBHOOK_ID, Platform
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import CONF_COORDINATOR, CONF_LOCK, CONF_WEBHOOK_INDEX, DOMAIN
+
+PLATFORMS: list[str] = [Platform.LOCK, Platform.SENSOR]
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@callback
+async def _handle_webhook(
+    hass: HomeAssistant, webhook_id: str, request: Request
+) -> None:
+    """Handle incoming Loqed messages."""
+    _LOGGER.debug("Callback received: %s", str(request.headers))
+    received_ts = request.headers["TIMESTAMP"]
+    received_hash = request.headers["HASH"]
+    body = await request.text()
+
+    _LOGGER.debug("Callback body: %s", body)
+
+    entry = next(
+        entry
+        for entry in hass.data[DOMAIN].values()
+        if entry[CONF_WEBHOOK_ID] == webhook_id
+    )
+    lock: loqed.Lock = entry[CONF_LOCK]
+    coordinator: LoqedDataCoordinator = entry[CONF_COORDINATOR]
+
+    event_data = await lock.receiveWebhook(body, received_hash, received_ts)
+    if "error" in event_data:
+        _LOGGER.warning("Incorrect callback received:: %s", event_data)
+        return
+
+    coordinator.async_set_updated_data(event_data)
+
+
+async def _ensure_webhooks(
+    hass: HomeAssistant, webhook_id: str, lock: loqed.Lock
+) -> int:
+    webhook.async_register(hass, DOMAIN, "Loqed", webhook_id, _handle_webhook)
+    webhook_url = webhook.async_generate_url(hass, webhook_id)
+    _LOGGER.info("Webhook URL: %s", webhook_url)
+
+    webhooks = await lock.getWebhooks()
+
+    webhook_index = next((x["id"] for x in webhooks if x["url"] == webhook_url), None)
+
+    if not webhook_index:
+        await lock.registerWebhook(webhook_url)
+        webhooks = await lock.getWebhooks()
+        webhook_index = next(x["id"] for x in webhooks if x["url"] == webhook_url)
+
+        _LOGGER.info("Webhook got index %s", webhook_index)
+
+    return int(webhook_index)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up loqed from a config entry."""
+    websession = async_get_clientsession(hass)
+    host = entry.data["host"]
+    apiclient = loqed.APIClient(websession, f"http://{host}")
+    api = loqed.LoqedAPI(apiclient)
+
+    lock = await api.async_get_lock(
+        entry.data["api_key"],
+        entry.data["bkey"],
+        entry.data["key_id"],
+        entry.data["host"],
+    )
+    webhook_id = entry.data[CONF_WEBHOOK_ID]
+    webhook_index = await _ensure_webhooks(hass, webhook_id, lock)
+    coordinator = LoqedDataCoordinator(hass, api)
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
+        CONF_WEBHOOK_ID: webhook_id,
+        CONF_LOCK: lock,
+        CONF_COORDINATOR: coordinator,
+        CONF_WEBHOOK_INDEX: webhook_index,
+    }
+
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    data = hass.data[DOMAIN][entry.entry_id]
+    webhook.async_unregister(hass, data[CONF_WEBHOOK_ID])
+    lock: loqed.Lock = data[CONF_LOCK]
+
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    try:
+        await lock.deleteWebhook(data[CONF_WEBHOOK_INDEX])
+    except Exception:  # pylint: disable=broad-except
+        _LOGGER.exception("Failed to delete webhook")
+        return False
+
+    return unload_ok
+
+
+class LoqedDataCoordinator(DataUpdateCoordinator):
+    """Data update coordinator for the loqed platform."""
+
+    def __init__(self, hass: HomeAssistant, api: loqed.LoqedAPI) -> None:
+        """Initialize the Loqed Data Update coordinator."""
+        super().__init__(hass, _LOGGER, name="Loqed sensors")
+        self._api = api
+
+    async def _async_update_data(self) -> dict[str, str]:
+        """Fetch data from API endpoint."""
+        async with async_timeout.timeout(10):
+            return await self._api.async_get_lock_details()

homeassistant/components/loqed/config_flow.py

@@ -0,0 +1,134 @@
+"""Config flow for loqed integration."""
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+import aiohttp
+from loqedAPI import loqed
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components import webhook
+from homeassistant.const import CONF_HOST, CONF_WEBHOOK_ID
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def validate_input(
+    hass: HomeAssistant, data: dict[str, Any], zeroconf_host: str | None
+) -> dict[str, Any]:
+    """Validate the user input allows us to connect."""
+
+    newdata = data
+    json_config = json.loads(data["config"])
+    newdata["ip"] = json_config["bridge_ip"]
+    newdata["host"] = json_config["bridge_mdns_hostname"]
+    newdata["bkey"] = json_config["bridge_key"]
+    newdata["key_id"] = int(json_config["lock_key_local_id"])
+    newdata["api_key"] = json_config["lock_key_key"]
+
+    if zeroconf_host is not None and zeroconf_host != newdata["host"]:
+        raise InvalidAuth(
+            f"Got config for {newdata['host']} while configuring {zeroconf_host} "
+        )
+
+    # 1. Checking loqed-connection
+    try:
+        session = async_get_clientsession(hass)
+
+        apiclient = loqed.APIClient(session, "http://" + newdata["ip"])
+        api = loqed.LoqedAPI(apiclient)
+        lock = await api.async_get_lock(
+            newdata["api_key"], newdata["bkey"], newdata["key_id"], newdata["host"]
+        )
+        newdata["id"] = lock.id
+        # checking getWebooks to check the bridgeKey
+        await lock.getWebhooks()
+    except (aiohttp.ClientError):
+        _LOGGER.error("HTTP Connection error to loqed lock")
+        raise CannotConnect from aiohttp.ClientError
+    return newdata
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for loqed."""
+
+    VERSION = 2
+
+    def __init__(self) -> None:
+        """Initialize the ConfigFlow for the LOQED integration."""
+        self._host: str | None = None
+
+    async def async_step_zeroconf(self, discovery_info) -> FlowResult:
+        """Handle zeroconf discovery."""
+        self._host = discovery_info.hostname.rstrip(".")
+
+        session = async_get_clientsession(self.hass)
+        apiclient = loqed.APIClient(session, f"http://{self._host}")
+        api = loqed.LoqedAPI(apiclient)
+        lock_data = await api.async_get_lock_details()
+
+        # Check if already exists
+        await self.async_set_unique_id(lock_data["bridge_mac_wifi"])
+        self._abort_if_unique_id_configured()
+        return await self.async_step_user()
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Show userform to user."""
+        user_data_schema = vol.Schema(
+            {
+                vol.Required("config"): str,
+            }
+        )
+        self.context["title_placeholders"] = {CONF_HOST: self._host}
+        if user_input is None:
+            return self.async_show_form(
+                step_id="user",
+                data_schema=user_data_schema,
+                description_placeholders={
+                    CONF_HOST: self._host,
+                    "config_url": "https://app.loqed.com/API-Config",
+                },
+            )
+
+        errors = {}
+
+        try:
+            info = await validate_input(self.hass, user_input, self._host)
+        except CannotConnect:
+            errors["base"] = "cannot_connect"
+        except InvalidAuth:
+            errors["base"] = "invalid_auth"
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Unexpected exception")
+            errors["base"] = "unknown"
+        else:
+            await self.async_set_unique_id(info["id"])
+            self._abort_if_unique_id_configured()
+
+            return self.async_create_entry(
+                title="LOQED Touch Smart Lock",
+                data=(user_input | {CONF_WEBHOOK_ID: webhook.async_generate_id()}),
+            )
+
+        return self.async_show_form(
+            step_id="user", data_schema=user_data_schema, errors=errors
+        )
+
+
+class CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""

homeassistant/components/loqed/const.py

@@ -0,0 +1,7 @@
+"""Constants for the loqed integration."""
+
+
+DOMAIN = "loqed"
+CONF_WEBHOOK_INDEX = "webhook_index"
+CONF_COORDINATOR = "coordinator"
+CONF_LOCK = "lock"