Add minimal UniFi Access integration

.strict-typing

@@ -507,6 +507,7 @@ homeassistant.components.trend.*
 homeassistant.components.tts.*
 homeassistant.components.twentemilieu.*
 homeassistant.components.unifi.*
+homeassistant.components.unifi_access.*
 homeassistant.components.unifiprotect.*
 homeassistant.components.upcloud.*
 homeassistant.components.update.*

homeassistant/brands/ubiquiti.json

@@ -1,5 +1,11 @@
 {
   "domain": "ubiquiti",
   "name": "Ubiquiti",
-  "integrations": ["unifi", "unifi_direct", "unifiled", "unifiprotect"]
+  "integrations": [
+    "unifi",
+    "unifi_direct",
+    "unifiled",
+    "unifiprotect",
+    "unifi_access"
+  ]
 }

homeassistant/components/unifi_access/__init__.py

@@ -0,0 +1,42 @@
+"""The UniFi Access integration."""
+
+from __future__ import annotations
+
+from uiaccessclient import ApiClient, WebsocketClient
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_API_TOKEN, CONF_HOST, Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import UniFiAccessDoorCoordinator
+from .data import UniFiAccessData
+
+PLATFORMS: list[Platform] = [Platform.LOCK]
+
+type UniFiAccessConfigEntry = ConfigEntry[UniFiAccessData]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: UniFiAccessConfigEntry) -> bool:
+    """Configure UniFi Access integration."""
+    api_client = ApiClient(entry.data.get(CONF_HOST), entry.data.get(CONF_API_TOKEN))
+    websocket_client = WebsocketClient(
+        entry.data.get(CONF_HOST), entry.data.get(CONF_API_TOKEN)
+    )
+
+    door_coordinator = UniFiAccessDoorCoordinator(hass, api_client, websocket_client)
+    await door_coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = UniFiAccessData(
+        api_client=api_client,
+        door_coordinator=door_coordinator,
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: UniFiAccessConfigEntry
+) -> bool:
+    """Unload UniFi Access integration."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/unifi_access/config_flow.py

@@ -0,0 +1,69 @@
+"""Config flow for the UniFi Access integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from uiaccessclient import ApiClient, SpaceApi
+from uiaccessclient.openapi.exceptions import ForbiddenException, UnauthorizedException
+import urllib3.exceptions
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_API_TOKEN, CONF_HOST
+from homeassistant.core import HomeAssistant
+
+from .const import DEFAULT_HOSTNAME, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST, default=DEFAULT_HOSTNAME): str,
+        vol.Required(CONF_API_TOKEN): str,
+    }
+)
+
+
+class UniFiAccessConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for UniFi Access."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Process configuration form."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
+
+            await _validate_input(self.hass, user_input, errors)
+            if not errors:
+                return self.async_create_entry(title="UniFi Access", data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+
+async def _validate_input(
+    hass: HomeAssistant, data: dict[str, Any], errors: dict[str, str]
+) -> None:
+    api_client = ApiClient(data[CONF_HOST], data[CONF_API_TOKEN])
+    space_api = SpaceApi(api_client)
+
+    try:
+        await hass.async_add_executor_job(space_api.fetch_all_doors)
+    except (
+        UnauthorizedException,
+        ForbiddenException,
+    ):
+        errors["base"] = "invalid_auth"
+    except urllib3.exceptions.HTTPError:
+        errors["base"] = "cannot_connect"
+    except Exception:
+        _LOGGER.exception("Unexpected exception")
+        errors["base"] = "unknown"

homeassistant/components/unifi_access/const.py

@@ -0,0 +1,5 @@
+"""Constants for the UniFi Access integration."""
+
+DOMAIN = "unifi_access"
+
+DEFAULT_HOSTNAME = "unifi.localdomain"

homeassistant/components/unifi_access/coordinator.py

@@ -0,0 +1,93 @@
+"""Coordinators for the UniFi Access integration."""
+
+from asyncio import CancelledError, Task
+from contextlib import suppress
+import logging
+
+from uiaccessclient import ApiClient, Door, NotificationEvent, SpaceApi, WebsocketClient
+from uiaccessclient.openapi.exceptions import (
+    ApiException,
+    ForbiddenException,
+    UnauthorizedException,
+)
+
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class UniFiAccessDoorCoordinator(DataUpdateCoordinator[dict[str, Door]]):
+    """Handles refreshing door data."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        api_client: ApiClient,
+        websocket_client: WebsocketClient,
+    ) -> None:
+        """Initialize the door coordinator class."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="UniFi Access door",
+            always_update=False,
+        )
+
+        self.task: Task[None] | None = None
+        self.space_api = SpaceApi(api_client)
+        self.websocket_client = websocket_client
+
+    async def _async_setup(self) -> None:
+        self.task = self.hass.async_create_task(
+            self.receive_updated_data(), eager_start=True
+        )
+
+    async def async_shutdown(self) -> None:
+        """Cancel any scheduled call, and ignore new runs."""
+        await super().async_shutdown()
+
+        if self.task is not None:
+            self.task.cancel()
+            with suppress(CancelledError):
+                await self.task
+
+    async def _async_update_data(self) -> dict[str, Door]:
+        return await self.hass.async_add_executor_job(self._update_data)
+
+    async def receive_updated_data(self) -> None:
+        """Start websocket receiver for updated data from UniFi Access."""
+        _LOGGER.debug("Starting UniFi Access websocket")
+        try:
+            async with self.websocket_client.device_notifications() as socket:
+                async for message in socket:
+                    _LOGGER.debug(
+                        "Received update from UniFi Access: %s", message.event
+                    )
+
+                    if message.event == NotificationEvent.DeviceUpdateV2:
+                        # WebSocket API is poorly documented so we will just use the REST API whenever we get
+                        # an update to fetch all the relevant data.
+                        self.async_set_updated_data(
+                            await self.hass.async_add_executor_job(self._update_data)
+                        )
+        except Exception as exc:
+            _LOGGER.error("Error in UniFi Access websocket receiver: %s", exc)
+            raise
+
+        _LOGGER.debug("UniFi Access websocket receiver has been cancelled")
+
+    def _update_data(self) -> dict[str, Door]:
+        _LOGGER.debug("Refreshing UniFi Access door data")
+        try:
+            response = self.space_api.fetch_all_doors()
+        except (
+            UnauthorizedException,
+            ForbiddenException,
+        ) as err:
+            raise ConfigEntryAuthFailed from err
+        except ApiException as err:
+            raise UpdateFailed(f"Error communicating with API: {err}") from err
+
+        return {door.id: door for door in response.data if door.is_bind_hub}

homeassistant/components/unifi_access/data.py

@@ -0,0 +1,15 @@
+"""Data structures for UniFi Access integration."""
+
+from dataclasses import dataclass
+
+from uiaccessclient import ApiClient
+
+from . import UniFiAccessDoorCoordinator
+
+
+@dataclass
+class UniFiAccessData:
+    """Data structure for UniFi Access integration."""
+
+    api_client: ApiClient
+    door_coordinator: UniFiAccessDoorCoordinator

homeassistant/components/unifi_access/lock.py

@@ -0,0 +1,75 @@
+"""Lock entities for the UniFi Access integration."""
+
+from typing import Any
+
+from uiaccessclient import ApiClient, SpaceApi
+
+from homeassistant.components.lock import LockEntity
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import UniFiAccessConfigEntry, UniFiAccessDoorCoordinator
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: UniFiAccessConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Configure lock entities."""
+    api_client = entry.runtime_data.api_client
+    door_coordinator = entry.runtime_data.door_coordinator
+
+    async_add_entities(
+        UniFiAccessDoorLock(hass, api_client, door_coordinator, door_id)
+        for door_id in door_coordinator.data
+    )
+
+
+class UniFiAccessDoorLock(CoordinatorEntity[UniFiAccessDoorCoordinator], LockEntity):
+    """Represents a UniFi Access door lock."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        api_client: ApiClient,
+        coordinator: UniFiAccessDoorCoordinator,
+        door_id: str,
+    ) -> None:
+        """Initialize the door lock."""
+        super().__init__(coordinator, context=door_id)
+
+        self.hass = hass
+        self.space_api = SpaceApi(api_client)
+
+        self._attr_unique_id = door_id
+        self._update_attributes()
+
+    @property
+    def translation_key(self) -> str:
+        """Return the translation key to translate the entity's states."""
+        return "door_lock"
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        self._update_attributes()
+        super()._handle_coordinator_update()
+
+    def _update_attributes(self) -> None:
+        assert isinstance(self.unique_id, str)
+        door = self.coordinator.data[self.unique_id]
+        self._attr_is_locked = door.door_lock_relay_status == "lock"
+        self._attr_is_open = door.door_position_status == "open"
+
+    async def async_unlock(self, **kwargs: Any) -> None:
+        """Lock the door."""
+        assert isinstance(self.unique_id, str)
+        await self.hass.async_add_executor_job(
+            self.space_api.remote_door_unlocking, self.unique_id
+        )
+
+        self._attr_is_locked = False
+        self.async_write_ha_state()

homeassistant/components/unifi_access/manifest.json

@@ -0,0 +1,13 @@
+{
+  "domain": "unifi_access",
+  "name": "UniFi Access",
+  "codeowners": ["@hagen93"],
+  "config_flow": true,
+  "dependencies": [],
+  "documentation": "https://www.home-assistant.io/integrations/unifi_access",
+  "homekit": {},
+  "iot_class": "local_push",
+  "requirements": ["uiaccessclient==0.9.3"],
+  "ssdp": [],
+  "zeroconf": []
+}