"""Matter lock.""" from __future__ import annotations import asyncio from typing import Any from chip.clusters import Objects as clusters from homeassistant.components.lock import ( LockEntity, LockEntityDescription, LockEntityFeature, ) from homeassistant.config_entries import ConfigEntry from homeassistant.const import ATTR_CODE, Platform from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddEntitiesCallback from .const import LOGGER from .entity import MatterEntity from .helpers import get_matter from .models import MatterDiscoverySchema DoorLockFeature = clusters.DoorLock.Bitmaps.Feature async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: """Set up Matter lock from Config Entry.""" matter = get_matter(hass) matter.register_platform_handler(Platform.LOCK, async_add_entities) class MatterLock(MatterEntity, LockEntity): """Representation of a Matter lock.""" _feature_map: int | None = None _optimistic_timer: asyncio.TimerHandle | None = None _platform_translation_key = "lock" @property def code_format(self) -> str | None: """Regex for code format or None if no code is required.""" if self.get_matter_attribute_value( clusters.DoorLock.Attributes.RequirePINforRemoteOperation ): min_pincode_length = int( self.get_matter_attribute_value( clusters.DoorLock.Attributes.MinPINCodeLength ) ) max_pincode_length = int( self.get_matter_attribute_value( clusters.DoorLock.Attributes.MaxPINCodeLength ) ) return f"^\\d{{{min_pincode_length},{max_pincode_length}}}$" return None async def send_device_command( self, command: clusters.ClusterCommand, timed_request_timeout_ms: int = 1000, ) -> None: """Send a command to the device.""" await self.matter_client.send_device_command( node_id=self._endpoint.node.node_id, endpoint_id=self._endpoint.endpoint_id, command=command, timed_request_timeout_ms=timed_request_timeout_ms, ) async def async_lock(self, **kwargs: Any) -> None: """Lock the lock with pin if needed.""" if not self._attr_is_locked: # optimistically signal locking to state machine self._attr_is_locking = True self.async_write_ha_state() # the lock should acknowledge the command with an attribute update # but bad things may happen, so guard against it with a timer. self._optimistic_timer = self.hass.loop.call_later( 30, self._reset_optimistic_state ) code: str | None = kwargs.get(ATTR_CODE) code_bytes = code.encode() if code else None await self.send_device_command( command=clusters.DoorLock.Commands.LockDoor(code_bytes) ) async def async_unlock(self, **kwargs: Any) -> None: """Unlock the lock with pin if needed.""" if self._attr_is_locked: # optimistically signal unlocking to state machine self._attr_is_unlocking = True self.async_write_ha_state() # the lock should acknowledge the command with an attribute update # but bad things may happen, so guard against it with a timer. self._optimistic_timer = self.hass.loop.call_later( 30, self._reset_optimistic_state ) code: str | None = kwargs.get(ATTR_CODE) code_bytes = code.encode() if code else None if self._attr_supported_features & LockEntityFeature.OPEN: # if the lock reports it has separate unbolt support, # the unlock command should unbolt only on the unlock command # and unlatch on the HA 'open' command. await self.send_device_command( command=clusters.DoorLock.Commands.UnboltDoor(code_bytes) ) else: await self.send_device_command( command=clusters.DoorLock.Commands.UnlockDoor(code_bytes) ) async def async_open(self, **kwargs: Any) -> None: """Open the door latch.""" # optimistically signal opening to state machine self._attr_is_opening = True self.async_write_ha_state() # the lock should acknowledge the command with an attribute update # but bad things may happen, so guard against it with a timer. self._optimistic_timer = self.hass.loop.call_later( 30 if self._attr_is_locked else 5, self._reset_optimistic_state ) code: str | None = kwargs.get(ATTR_CODE) code_bytes = code.encode() if code else None await self.send_device_command( command=clusters.DoorLock.Commands.UnlockDoor(code_bytes) ) @callback def _update_from_device(self) -> None: """Update the entity from the device.""" # always calculate the features as they can dynamically change self._calculate_features() lock_state = self.get_matter_attribute_value( clusters.DoorLock.Attributes.LockState ) # always reset the optimisically (un)locking state on state update self._reset_optimistic_state(write_state=False) LOGGER.debug("Lock state: %s for %s", lock_state, self.entity_id) if lock_state == clusters.DoorLock.Enums.DlLockState.kUnlatched: self._attr_is_locked = False self._attr_is_open = True elif lock_state == clusters.DoorLock.Enums.DlLockState.kLocked: self._attr_is_locked = True self._attr_is_open = False elif lock_state in ( clusters.DoorLock.Enums.DlLockState.kUnlocked, clusters.DoorLock.Enums.DlLockState.kNotFullyLocked, ): self._attr_is_locked = False self._attr_is_open = False else: # Treat any other state as unknown. # NOTE: A null state can happen during device startup. self._attr_is_locked = None self._attr_is_open = None @callback def _reset_optimistic_state(self, write_state: bool = True) -> None: if self._optimistic_timer and not self._optimistic_timer.cancelled(): self._optimistic_timer.cancel() self._optimistic_timer = None self._attr_is_locking = False self._attr_is_unlocking = False self._attr_is_opening = False if write_state: self.async_write_ha_state() @callback def _calculate_features( self, ) -> None: """Calculate features for HA Lock platform from Matter FeatureMap.""" feature_map = int( self.get_matter_attribute_value(clusters.DoorLock.Attributes.FeatureMap) ) # NOTE: the featuremap can dynamically change, so we need to update the # supported features if the featuremap changes. if self._feature_map == feature_map: return self._feature_map = feature_map supported_features = LockEntityFeature(0) # determine if lock supports optional open/unbolt feature if bool(feature_map & DoorLockFeature.kUnbolt): supported_features |= LockEntityFeature.OPEN self._attr_supported_features = supported_features DISCOVERY_SCHEMAS = [ MatterDiscoverySchema( platform=Platform.LOCK, entity_description=LockEntityDescription( key="MatterLock", name=None, ), entity_class=MatterLock, required_attributes=(clusters.DoorLock.Attributes.LockState,), ), ]