diff --git a/homeassistant/components/zha/core/channels/closures.py b/homeassistant/components/zha/core/channels/closures.py index 0326f18ac69..e427bc962b4 100644 --- a/homeassistant/components/zha/core/channels/closures.py +++ b/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): diff --git a/homeassistant/components/zha/lock.py b/homeassistant/components/zha/lock.py index 2ed186d807c..5684b22db6a 100644 --- a/homeassistant/components/zha/lock.py +++ b/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) diff --git a/homeassistant/components/zha/services.yaml b/homeassistant/components/zha/services.yaml index 74793d6000f..e756edbc48b 100644 --- a/homeassistant/components/zha/services.yaml +++ b/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: diff --git a/tests/components/zha/test_lock.py b/tests/components/zha/test_lock.py index 6c464efd7b2..72ba0aba9c5 100644 --- a/tests/components/zha/test_lock.py +++ b/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