Add Schlage integration (home-assistant#93777)

Co-authored-by: J. Nick Koston <[email protected]>
Co-authored-by: Marc Mueller <[email protected]>

CODEOWNERS

@@ -1077,6 +1077,8 @@ build.json @home-assistant/supervisor
 /tests/components/scene/ @home-assistant/core
 /homeassistant/components/schedule/ @home-assistant/core
 /tests/components/schedule/ @home-assistant/core
+/homeassistant/components/schlage/ @dknowles2
+/tests/components/schlage/ @dknowles2
 /homeassistant/components/schluter/ @prairieapps
 /homeassistant/components/scrape/ @fabaff @gjohansson-ST @epenet
 /tests/components/scrape/ @fabaff @gjohansson-ST @epenet

homeassistant/components/schlage/__init__.py

@@ -0,0 +1,39 @@
+"""The Schlage integration."""
+from __future__ import annotations
+
+from pycognito.exceptions import WarrantException
+import pyschlage
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN, LOGGER
+from .coordinator import SchlageDataUpdateCoordinator
+
+PLATFORMS: list[Platform] = [Platform.LOCK]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Schlage from a config entry."""
+    username = entry.data[CONF_USERNAME]
+    password = entry.data[CONF_PASSWORD]
+    try:
+        auth = await hass.async_add_executor_job(pyschlage.Auth, username, password)
+    except WarrantException as ex:
+        LOGGER.error("Schlage authentication failed: %s", ex)
+        return False
+
+    coordinator = SchlageDataUpdateCoordinator(hass, username, pyschlage.Schlage(auth))
+    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."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/schlage/config_flow.py

@@ -0,0 +1,58 @@
+"""Config flow for Schlage integration."""
+from __future__ import annotations
+
+from typing import Any
+
+import pyschlage
+from pyschlage.exceptions import NotAuthorizedError
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DOMAIN, LOGGER
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
+)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Schlage."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            username = user_input[CONF_USERNAME]
+            password = user_input[CONF_PASSWORD]
+            try:
+                user_id = await self.hass.async_add_executor_job(
+                    _authenticate, username, password
+                )
+            except NotAuthorizedError:
+                errors["base"] = "invalid_auth"
+            except Exception:  # pylint: disable=broad-except
+                LOGGER.exception("Unknown error")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(user_id)
+                return self.async_create_entry(title=username, data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+
+def _authenticate(username: str, password: str) -> str:
+    """Authenticate with the Schlage API."""
+    auth = pyschlage.Auth(username, password)
+    auth.authenticate()
+    # The user_id property will make a blocking call if it's not already
+    # cached. To avoid blocking the event loop, we read it here.
+    return auth.user_id

homeassistant/components/schlage/const.py

@@ -0,0 +1,9 @@
+"""Constants for the Schlage integration."""
+
+from datetime import timedelta
+import logging
+
+DOMAIN = "schlage"
+LOGGER = logging.getLogger(__package__)
+MANUFACTURER = "Schlage"
+UPDATE_INTERVAL = timedelta(seconds=30)

homeassistant/components/schlage/coordinator.py

@@ -0,0 +1,41 @@
+"""DataUpdateCoordinator for the Schlage integration."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from pyschlage import Lock, Schlage
+from pyschlage.exceptions import Error
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN, LOGGER, UPDATE_INTERVAL
+
+
+@dataclass
+class SchlageData:
+    """Container for cached data from the Schlage API."""
+
+    locks: dict[str, Lock]
+
+
+class SchlageDataUpdateCoordinator(DataUpdateCoordinator[SchlageData]):
+    """The Schlage data update coordinator."""
+
+    def __init__(self, hass: HomeAssistant, username: str, api: Schlage) -> None:
+        """Initialize the class."""
+        super().__init__(
+            hass, LOGGER, name=f"{DOMAIN} ({username})", update_interval=UPDATE_INTERVAL
+        )
+        self.api = api
+
+    async def _async_update_data(self) -> SchlageData:
+        """Fetch the latest data from the Schlage API."""
+        try:
+            return await self.hass.async_add_executor_job(self._update_data)
+        except Error as ex:
+            raise UpdateFailed("Failed to refresh Schlage data") from ex
+
+    def _update_data(self) -> SchlageData:
+        """Fetch the latest data from the Schlage API."""
+        return SchlageData(locks={lock.device_id: lock for lock in self.api.locks()})

homeassistant/components/schlage/lock.py

@@ -0,0 +1,84 @@
+"""Platform for Schlage lock integration."""
+from __future__ import annotations
+
+from typing import Any
+
+from pyschlage.lock import Lock
+
+from homeassistant.components.lock import LockEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN, MANUFACTURER
+from .coordinator import SchlageDataUpdateCoordinator
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Schlage WiFi locks based on a config entry."""
+    coordinator: SchlageDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
+    async_add_entities(
+        SchlageLockEntity(coordinator=coordinator, device_id=device_id)
+        for device_id in coordinator.data.locks
+    )
+
+
+class SchlageLockEntity(CoordinatorEntity[SchlageDataUpdateCoordinator], LockEntity):
+    """Schlage lock entity."""
+
+    _attr_has_entity_name = True
+    _attr_name = None
+
+    def __init__(
+        self, coordinator: SchlageDataUpdateCoordinator, device_id: str
+    ) -> None:
+        """Initialize a Schlage Lock."""
+        super().__init__(coordinator=coordinator)
+        self.device_id = device_id
+        self._attr_unique_id = device_id
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device_id)},
+            name=self._lock.name,
+            manufacturer=MANUFACTURER,
+            model=self._lock.model_name,
+            sw_version=self._lock.firmware_version,
+        )
+        self._update_attrs()
+
+    @property
+    def _lock(self) -> Lock:
+        """Fetch the Schlage lock from our coordinator."""
+        return self.coordinator.data.locks[self.device_id]
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        # When is_locked is None the lock is unavailable.
+        return super().available and self._lock.is_locked is not None
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self._update_attrs()
+        return super()._handle_coordinator_update()
+
+    def _update_attrs(self) -> None:
+        """Update our internal state attributes."""
+        self._attr_is_locked = self._lock.is_locked
+        self._attr_is_jammed = self._lock.is_jammed
+
+    async def async_lock(self, **kwargs: Any) -> None:
+        """Lock the device."""
+        await self.hass.async_add_executor_job(self._lock.lock)
+        await self.coordinator.async_request_refresh()
+
+    async def async_unlock(self, **kwargs: Any) -> None:
+        """Unlock the device."""
+        await self.hass.async_add_executor_job(self._lock.unlock)
+        await self.coordinator.async_request_refresh()

homeassistant/components/schlage/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "schlage",
+  "name": "Schlage",
+  "codeowners": ["@dknowles2"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/schlage",
+  "iot_class": "cloud_polling",
+  "requirements": ["pyschlage==2023.5.0"]
+}

homeassistant/components/schlage/strings.json

@@ -0,0 +1,19 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "username": "[%key:common::config_flow::data::username%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        }
+      }
+    },
+    "error": {
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -395,6 +395,7 @@
         "rympro",
         "sabnzbd",
         "samsungtv",
+        "schlage",
         "scrape",
         "screenlogic",
         "season",

homeassistant/generated/integrations.json

@@ -4849,6 +4849,12 @@
       "config_flow": false,
       "iot_class": "local_push"
     },
+    "schlage": {
+      "name": "Schlage",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "cloud_polling"
+    },
     "schluter": {
       "name": "Schluter",
       "integration_type": "hub",

requirements_all.txt

@@ -1981,6 +1981,9 @@ pysabnzbd==1.1.1
 # homeassistant.components.saj
 pysaj==0.0.16
 
+# homeassistant.components.schlage
+pyschlage==2023.5.0
+
 # homeassistant.components.sensibo
 pysensibo==1.0.31
 

requirements_test_all.txt

@@ -1473,6 +1473,9 @@ pyrympro==0.0.7
 # homeassistant.components.sabnzbd
 pysabnzbd==1.1.1
 
+# homeassistant.components.schlage
+pyschlage==2023.5.0
+
 # homeassistant.components.sensibo
 pysensibo==1.0.31
 

tests/components/schlage/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Schlage integration."""

tests/components/schlage/conftest.py

@@ -0,0 +1,48 @@
+"""Common fixtures for the Schlage tests."""
+from collections.abc import Generator
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from homeassistant.components.schlage.const import DOMAIN
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture
+def mock_config_entry() -> MockConfigEntry:
+    """Mock ConfigEntry."""
+    return MockConfigEntry(
+        title="[email protected]",
+        domain=DOMAIN,
+        data={
+            CONF_USERNAME: "[email protected]",
+            CONF_PASSWORD: "hunter2",
+        },
+        unique_id="abc123",
+    )
+
+
+@pytest.fixture
+def mock_setup_entry() -> Generator[AsyncMock, None, None]:
+    """Override async_setup_entry."""
+    with patch(
+        "homeassistant.components.schlage.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        yield mock_setup_entry
+
+
+@pytest.fixture
+def mock_schlage():
+    """Mock pyschlage.Schlage."""
+    with patch("pyschlage.Schlage", autospec=True) as mock_schlage:
+        yield mock_schlage.return_value
+
+
+@pytest.fixture
+def mock_pyschlage_auth():
+    """Mock pyschlage.Auth."""
+    with patch("pyschlage.Auth", autospec=True) as mock_auth:
+        mock_auth.return_value.user_id = "abc123"
+        yield mock_auth.return_value