Add new integration Loqed

CODEOWNERS

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

homeassistant/components/loqed/__init__.py

@@ -0,0 +1,55 @@
+"""The loqed integration."""
+from __future__ import annotations
+
+import logging
+import re
+
+from loqedAPI import loqed
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+from .coordinator import LoqedDataCoordinator
+
+PLATFORMS: list[str] = [Platform.LOCK]
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+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["bridge_ip"]
+    apiclient = loqed.APIClient(websession, f"http://{host}")
+    api = loqed.LoqedAPI(apiclient)
+
+    lock = await api.async_get_lock(
+        entry.data["lock_key_key"],
+        entry.data["bridge_key"],
+        int(entry.data["lock_key_local_id"]),
+        re.sub(r"LOQED-([a-f0-9]+)\.local", r"\1", entry.data["bridge_mdns_hostname"]),
+    )
+    coordinator = LoqedDataCoordinator(hass, api, lock, entry)
+    await coordinator.ensure_webhooks()
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+
+    await coordinator.async_config_entry_first_refresh()
+
+    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."""
+    coordinator: LoqedDataCoordinator = hass.data[DOMAIN][entry.entry_id]
+
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    await coordinator.remove_webhooks()
+
+    return unload_ok

homeassistant/components/loqed/config_flow.py

@@ -0,0 +1,166 @@
+"""Config flow for loqed integration."""
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any
+
+import aiohttp
+from loqedAPI import cloud_loqed, loqed
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components import webhook
+from homeassistant.components.zeroconf import ZeroconfServiceInfo
+from homeassistant.const import CONF_API_TOKEN, CONF_HOST, CONF_NAME, 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__)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Loqed."""
+
+    VERSION = 1
+    DOMAIN = DOMAIN
+    _host: str | None = None
+
+    async def validate_input(
+        self, hass: HomeAssistant, data: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Validate the user input allows us to connect."""
+
+        # 1. Checking loqed-connection
+        try:
+            session = async_get_clientsession(hass)
+            cloud_api_client = cloud_loqed.CloudAPIClient(
+                session,
+                data[CONF_API_TOKEN],
+            )
+            cloud_client = cloud_loqed.LoqedCloudAPI(cloud_api_client)
+            lock_data = await cloud_client.async_get_locks()
+        except aiohttp.ClientError:
+            _LOGGER.error("HTTP Connection error to loqed API")
+            raise CannotConnect from aiohttp.ClientError
+
+        try:
+            selected_lock = next(
+                lock
+                for lock in lock_data["data"]
+                if lock["bridge_ip"] == self._host or lock["name"] == data.get("name")
+            )
+
+            apiclient = loqed.APIClient(session, f"http://{selected_lock['bridge_ip']}")
+            api = loqed.LoqedAPI(apiclient)
+            lock = await api.async_get_lock(
+                selected_lock["backend_key"],
+                selected_lock["bridge_key"],
+                selected_lock["local_id"],
+                selected_lock["bridge_ip"],
+            )
+
+            # checking getWebooks to check the bridgeKey
+            await lock.getWebhooks()
+            return {
+                "lock_key_key": selected_lock["key_secret"],
+                "bridge_key": selected_lock["bridge_key"],
+                "lock_key_local_id": selected_lock["local_id"],
+                "bridge_mdns_hostname": selected_lock["bridge_hostname"],
+                "bridge_ip": selected_lock["bridge_ip"],
+                "name": selected_lock["name"],
+                "id": selected_lock["id"],
+            }
+        except StopIteration:
+            raise InvalidAuth from StopIteration
+        except aiohttp.ClientError:
+            _LOGGER.error("HTTP Connection error to loqed lock")
+            raise CannotConnect from aiohttp.ClientError
+
+    async def async_step_zeroconf(
+        self, discovery_info: ZeroconfServiceInfo
+    ) -> FlowResult:
+        """Handle zeroconf discovery."""
+        host = discovery_info.host
+        self._host = host
+
+        session = async_get_clientsession(self.hass)
+        apiclient = loqed.APIClient(session, f"http://{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({CONF_HOST: host})
+
+        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(CONF_API_TOKEN): str,
+                }
+            )
+            if self._host
+            else vol.Schema(
+                {
+                    vol.Required(CONF_NAME): str,
+                    vol.Required(CONF_API_TOKEN): str,
+                }
+            )
+        )
+
+        if user_input is None:
+            return self.async_show_form(
+                step_id="user",
+                data_schema=user_data_schema,
+                description_placeholders={
+                    "config_url": "https://integrations.production.loqed.com/personal-access-tokens",
+                },
+            )
+
+        errors = {}
+
+        try:
+            info = await self.validate_input(self.hass, user_input)
+        except CannotConnect:
+            errors["base"] = "cannot_connect"
+        except InvalidAuth:
+            errors["base"] = "invalid_auth"
+        else:
+            await self.async_set_unique_id(
+                re.sub(r"LOQED-([a-f0-9]+)\.local", r"\1", info["bridge_mdns_hostname"])
+            )
+            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()} | info
+                ),
+            )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=user_data_schema,
+            errors=errors,
+            description_placeholders={
+                "config_url": "https://integrations.production.loqed.com/personal-access-tokens",
+            },
+        )
+
+
+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,6 @@
+"""Constants for the loqed integration."""
+
+
+DOMAIN = "loqed"
+OAUTH2_AUTHORIZE = "https://app.loqed.com/API/integration_oauth3/login.php"
+OAUTH2_TOKEN = "https://app.loqed.com/API/integration_oauth3/token.php"

homeassistant/components/loqed/coordinator.py

@@ -0,0 +1,152 @@
+"""Provides the coordinator for a LOQED lock."""
+import logging
+from typing import TypedDict
+
+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
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class BatteryMessage(TypedDict):
+    """Properties in a battery update message."""
+
+    mac_wifi: str
+    mac_ble: str
+    battery_type: str
+    battery_percentage: int
+
+
+class StateReachedMessage(TypedDict):
+    """Properties in a battery update message."""
+
+    requested_state: str
+    requested_state_numeric: int
+    event_type: str
+    key_local_id: int
+    mac_wifi: str
+    mac_ble: str
+
+
+class TransitionMessage(TypedDict):
+    """Properties in a battery update message."""
+
+    go_to_state: str
+    go_to_state_numeric: int
+    event_type: str
+    key_local_id: int
+    mac_wifi: str
+    mac_ble: str
+
+
+class StatusMessage(TypedDict):
+    """Properties returned by the status endpoint of the bridhge."""
+
+    battery_percentage: int
+    battery_type: str
+    battery_type_numeric: int
+    battery_voltage: float
+    bolt_state: str
+    bolt_state_numeric: int
+    bridge_mac_wifi: str
+    bridge_mac_ble: str
+    lock_online: int
+    webhooks_number: int
+    ip_address: str
+    up_timestamp: int
+    wifi_strength: int
+    ble_strength: int
+
+
+class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]):
+    """Data update coordinator for the loqed platform."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        api: loqed.LoqedAPI,
+        lock: loqed.Lock,
+        entry: ConfigEntry,
+    ) -> None:
+        """Initialize the Loqed Data Update coordinator."""
+        super().__init__(hass, _LOGGER, name="Loqed sensors")
+        self._hass = hass
+        self._api = api
+        self._entry = entry
+        self.lock = lock
+
+    async def _async_update_data(self) -> StatusMessage:
+        """Fetch data from API endpoint."""
+        async with async_timeout.timeout(10):
+            return await self._api.async_get_lock_details()
+
+    @callback
+    async def _handle_webhook(
+        self, hass: HomeAssistant, webhook_id: str, request: Request
+    ) -> None:
+        """Handle incoming Loqed messages."""
+        _LOGGER.debug("Callback received: %s", request.headers)
+        received_ts = request.headers["TIMESTAMP"]
+        received_hash = request.headers["HASH"]
+        body = await request.text()
+
+        _LOGGER.debug("Callback body: %s", body)
+
+        event_data = await self.lock.receiveWebhook(body, received_hash, received_ts)
+        if "error" in event_data:
+            _LOGGER.warning("Incorrect callback received:: %s", event_data)
+            return
+
+        self.async_update_listeners()
+
+    async def ensure_webhooks(self) -> None:
+        """Register webhook on LOQED bridge."""
+        webhook_id = self._entry.data[CONF_WEBHOOK_ID]
+
+        webhook.async_register(
+            self.hass, DOMAIN, "Loqed", webhook_id, self._handle_webhook
+        )
+        webhook_url = webhook.async_generate_url(self.hass, webhook_id)
+        _LOGGER.info("Webhook URL: %s", webhook_url)
+
+        webhooks = await self.lock.getWebhooks()
+
+        webhook_index = next(
+            (x["id"] for x in webhooks if x["url"] == webhook_url), None
+        )
+
+        if not webhook_index:
+            await self.lock.registerWebhook(webhook_url)
+            webhooks = await self.lock.getWebhooks()
+            webhook_index = next(x["id"] for x in webhooks if x["url"] == webhook_url)
+
+            _LOGGER.info("Webhook got index %s", webhook_index)
+
+    async def remove_webhooks(self) -> None:
+        """Remove webhook from LOQED bridge."""
+        webhook_id = self._entry.data[CONF_WEBHOOK_ID]
+        webhook_url = webhook.async_generate_url(self.hass, webhook_id)
+
+        webhook.async_unregister(
+            self.hass,
+            webhook_id,
+        )
+        _LOGGER.info("Webhook URL: %s", webhook_url)
+
+        webhooks = await self.lock.getWebhooks()
+
+        webhook_index = next(
+            (x["id"] for x in webhooks if x["url"] == webhook_url), None
+        )
+
+        if webhook_index:
+            await self.lock.deleteWebhook(webhook_index)