ZHA lock code services and events (#47208)

* ZHA lock code services and events

* ZHA Locks: A few more services, use the library functions

* Catch exception when command id is not in command list

* Add tests for lock code services

* Add tests for enable/disable

* Better document code slot ID shifting

* Simplify cluster commands

homeassistant/components/zha/core/channels/closures.py

@@ -23,6 +23,27 @@ class DoorLockChannel(ZigbeeChannel):
                 f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", 0, "lock_state", result
             )
 
+    @callback
+    def cluster_command(self, tsn, command_id, args):
+        """Handle a cluster command received on this cluster."""
+
+        if (
+            self._cluster.client_commands is None
+            or self._cluster.client_commands.get(command_id) is None
+        ):
+            return
+
+        command_name = self._cluster.client_commands.get(command_id, [command_id])[0]
+        if command_name == "operation_event_notification":
+            self.zha_send_event(
+                command_name,
+                {
+                    "source": args[0].name,
+                    "operation": args[1].name,
+                    "code_slot": (args[2] + 1),  # start code slots at 1
+                },
+            )
+
     @callback
     def attribute_updated(self, attrid, value):
         """Handle attribute update from lock cluster."""
@@ -35,6 +56,53 @@ class DoorLockChannel(ZigbeeChannel):
                 f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", attrid, attr_name, value
             )
 
+    async def async_set_user_code(self, code_slot: int, user_code: str) -> None:
+        """Set the user code for the code slot."""
+
+        await self.set_pin_code(
+            code_slot - 1,  # start code slots at 1, Zigbee internals use 0
+            closures.DoorLock.UserStatus.Enabled,
+            closures.DoorLock.UserType.Unrestricted,
+            user_code,
+        )
+
+    async def async_enable_user_code(self, code_slot: int) -> None:
+        """Enable the code slot."""
+
+        await self.set_user_status(code_slot - 1, closures.DoorLock.UserStatus.Enabled)
+
+    async def async_disable_user_code(self, code_slot: int) -> None:
+        """Disable the code slot."""
+
+        await self.set_user_status(code_slot - 1, closures.DoorLock.UserStatus.Disabled)
+
+    async def async_get_user_code(self, code_slot: int) -> int:
+        """Get the user code from the code slot."""
+
+        result = await self.get_pin_code(code_slot - 1)
+        return result
+
+    async def async_clear_user_code(self, code_slot: int) -> None:
+        """Clear the code slot."""
+
+        await self.clear_pin_code(code_slot - 1)
+
+    async def async_clear_all_user_codes(self) -> None:
+        """Clear all code slots."""
+
+        await self.clear_all_pin_codes()
+
+    async def async_set_user_type(self, code_slot: int, user_type: str) -> None:
+        """Set user type."""
+
+        await self.set_user_type(code_slot - 1, user_type)
+
+    async def async_get_user_type(self, code_slot: int) -> str:
+        """Get user type."""
+
+        result = await self.get_user_type(code_slot - 1)
+        return result
+
 
 @registries.ZIGBEE_CHANNEL_REGISTRY.register(closures.Shade.cluster_id)
 class Shade(ZigbeeChannel):

homeassistant/components/zha/lock.py

@@ -1,6 +1,7 @@
 """Locks on Zigbee Home Automation networks."""
 import functools
 
+import voluptuous as vol
 from zigpy.zcl.foundation import Status
 
 from homeassistant.components.lock import (
@@ -10,6 +11,7 @@ from homeassistant.components.lock import (
     LockEntity,
 )
 from homeassistant.core import callback
+from homeassistant.helpers import config_validation as cv, entity_platform
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 
 from .core import discovery
@@ -29,6 +31,11 @@ STRICT_MATCH = functools.partial(ZHA_ENTITIES.strict_match, DOMAIN)
 
 VALUE_TO_STATE = dict(enumerate(STATE_LIST))
 
+SERVICE_SET_LOCK_USER_CODE = "set_lock_user_code"
+SERVICE_ENABLE_LOCK_USER_CODE = "enable_lock_user_code"
+SERVICE_DISABLE_LOCK_USER_CODE = "disable_lock_user_code"
+SERVICE_CLEAR_LOCK_USER_CODE = "clear_lock_user_code"
+
 
 async def async_setup_entry(hass, config_entry, async_add_entities):
     """Set up the Zigbee Home Automation Door Lock from config entry."""
@@ -43,6 +50,42 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
     )
     hass.data[DATA_ZHA][DATA_ZHA_DISPATCHERS].append(unsub)
 
+    platform = entity_platform.current_platform.get()
+    assert platform
+
+    platform.async_register_entity_service(  # type: ignore
+        SERVICE_SET_LOCK_USER_CODE,
+        {
+            vol.Required("code_slot"): vol.Coerce(int),
+            vol.Required("user_code"): cv.string,
+        },
+        "async_set_lock_user_code",
+    )
+
+    platform.async_register_entity_service(  # type: ignore
+        SERVICE_ENABLE_LOCK_USER_CODE,
+        {
+            vol.Required("code_slot"): vol.Coerce(int),
+        },
+        "async_enable_lock_user_code",
+    )
+
+    platform.async_register_entity_service(  # type: ignore
+        SERVICE_DISABLE_LOCK_USER_CODE,
+        {
+            vol.Required("code_slot"): vol.Coerce(int),
+        },
+        "async_disable_lock_user_code",
+    )
+
+    platform.async_register_entity_service(  # type: ignore
+        SERVICE_CLEAR_LOCK_USER_CODE,
+        {
+            vol.Required("code_slot"): vol.Coerce(int),
+        },
+        "async_clear_lock_user_code",
+    )
+
 
 @STRICT_MATCH(channel_names=CHANNEL_DOORLOCK)
 class ZhaDoorLock(ZhaEntity, LockEntity):
@@ -116,3 +159,27 @@ class ZhaDoorLock(ZhaEntity, LockEntity):
     async def refresh(self, time):
         """Call async_get_state at an interval."""
         await self.async_get_state(from_cache=False)
+
+    async def async_set_lock_user_code(self, code_slot: int, user_code: str) -> None:
+        """Set the user_code to index X on the lock."""
+        if self._doorlock_channel:
+            await self._doorlock_channel.async_set_user_code(code_slot, user_code)
+            self.debug("User code at slot %s set", code_slot)
+
+    async def async_enable_lock_user_code(self, code_slot: int) -> None:
+        """Enable user_code at index X on the lock."""
+        if self._doorlock_channel:
+            await self._doorlock_channel.async_enable_user_code(code_slot)
+            self.debug("User code at slot %s enabled", code_slot)
+
+    async def async_disable_lock_user_code(self, code_slot: int) -> None:
+        """Disable user_code at index X on the lock."""
+        if self._doorlock_channel:
+            await self._doorlock_channel.async_disable_user_code(code_slot)
+            self.debug("User code at slot %s disabled", code_slot)
+
+    async def async_clear_lock_user_code(self, code_slot: int) -> None:
+        """Clear the user_code at index X on the lock."""
+        if self._doorlock_channel:
+            await self._doorlock_channel.async_clear_user_code(code_slot)
+            self.debug("User code at slot %s cleared", code_slot)

homeassistant/components/zha/services.yaml

@@ -163,3 +163,74 @@ warning_device_warn:
       description: >-
         Indicates the intensity of the strobe as shown in Table 8-23 of the ZCL spec. This attribute is designed to vary the output of the strobe (i.e., brightness) and not its frequency, which is detailed in section 8.4.2.3.1.6 of the ZCL spec.
       example: 2
+
+clear_lock_user_code:
+  name: Clear lock user
+  description: Clear a user code from a lock
+  target:
+    entity:
+      domain: lock
+      integration: zha
+  fields:
+    code_slot:
+      name: Code slot
+      description: Code slot to clear code from
+      required: true
+      example: 1
+      selector:
+        text:
+
+enable_lock_user_code:
+  name: Enable lock user
+  description: Enable a user code on a lock
+  target:
+    entity:
+      domain: lock
+      integration: zha
+  fields:
+    code_slot:
+      name: Code slot
+      description: Code slot to enable
+      required: true
+      example: 1
+      selector:
+        text:
+
+disable_lock_user_code:
+  name: Disable lock user
+  description: Disable a user code on a lock
+  target:
+    entity:
+      domain: lock
+      integration: zha
+  fields:
+    code_slot:
+      name: Code slot
+      description: Code slot to disable
+      required: true
+      example: 1
+      selector:
+        text:
+
+set_lock_user_code:
+  name: Set lock user code
+  description: Set a user code on a lock
+  target:
+    entity:
+      domain: lock
+      integration: zha
+  fields:
+    code_slot:
+      name: Code slot
+      description: Code slot to set the code in
+      required: true
+      example: 1
+      selector:
+        text:
+    user_code:
+      name: Code
+      description: Code to set
+      required: true
+      example: 1234
+      selector:
+        text:

tests/components/zha/test_lock.py

@@ -16,6 +16,9 @@ from tests.common import mock_coro
 
 LOCK_DOOR = 0
 UNLOCK_DOOR = 1
+SET_PIN_CODE = 5
+CLEAR_PIN_CODE = 7
+SET_USER_STATUS = 9
 
 
 @pytest.fixture
@@ -68,6 +71,18 @@ async def test_lock(hass, lock):
     # unlock from HA
     await async_unlock(hass, cluster, entity_id)
 
+    # set user code
+    await async_set_user_code(hass, cluster, entity_id)
+
+    # clear user code
+    await async_clear_user_code(hass, cluster, entity_id)
+
+    # enable user code
+    await async_enable_user_code(hass, cluster, entity_id)
+
+    # disable user code
+    await async_disable_user_code(hass, cluster, entity_id)
+
 
 async def async_lock(hass, cluster, entity_id):
     """Test lock functionality from hass."""
@@ -95,3 +110,91 @@ async def async_unlock(hass, cluster, entity_id):
         assert cluster.request.call_count == 1
         assert cluster.request.call_args[0][0] is False
         assert cluster.request.call_args[0][1] == UNLOCK_DOOR
+
+
+async def async_set_user_code(hass, cluster, entity_id):
+    """Test set lock code functionality from hass."""
+    with patch(
+        "zigpy.zcl.Cluster.request", return_value=mock_coro([zcl_f.Status.SUCCESS])
+    ):
+        # set lock code via service call
+        await hass.services.async_call(
+            "zha",
+            "set_lock_user_code",
+            {"entity_id": entity_id, "code_slot": 3, "user_code": "13246579"},
+            blocking=True,
+        )
+        assert cluster.request.call_count == 1
+        assert cluster.request.call_args[0][0] is False
+        assert cluster.request.call_args[0][1] == SET_PIN_CODE
+        assert cluster.request.call_args[0][3] == 2  # user slot 3 => internal slot 2
+        assert cluster.request.call_args[0][4] == closures.DoorLock.UserStatus.Enabled
+        assert (
+            cluster.request.call_args[0][5] == closures.DoorLock.UserType.Unrestricted
+        )
+        assert cluster.request.call_args[0][6] == "13246579"
+
+
+async def async_clear_user_code(hass, cluster, entity_id):
+    """Test clear lock code functionality from hass."""
+    with patch(
+        "zigpy.zcl.Cluster.request", return_value=mock_coro([zcl_f.Status.SUCCESS])
+    ):
+        # set lock code via service call
+        await hass.services.async_call(
+            "zha",
+            "clear_lock_user_code",
+            {
+                "entity_id": entity_id,
+                "code_slot": 3,
+            },
+            blocking=True,
+        )
+        assert cluster.request.call_count == 1
+        assert cluster.request.call_args[0][0] is False
+        assert cluster.request.call_args[0][1] == CLEAR_PIN_CODE
+        assert cluster.request.call_args[0][3] == 2  # user slot 3 => internal slot 2
+
+
+async def async_enable_user_code(hass, cluster, entity_id):
+    """Test enable lock code functionality from hass."""
+    with patch(
+        "zigpy.zcl.Cluster.request", return_value=mock_coro([zcl_f.Status.SUCCESS])
+    ):
+        # set lock code via service call
+        await hass.services.async_call(
+            "zha",
+            "enable_lock_user_code",
+            {
+                "entity_id": entity_id,
+                "code_slot": 3,
+            },
+            blocking=True,
+        )
+        assert cluster.request.call_count == 1
+        assert cluster.request.call_args[0][0] is False
+        assert cluster.request.call_args[0][1] == SET_USER_STATUS
+        assert cluster.request.call_args[0][3] == 2  # user slot 3 => internal slot 2
+        assert cluster.request.call_args[0][4] == closures.DoorLock.UserStatus.Enabled
+
+
+async def async_disable_user_code(hass, cluster, entity_id):
+    """Test disable lock code functionality from hass."""
+    with patch(
+        "zigpy.zcl.Cluster.request", return_value=mock_coro([zcl_f.Status.SUCCESS])
+    ):
+        # set lock code via service call
+        await hass.services.async_call(
+            "zha",
+            "disable_lock_user_code",
+            {
+                "entity_id": entity_id,
+                "code_slot": 3,
+            },
+            blocking=True,
+        )
+        assert cluster.request.call_count == 1
+        assert cluster.request.call_args[0][0] is False
+        assert cluster.request.call_args[0][1] == SET_USER_STATUS
+        assert cluster.request.call_args[0][3] == 2  # user slot 3 => internal slot 2
+        assert cluster.request.call_args[0][4] == closures.DoorLock.UserStatus.Disabled