Add Level Lock component (Level Connect)

homeassistant/brands/level.json

@@ -1,5 +1,6 @@
 {
   "domain": "level",
-  "name": "Level",
+  "name": "Level Lock",
+  "integrations": ["levelhome"],
   "iot_standards": ["matter"]
 }

homeassistant/components/levelhome/__init__.py

@@ -0,0 +1,106 @@
+"""The Level Lock integration."""
+
+from __future__ import annotations
+
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
+
+from . import auth as auth_mod
+from ._lib.level_ha import WebsocketManager as LevelWebsocketManager
+from .const import (
+    CONF_OAUTH2_BASE_URL,
+    CONF_PARTNER_BASE_URL,
+    DEFAULT_PARTNER_BASE_URL,
+    DOMAIN,
+)
+from .coordinator import LevelLocksCoordinator
+
+# For your initial PR, limit it to 1 platform.
+_PLATFORMS: list[Platform] = [Platform.LOCK]
+
+type LevelHomeConfigEntry = ConfigEntry[LevelLocksCoordinator]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: LevelHomeConfigEntry) -> bool:
+    """Set up Level Lock from a config entry."""
+    _LOGGER = logging.getLogger(__name__)
+
+    _LOGGER.info("Setting up Level Lock config entry")
+    if entry.options:
+        hass.data.setdefault(DOMAIN, {})[CONF_OAUTH2_BASE_URL] = entry.options.get(
+            CONF_OAUTH2_BASE_URL
+        )
+        hass.data[DOMAIN][CONF_PARTNER_BASE_URL] = entry.options.get(
+            CONF_PARTNER_BASE_URL
+        )
+    implementation = (
+        await config_entry_oauth2_flow.async_get_config_entry_implementation(
+            hass, entry
+        )
+    )
+
+    oauth_session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
+    client_session = aiohttp_client.async_get_clientsession(hass)
+    config_auth = auth_mod.AsyncConfigEntryAuth(client_session, oauth_session)
+
+    _LOGGER.info("Ensuring token is valid before starting WebSocket")
+    await oauth_session.async_ensure_token_valid()
+    _LOGGER.info("Token validated, proceeding with setup")
+
+    base_url = (hass.data.get(DOMAIN) or {}).get(
+        CONF_PARTNER_BASE_URL
+    ) or DEFAULT_PARTNER_BASE_URL
+    _LOGGER.info("Using base URL: %s", base_url)
+
+    async def _get_token() -> str:
+        return await config_auth.async_get_access_token()
+
+    async def _on_state(
+        lock_id: str, is_locked: bool | None, payload: dict | None
+    ) -> None:
+        await coordinator.async_handle_push_update(lock_id, is_locked, payload)
+
+    async def _on_devices(devices: list[dict]) -> None:
+        await coordinator.async_handle_devices_update(devices)
+
+    ws_manager = LevelWebsocketManager(
+        client_session, base_url, _get_token, _on_state, _on_devices
+    )
+    _LOGGER.info("Starting WebSocket manager")
+    await ws_manager.async_start()
+    _LOGGER.info("WebSocket manager started")
+
+    coordinator = LevelLocksCoordinator(hass, ws_manager, config_entry=entry)
+    _LOGGER.info("Starting coordinator first refresh")
+    await coordinator.async_config_entry_first_refresh()
+    _LOGGER.info(
+        "Coordinator first refresh completed with %d devices: %s",
+        len(coordinator.data) if coordinator.data else 0,
+        list(coordinator.data.keys()) if coordinator.data else [],
+    )
+
+    entry.runtime_data = coordinator
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
+        "ws_manager": ws_manager,
+    }
+
+    await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
+    _LOGGER.info("Level Lock setup completed")
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: LevelHomeConfigEntry) -> bool:
+    """Unload a config entry."""
+    unloaded = await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)
+    if DOMAIN in hass.data and entry.entry_id in hass.data[DOMAIN]:
+        data = hass.data[DOMAIN].pop(entry.entry_id, {})
+        ws_manager: LevelWebsocketManager | None = data.get("ws_manager")
+        if ws_manager is not None:
+            await ws_manager.async_stop()
+    return unloaded

homeassistant/components/levelhome/_lib/__init__.py

@@ -0,0 +1 @@
+"""Level Lock API library package."""

homeassistant/components/levelhome/_lib/client.py

@@ -0,0 +1,93 @@
+"""Library HTTP client for Level Lock (no Home Assistant dependencies)."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from http import HTTPStatus
+from typing import Any
+
+from aiohttp import ClientError, ClientSession
+
+from .protocol import coerce_is_locked
+
+# Async token provider callable type
+TokenProvider = Callable[[], Awaitable[str]]
+
+
+class ApiError(Exception):
+    """Raised when a Level HTTP API call fails."""
+
+
+class Client:
+    """Minimal async HTTP client for Level endpoints (library-style).
+
+    This class is intentionally HA-agnostic. It relies on an injected
+    aiohttp ClientSession and an async token provider callable.
+    """
+
+    def __init__(
+        self, session: ClientSession, base_url: str, get_token: TokenProvider
+    ) -> None:
+        """Initialize the Level API client."""
+        self._session = session
+        self._base_url = base_url.rstrip("/")
+        self._get_token = get_token
+
+    async def _request(
+        self, method: str, path: str, *, json: dict[str, Any] | None = None
+    ) -> Any:
+        token = await self._get_token()
+        url = f"{self._base_url}{path}"
+        headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
+        try:
+            async with self._session.request(
+                method, url, headers=headers, json=json
+            ) as resp:
+                if resp.status >= HTTPStatus.BAD_REQUEST:
+                    text = await resp.text()
+                    raise ApiError(f"HTTP {resp.status} for {method} {path}: {text}")
+                if resp.content_type == "application/json":
+                    return await resp.json()
+                return await resp.text()
+        except ClientError as err:  # aiohttp error
+            raise ApiError(f"API error: {err}") from err
+
+    async def async_list_locks(self) -> list[dict[str, Any]]:
+        """Return a list of locks as raw dictionaries from the API."""
+        data = await self._request("GET", "/v1/locks")
+        locks = list(data.get("locks", []))
+        for lock in locks:
+            if "uuid" not in lock and "id" in lock:
+                lock["uuid"] = lock["id"]
+        return locks
+
+    async def async_get_lock_status(self, lock_id: str) -> dict[str, Any]:
+        """Return the raw status payload for a lock."""
+        return await self._request("GET", f"/v1/locks/{lock_id}")
+
+    async def async_lock(self, lock_id: str) -> None:
+        """Send lock command to the specified lock."""
+        await self._request("POST", f"/v1/locks/{lock_id}/lock")
+
+    async def async_unlock(self, lock_id: str) -> None:
+        """Send unlock command to the specified lock."""
+        await self._request("POST", f"/v1/locks/{lock_id}/unlock")
+
+    async def async_list_locks_normalized(self) -> list[dict[str, Any]]:
+        """Return locks with derived boolean is_locked alongside raw state."""
+        locks = await self.async_list_locks()
+        normalized: list[dict[str, Any]] = []
+        for item in locks:
+            state = item.get("state")
+            normalized.append(
+                {
+                    **item,
+                    "is_locked": coerce_is_locked(state),
+                }
+            )
+        return normalized
+
+    async def async_get_lock_status_bool(self, lock_id: str) -> bool | None:
+        """Return boolean locked status derived from the raw status payload."""
+        data = await self.async_get_lock_status(lock_id)
+        return coerce_is_locked(data.get("state"))

homeassistant/components/levelhome/_lib/level_ha.py

@@ -0,0 +1,18 @@
+"""Local import seam that re-exports library-style API.
+
+This allows later switching to an external package by changing imports here.
+"""
+
+from __future__ import annotations
+
+from aiohttp import ClientSession  # re-exported for type checkers in consumers
+
+from .client import ApiError, Client  # re-export from library client
+from .ws import LevelWebsocketManager as WebsocketManager  # re-export
+
+__all__ = [
+    "ApiError",
+    "Client",
+    "ClientSession",
+    "WebsocketManager",
+]

homeassistant/components/levelhome/_lib/protocol.py

@@ -0,0 +1,29 @@
+"""Library-shared protocol helpers for Level Lock.
+
+Pure helper module to avoid circular imports.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def coerce_is_locked(state: Any) -> bool | None:
+    """Convert vendor state to boolean locked status or None for unknown.
+
+    Transitional states (e.g., "locking"/"unlocking") return None.
+    """
+
+    if state is None:
+        return None
+    if isinstance(state, str):
+        lowered = state.lower()
+        if lowered in ("locked", "lock", "secure"):
+            return True
+        if lowered in ("unlocked", "unlock", "insecure"):
+            return False
+        if lowered in ("locking", "unlocking"):
+            return None
+    if isinstance(state, bool):
+        return state
+    return None