Feat: Add services for managing Schlage door codes

This commit introduces three new services for the Schlage integration:
- `add_code`: Allows users to add a new access code to a Schlage lock.
- `delete_code`: Enables users to delete an existing access code from a
  Schlage lock.
- `get_codes`: Retrieves all access codes associated with a Schlage
  lock.

Signed-off-by: Andrew Grimberg <[email protected]>

Author: tykeal

homeassistant/components/schlage/__init__.py

@@ -10,6 +10,7 @@
 from homeassistant.exceptions import ConfigEntryAuthFailed
 
 from .coordinator import SchlageConfigEntry, SchlageDataUpdateCoordinator
+from .services import async_setup_services
 
 PLATFORMS: list[Platform] = [
     Platform.BINARY_SENSOR,
@@ -33,6 +34,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: SchlageConfigEntry) -> b
         hass, entry, username, pyschlage.Schlage(auth)
     )
     entry.runtime_data = coordinator
+
+    await async_setup_services(hass)
+
     await coordinator.async_config_entry_first_refresh()
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     return True

homeassistant/components/schlage/const.py

@@ -7,3 +7,7 @@
 LOGGER = logging.getLogger(__package__)
 MANUFACTURER = "Schlage"
 UPDATE_INTERVAL = timedelta(seconds=30)
+
+SERVICE_ADD_CODE = "add_code"
+SERVICE_DELETE_CODE = "delete_code"
+SERVICE_GET_CODES = "get_codes"

homeassistant/components/schlage/coordinator.py

@@ -65,7 +65,7 @@ def __init__(
     async def _async_update_data(self) -> SchlageData:
         """Fetch the latest data from the Schlage API."""
         try:
-            locks = await self.hass.async_add_executor_job(self.api.locks)
+            locks = await self.hass.async_add_executor_job(self.api.locks, True)
         except NotAuthorizedError as ex:
             raise ConfigEntryAuthFailed from ex
         except SchlageError as ex:

homeassistant/components/schlage/services.py

@@ -0,0 +1,295 @@
+"""Services for Schlage Lock integration."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+from typing import Any
+
+from pyschlage.code import AccessCode
+
+from homeassistant.core import (
+    HomeAssistant,
+    ServiceCall,
+    ServiceResponse,
+    SupportsResponse,
+)
+from homeassistant.helpers import device_registry as dr, entity_registry as er
+
+from .const import (
+    DOMAIN,
+    LOGGER,
+    SERVICE_ADD_CODE,
+    SERVICE_DELETE_CODE,
+    SERVICE_GET_CODES,
+)
+from .coordinator import SchlageDataUpdateCoordinator
+
+
+def _add_coordinator_from_device(
+    device_id: str,
+    call: ServiceCall,
+    device_registry: dr.DeviceRegistry,
+    coordinators: set[SchlageDataUpdateCoordinator],
+) -> None:
+    """Add coordinator from device ID if found."""
+    LOGGER.debug("Device ID provided: %s", device_id)
+    device_entry = device_registry.async_get(device_id)
+    if not device_entry:
+        LOGGER.error("Device not found in registry: %s", device_id)
+        return
+
+    LOGGER.debug("Device found in registry: %s", device_entry)
+    for config_entry_id in device_entry.config_entries:
+        config_entry = call.hass.config_entries.async_get_entry(config_entry_id)
+        if not config_entry or config_entry.domain != DOMAIN:
+            continue
+
+        LOGGER.debug("Config entry matches domain '%s': %s", DOMAIN, config_entry)
+        coordinator: SchlageDataUpdateCoordinator = config_entry.runtime_data
+        if coordinator:
+            coordinators.add(coordinator)
+            LOGGER.debug("Coordinator added for config entry %s", config_entry.entry_id)
+        else:
+            LOGGER.debug(
+                "No coordinator found for config entry %s", config_entry.entry_id
+            )
+
+
+def _add_coordinator_from_entity(
+    entity_id: str,
+    call: ServiceCall,
+    entity_registry: er.EntityRegistry,
+    coordinators: set[SchlageDataUpdateCoordinator],
+) -> None:
+    """Add coordinator from entity ID if found."""
+    LOGGER.debug("Entity ID provided: %s", entity_id)
+    entity_entry = entity_registry.async_get(entity_id)
+    if not entity_entry:
+        LOGGER.error("Entity not found in registry: %s", entity_id)
+        return
+
+    LOGGER.debug("Entity found in registry: %s", entity_entry)
+    if not entity_entry.config_entry_id:
+        LOGGER.debug("Entity has no config entry ID: %s", entity_id)
+        return
+
+    config_entry = call.hass.config_entries.async_get_entry(
+        entity_entry.config_entry_id
+    )
+    if not config_entry or config_entry.domain != DOMAIN:
+        LOGGER.debug(
+            "Config entry does not match domain '%s': %s", DOMAIN, config_entry
+        )
+        return
+
+    LOGGER.debug("Config entry matches domain '%s': %s", DOMAIN, config_entry)
+    coordinator: SchlageDataUpdateCoordinator = config_entry.runtime_data
+    if coordinator:
+        coordinators.add(coordinator)
+        LOGGER.debug("Coordinator added for config entry %s", config_entry.entry_id)
+    else:
+        LOGGER.debug("No coordinator found for config entry %s", config_entry.entry_id)
+
+
+def _get_coordinator(call: ServiceCall) -> set[SchlageDataUpdateCoordinator]:
+    """Get all unique coordinators related to the service call."""
+    coordinators: set[SchlageDataUpdateCoordinator] = set()
+    device_registry = dr.async_get(call.hass)
+    entity_registry = er.async_get(call.hass)
+
+    if "device_id" in call.data:
+        for device_id in call.data["device_id"]:
+            _add_coordinator_from_device(device_id, call, device_registry, coordinators)
+
+    if "entity_id" in call.data:
+        for entity_id in call.data["entity_id"]:
+            _add_coordinator_from_entity(entity_id, call, entity_registry, coordinators)
+
+    return coordinators
+
+
+def _get_lock_id_from_entity_id(hass: HomeAssistant, entity_id: str) -> str | None:
+    """Extract the lock ID from the entity ID."""
+
+    entity_registry = er.async_get(hass)
+    entity_entry = entity_registry.async_get(entity_id)
+    if entity_entry and entity_entry.unique_id:
+        return entity_entry.unique_id
+    return None
+
+
+def _get_codes_from_lock(
+    coordinator: SchlageDataUpdateCoordinator, lock_id: str
+) -> dict[str, Any]:
+    """Get access codes from a specific lock."""
+    if lock_id in coordinator.data.locks:
+        lock_data = coordinator.data.locks[lock_id]
+        if lock_data.lock.access_codes:
+            return {
+                code: {
+                    "name": lock_data.lock.access_codes[code].name,
+                    "code": lock_data.lock.access_codes[code].code,
+                }
+                for code in lock_data.lock.access_codes
+            }
+    return {}
+
+
+def _get_code_id_from_name(codes: dict[str, Any], name: str) -> str | None:
+    """Get the code ID from the name."""
+    for code_id, code_data in codes.items():
+        if code_data["name"].lower() == name.lower():
+            return code_id
+    return None
+
+
+async def service_get_codes(call: ServiceCall) -> ServiceResponse:
+    """Handle the get_codes service call."""
+
+    LOGGER.debug("Service call to get_codes received: %s", call)
+    LOGGER.debug("Service call data: %s", call.data)
+
+    coordinators: set[SchlageDataUpdateCoordinator] = _get_coordinator(call)
+    LOGGER.debug("Coordinators found: %s", coordinators)
+
+    resp: dict[str, Any] = {}
+
+    if not coordinators:
+        LOGGER.error("No coordinators found for the provided device or entity IDs")
+
+    if "entity_id" in call.data:
+        for entity_id in call.data["entity_id"]:
+            LOGGER.debug("Processing entity ID: %s", entity_id)
+            lock_id = _get_lock_id_from_entity_id(call.hass, entity_id)
+            if lock_id:
+                LOGGER.debug("Lock ID extracted from entity ID: %s", lock_id)
+                for coordinator in coordinators:
+                    LOGGER.debug("Processing coordinator: %s", coordinator)
+                    resp.update({entity_id: _get_codes_from_lock(coordinator, lock_id)})
+            else:
+                error_msg = f"No lock ID found for entity ID: {entity_id}"
+                LOGGER.error(error_msg)
+
+    return resp
+
+
+async def service_add_code(call: ServiceCall) -> None:
+    """Handle the add_code service call."""
+    LOGGER.debug("Service call to add_code received: %s", call)
+
+    coordinators: set[SchlageDataUpdateCoordinator] = _get_coordinator(call)
+
+    if "entity_id" not in call.data:
+        raise ValueError("The 'entity_id' field is required in the service call.")
+
+    if not coordinators:
+        raise ValueError("No coordinators found for the provided device or entity IDs.")
+
+    # validate code and name
+    code = call.data["code"]
+    code_pattern = re.compile(r"^\d{4,8}$")
+    if not isinstance(code, str) or not code_pattern.match(code):
+        raise ValueError("Code must be a string between 4 and 8 digits long.")
+
+    name = call.data["name"]
+
+    for entity_id in call.data["entity_id"]:
+        LOGGER.debug("Processing entity ID: %s", entity_id)
+        lock_id = _get_lock_id_from_entity_id(call.hass, entity_id)
+        if lock_id:
+            LOGGER.debug("Lock ID extracted from entity ID: %s", lock_id)
+            for coordinator in coordinators:
+                LOGGER.debug("Processing coordinator: %s", coordinator)
+                codes = _get_codes_from_lock(coordinator, lock_id)
+                if codes:
+                    if any(
+                        code in inner_codes.values() for inner_codes in codes.values()
+                    ):
+                        error_msg = f"Code already exists for lock ID {lock_id}: {code}"
+                        raise ValueError(error_msg)
+                    if any(
+                        name.lower() in str(inner_codes.values()).lower()
+                        for inner_codes in codes.values()
+                    ):
+                        error_msg = f"Name already exists for lock ID {lock_id}: {name}"
+                        raise ValueError(error_msg)
+
+                if lock_id in coordinator.data.locks:
+                    lock = coordinator.data.locks[lock_id].lock
+                    LOGGER.info(
+                        "Adding code for lock ID %s with name: %s, code: %s",
+                        lock_id,
+                        name,
+                        code,
+                    )
+                    access_code = AccessCode(
+                        name=name,
+                        code=code,
+                    )
+                    await asyncio.to_thread(lock.add_access_code, access_code)
+        else:
+            error_msg = f"No lock ID found for entity ID: {entity_id}"
+            raise ValueError(error_msg)
+
+
+async def service_delete_code(call: ServiceCall) -> None:
+    """Handle the delete_code service call."""
+    LOGGER.debug("Service call to delete_code received: %s", call)
+
+    coordinators: set[SchlageDataUpdateCoordinator] = _get_coordinator(call)
+
+    if "entity_id" not in call.data:
+        raise ValueError("The 'entity_id' field is required in the service call.")
+
+    if not coordinators:
+        raise ValueError("No coordinators found for the provided device or entity IDs.")
+    name = call.data["name"]
+
+    for entity_id in call.data["entity_id"]:
+        LOGGER.debug("Processing entity ID: %s", entity_id)
+        lock_id = _get_lock_id_from_entity_id(call.hass, entity_id)
+        if lock_id:
+            LOGGER.debug("Lock ID extracted from entity ID: %s", lock_id)
+            for coordinator in coordinators:
+                LOGGER.debug("Processing coordinator: %s", coordinator)
+                codes = _get_codes_from_lock(coordinator, lock_id)
+                if codes:
+                    code_id = _get_code_id_from_name(codes, name)
+                    if not code_id:
+                        continue  # No code found with the given name
+                    LOGGER.debug(
+                        "Deleting code with ID %s for lock ID %s", code_id, lock_id
+                    )
+                    if lock_id in coordinator.data.locks:
+                        lock = coordinator.data.locks[lock_id].lock
+                        if lock.access_codes and code_id in lock.access_codes:
+                            await asyncio.to_thread(lock.access_codes[code_id].delete)
+                            LOGGER.debug(
+                                "Code with ID %s deleted for lock ID %s",
+                                code_id,
+                                lock_id,
+                            )
+
+
+async def async_setup_services(hass: HomeAssistant) -> None:
+    """Set up Schlage services."""
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_GET_CODES,
+        service_get_codes,
+        supports_response=SupportsResponse.ONLY,
+    )
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_ADD_CODE,
+        service_add_code,
+    )
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_DELETE_CODE,
+        service_delete_code,
+    )

homeassistant/components/schlage/services.yaml

@@ -0,0 +1,66 @@
+get_codes:
+  name: Get Access Codes
+  description: Retrieve access codes from the lock.
+  fields:
+    entity_id:
+      description: The ID of the lock to retrieve access codes from.
+      required: true
+      selector:
+        entity:
+          filter:
+            domain: lock
+            integration: schlage
+          multiple: true
+
+add_code:
+  name: Add Access Code
+  description: Add a new access code to the lock.
+  fields:
+    entity_id:
+      description: The ID of the lock to retrieve access codes from.
+      required: true
+      selector:
+        entity:
+          filter:
+            domain: lock
+            integration: schlage
+          multiple: true
+    code:
+      required: true
+      description: |
+        The access code to add. Must be unique to the lock and be
+        between 4 and 8 digits long.
+      example: "1111"
+      selector:
+        text:
+          multiline: false
+    name:
+      required: true
+      description: |
+        The name associated with the access code. Must be unique to the lock.
+      example: "Example Person"
+      selector:
+        text:
+          multiline: false
+
+delete_code:
+  name: Delete Access Code
+  description: Delete an access code from the lock.
+  fields:
+    entity_id:
+      description: The ID of the lock to delete access codes from.
+      required: true
+      selector:
+        entity:
+          filter:
+            domain: lock
+            integration: schlage
+          multiple: true
+    name:
+      required: true
+      description: |
+        The name associated with the access code. Must be unique to the lock.
+      example: "Example Person"
+      selector:
+        text:
+          multiline: false