Add support for dormakaba dKey locks

.coveragerc

@@ -212,6 +212,8 @@ omit =
     homeassistant/components/doorbird/camera.py
     homeassistant/components/doorbird/entity.py
     homeassistant/components/doorbird/util.py
+    homeassistant/components/dormakaba_dkey/__init__.py
+    homeassistant/components/dormakaba_dkey/lock.py
     homeassistant/components/dovado/*
     homeassistant/components/downloader/*
     homeassistant/components/dsmr_reader/__init__.py

CODEOWNERS

@@ -275,6 +275,8 @@ build.json @home-assistant/supervisor
 /tests/components/dnsip/ @gjohansson-ST
 /homeassistant/components/doorbird/ @oblogic7 @bdraco @flacjacket
 /tests/components/doorbird/ @oblogic7 @bdraco @flacjacket
+/homeassistant/components/dormakaba_dkey/ @emontnemery
+/tests/components/dormakaba_dkey/ @emontnemery
 /homeassistant/components/dsmr/ @Robbie1221 @frenck
 /tests/components/dsmr/ @Robbie1221 @frenck
 /homeassistant/components/dsmr_reader/ @depl0y @glodenox

homeassistant/components/dormakaba_dkey/__init__.py

@@ -0,0 +1,97 @@
+"""The Dormakaba dKey integration."""
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+
+from py_dormakaba_dkey import DKEYLock
+from py_dormakaba_dkey.errors import DKEY_EXCEPTIONS
+from py_dormakaba_dkey.models import AssociationData
+
+from homeassistant.components import bluetooth
+from homeassistant.components.bluetooth.match import ADDRESS, BluetoothCallbackMatcher
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ADDRESS, EVENT_HOMEASSISTANT_STOP, Platform
+from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import CONF_ASSOCIATION_DATA, DOMAIN, UPDATE_SECONDS
+from .models import DormakabaDkeyData
+
+PLATFORMS: list[Platform] = [Platform.LOCK]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Dormakaba dKey from a config entry."""
+    address: str = entry.data[CONF_ADDRESS]
+    ble_device = bluetooth.async_ble_device_from_address(hass, address.upper(), True)
+    if not ble_device:
+        raise ConfigEntryNotReady(f"Could not find dKey device with address {address}")
+
+    lock = DKEYLock(ble_device)
+    lock.set_association_data(
+        AssociationData.from_json(entry.data[CONF_ASSOCIATION_DATA])
+    )
+
+    @callback
+    def _async_update_ble(
+        service_info: bluetooth.BluetoothServiceInfoBleak,
+        change: bluetooth.BluetoothChange,
+    ) -> None:
+        """Update from a ble callback."""
+        lock.set_ble_device_and_advertisement_data(
+            service_info.device, service_info.advertisement
+        )
+
+    entry.async_on_unload(
+        bluetooth.async_register_callback(
+            hass,
+            _async_update_ble,
+            BluetoothCallbackMatcher({ADDRESS: address}),
+            bluetooth.BluetoothScanningMode.PASSIVE,
+        )
+    )
+
+    async def _async_update() -> None:
+        """Update the device state."""
+        try:
+            await lock.update()
+            await lock.disconnect()
+        except DKEY_EXCEPTIONS as ex:
+            raise UpdateFailed(str(ex)) from ex
+
+    coordinator = DataUpdateCoordinator(
+        hass,
+        _LOGGER,
+        name=lock.name,
+        update_method=_async_update,
+        update_interval=timedelta(seconds=UPDATE_SECONDS),
+    )
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = DormakabaDkeyData(
+        lock, coordinator
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    async def _async_stop(event: Event) -> None:
+        """Close the connection."""
+        await lock.disconnect()
+
+    entry.async_on_unload(
+        hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_stop)
+    )
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        data: DormakabaDkeyData = hass.data[DOMAIN].pop(entry.entry_id)
+        await data.lock.disconnect()
+
+    return unload_ok

homeassistant/components/dormakaba_dkey/config_flow.py

@@ -0,0 +1,157 @@
+"""Config flow for Dormakaba dKey integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from bleak import BleakError
+from py_dormakaba_dkey import DKEYLock, device_filter, errors as dkey_errors
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components.bluetooth import (
+    BluetoothServiceInfoBleak,
+    async_discovered_service_info,
+)
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import CONF_ASSOCIATION_DATA, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_ASSOCIATE_SCHEMA = vol.Schema(
+    {
+        vol.Required("activation_code"): str,
+    }
+)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Dormakaba dKey."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._lock: DKEYLock | None = None
+        # Populated by user step
+        self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {}
+        # Populated by bluetooth and user steps
+        self._discovery_info: BluetoothServiceInfoBleak | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the user step to pick discovered device."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            address = user_input[CONF_ADDRESS]
+            await self.async_set_unique_id(address, raise_on_progress=False)
+            # Guard against the user selecting a device which has been configured by
+            # another flow.
+            self._abort_if_unique_id_configured()
+            self._discovery_info = self._discovered_devices[address]
+            return await self.async_step_associate()
+
+        current_addresses = self._async_current_ids()
+        for discovery in async_discovered_service_info(self.hass):
+            if (
+                discovery.address in current_addresses
+                or discovery.address in self._discovered_devices
+                or not device_filter(discovery.advertisement)
+            ):
+                continue
+            self._discovered_devices[discovery.address] = discovery
+
+        if not self._discovered_devices:
+            return self.async_abort(reason="no_devices_found")
+
+        data_schema = vol.Schema(
+            {
+                vol.Required(CONF_ADDRESS): vol.In(
+                    {
+                        service_info.address: (
+                            f"{service_info.name} ({service_info.address})"
+                        )
+                        for service_info in self._discovered_devices.values()
+                    }
+                ),
+            }
+        )
+        return self.async_show_form(
+            step_id="user",
+            data_schema=data_schema,
+            errors=errors,
+        )
+
+    async def async_step_bluetooth(
+        self, discovery_info: BluetoothServiceInfoBleak
+    ) -> FlowResult:
+        """Handle the Bluetooth discovery step."""
+        await self.async_set_unique_id(discovery_info.address)
+        self._abort_if_unique_id_configured()
+        self._discovery_info = discovery_info
+        name = self._discovery_info.name or self._discovery_info.address
+        self.context["title_placeholders"] = {"name": name}
+        return await self.async_step_bluetooth_confirm()
+
+    async def async_step_bluetooth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle bluetooth confirm step."""
+        # mypy is not aware that we can't get here without having these set already
+        assert self._discovery_info is not None
+
+        if user_input is None:
+            name = self._discovery_info.name or self._discovery_info.address
+            return self.async_show_form(
+                step_id="bluetooth_confirm",
+                description_placeholders={"name": name},
+            )
+
+        return await self.async_step_associate()
+
+    async def async_step_associate(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle associate step."""
+        # mypy is not aware that we can't get here without having these set already
+        assert self._discovery_info is not None
+
+        if user_input is None:
+            return self.async_show_form(
+                step_id="associate", data_schema=STEP_ASSOCIATE_SCHEMA
+            )
+
+        errors = {}
+        if not self._lock:
+            self._lock = DKEYLock(self._discovery_info.device)
+        lock = self._lock
+
+        try:
+            association_data = await lock.associate(user_input["activation_code"])
+        except BleakError:
+            return self.async_abort(reason="cannot_connect")
+        except dkey_errors.InvalidActivationCode:
+            errors["base"] = "invalid_code"
+        except dkey_errors.WrongActivationCode:
+            errors["base"] = "wrong_code"
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Unexpected exception")
+            return self.async_abort(reason="unknown")
+        else:
+            return self.async_create_entry(
+                title=lock.device_info.device_name
+                or lock.device_info.device_id
+                or lock.name,
+                data={
+                    CONF_ADDRESS: self._discovery_info.device.address,
+                    CONF_ASSOCIATION_DATA: association_data.to_json(),
+                },
+            )
+
+        return self.async_show_form(
+            step_id="associate", data_schema=STEP_ASSOCIATE_SCHEMA, errors=errors
+        )

homeassistant/components/dormakaba_dkey/const.py

@@ -0,0 +1,7 @@
+"""Constants for the Dormakaba dKey integration."""
+
+DOMAIN = "dormakaba_dkey"
+
+UPDATE_SECONDS = 120
+
+CONF_ASSOCIATION_DATA = "association_data"

homeassistant/components/dormakaba_dkey/lock.py

@@ -0,0 +1,84 @@
+"""Dormakaba dKey integration lock platform."""
+from __future__ import annotations
+
+from typing import Any
+
+from py_dormakaba_dkey import DKEYLock
+from py_dormakaba_dkey.commands import Notifications, UnlockStatus
+
+from homeassistant.components.lock import LockEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+)
+
+from .const import DOMAIN
+from .models import DormakabaDkeyData
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the lock platform for Dormakaba dKey."""
+    data: DormakabaDkeyData = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities([DormakabaDkeyLock(data.coordinator, data.lock)])
+
+
+class DormakabaDkeyLock(CoordinatorEntity[DataUpdateCoordinator[None]], LockEntity):
+    """Representation of Dormakaba dKey lock."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self, coordinator: DataUpdateCoordinator[None], lock: DKEYLock
+    ) -> None:
+        """Initialize a Dormakaba dKey lock."""
+        super().__init__(coordinator)
+        self._lock = lock
+        self._attr_unique_id = lock.address
+        self._attr_device_info = DeviceInfo(
+            name=lock.device_info.device_name or lock.device_info.device_id,
+            model="MTL 9291",
+            sw_version=lock.device_info.sw_version,
+            connections={(dr.CONNECTION_BLUETOOTH, lock.address)},
+        )
+        self._async_update_attrs()
+
+    @callback
+    def _async_update_attrs(self) -> None:
+        """Handle updating _attr values."""
+        self._attr_is_locked = self._lock.state.unlock_status in (
+            UnlockStatus.LOCKED,
+            UnlockStatus.SECURITY_LOCKED,
+        )
+
+    async def async_lock(self, **kwargs: Any) -> None:
+        """Lock the lock."""
+        await self._lock.lock()
+
+    async def async_unlock(self, **kwargs: Any) -> None:
+        """Unlock the lock."""
+        await self._lock.unlock()
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle data update."""
+        self._async_update_attrs()
+        self.async_write_ha_state()
+
+    @callback
+    def _handle_state_update(self, update: Notifications) -> None:
+        """Handle data update."""
+        self.coordinator.async_set_updated_data(None)
+
+    async def async_added_to_hass(self) -> None:
+        """Register callbacks."""
+        self.async_on_remove(self._lock.register_callback(self._handle_state_update))
+        return await super().async_added_to_hass()

homeassistant/components/dormakaba_dkey/manifest.json

@@ -0,0 +1,15 @@
+{
+  "domain": "dormakaba_dkey",
+  "name": "Dormakaba dKey",
+  "bluetooth": [
+    { "service_uuid": "e7a60000-6639-429f-94fd-86de8ea26897" },
+    { "service_uuid": "e7a60001-6639-429f-94fd-86de8ea26897" }
+  ],
+  "codeowners": ["@emontnemery"],
+  "config_flow": true,
+  "dependencies": ["bluetooth_adapters"],
+  "documentation": "https://www.home-assistant.io/integrations/dormakaba_dkey",
+  "integration_type": "device",
+  "iot_class": "local_polling",
+  "requirements": ["py-dormakaba-dkey==1.0.1"]
+}

homeassistant/components/dormakaba_dkey/models.py

@@ -0,0 +1,16 @@
+"""The Dormakaba dKey integration models."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from py_dormakaba_dkey import DKEYLock
+
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+
+@dataclass
+class DormakabaDkeyData:
+    """Data for the Dormakaba dKey integration."""
+
+    lock: DKEYLock
+    coordinator: DataUpdateCoordinator[None]