ZHA lock code services and events

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

@@ -23,6 +23,27 @@ async def async_update(self):
                 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,63 @@ def attribute_updated(self, attrid, value):
                 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."""
+
+        set_pin_code = self.__getattr__("set_pin_code")
+        await 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."""
+
+        set_user_status = self.__getattr__("set_user_status")
+        await 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."""
+
+        set_user_status = self.__getattr__("set_user_status")
+        await 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."""
+
+        get_pin_code = self.__getattr__("get_pin_code")
+        result = await get_pin_code(*(code_slot - 1,))
+        return result
+
+    async def async_clear_user_code(self, code_slot: int) -> None:
+        """Clear the code slot."""
+
+        clear_pin_code = self.__getattr__("clear_pin_code")
+        await clear_pin_code(*(code_slot - 1,))
+
+    async def async_clear_all_user_codes(self) -> None:
+        """Clear all code slots."""
+
+        clear_all_pin_codes = self.__getattr__("clear_all_pin_codes")
+        await clear_all_pin_codes(*())
+
+    async def async_set_user_type(self, code_slot: int, user_type: str) -> None:
+        """Set user type."""
+
+        set_user_type = self.__getattr__("set_user_type")
+        await set_user_type(*(code_slot - 1, user_type))
+
+    async def async_get_user_type(self, code_slot: int) -> str:
+        """Get user type."""
+
+        get_user_type = self.__getattr__("get_user_type")
+        result = await 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 @@
     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 @@
 
 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 @@ async def async_get_state(self, from_cache=True):
     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 @@
 
 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