"""ZHA quirk for Aqara Smart Door Lock S2 Pro (ZNMS13LM / lumi.lock.acn03).
Based on the Zigbee2MQTT converter for ZNMS13LM and the actual device signature
from a Z2M backup. The lock reports state and events via manufacturer-specific
attributes on the ``closuresDoorLock`` cluster (0x0101).
Events are emitted as ``zha_event`` and can be used in Home Assistant
automations. The standard ``lock_state`` attribute is updated so the HA lock
entity reflects the current lock/unlock state.
"""
from __future__ import annotations
import logging
from typing import Any, Final
from zigpy import types as t
from zigpy.profiles import zha
from zigpy.zcl.clusters.closures import DoorLock
from zigpy.zcl.clusters.general import (
Basic,
DeviceTemperature,
Groups,
Identify,
Ota,
PollControl,
PowerConfiguration,
Scenes,
Time,
)
from zigpy.zcl.foundation import ZCLAttributeDef
from zigpy.zdo.types import NodeDescriptor
from zhaquirks import CustomCluster
from zhaquirks.const import (
DEVICE_TYPE,
ENDPOINTS,
INPUT_CLUSTERS,
MODELS_INFO,
NODE_DESCRIPTOR,
OUTPUT_CLUSTERS,
PROFILE_ID,
ZHA_SEND_EVENT,
)
from zhaquirks.xiaomi import (
LUMI,
BasicCluster,
XIAOMI_NODE_DESC,
XiaomiCustomDevice,
XiaomiPowerConfiguration,
)
# Manufacturer-specific attributes on the DoorLock cluster used by lumi.lock.acn03.
# Values arrive as ASCII strings starting with "ML".
ATTR_FINGER_PW_SUCCESS = 0xFF10 # 65296
ATTR_FINGER_PW_FAIL_BELL = 0xFF11 # 65297
ATTR_PW_ADD_DEL = 0xFF01 # 65281
ATTR_SET_LANGUAGE = 0xFFE2 # 65522
ATTR_LOCK_STATUS = 0xFFF6 # 65526
# (action, state, reverse) for lock final status reports (attribute 0xFFE6).
LOCK_STATUS_COMMANDS = {
"0341": ("reverse_lock_cancel", "UNLOCK", "UNLOCK"),
"0351": ("reverse_lock_cancel", "LOCK", "UNLOCK"),
"0245": ("reverse_lock", "UNLOCK", "LOCK"),
"0255": ("reverse_lock", "LOCK", "LOCK"),
"1355": ("reverse_lock", "LOCK", "LOCK"),
"1351": ("locked", "LOCK", "UNLOCK"),
"1451": ("locked", "LOCK", "UNLOCK"),
"0640": ("lock_opened_outside", "UNLOCK", "UNLOCK"),
"0600": ("lock_opened_outside", "UNLOCK", "UNLOCK"),
"2300": ("lock_opened_inside", "UNLOCK", "UNLOCK"),
"0540": ("lock_opened_inside", "UNLOCK", "UNLOCK"),
"0440": ("lock_opened_inside", "UNLOCK", "UNLOCK"),
"2400": ("door_closed", "UNLOCK", "UNLOCK"),
"2401": ("door_closed", "UNLOCK", "UNLOCK"),
}
LOCK_STATE_MAP = {
"LOCK": DoorLock.LockState.Locked,
"UNLOCK": DoorLock.LockState.Unlocked,
}
_LOGGER = logging.getLogger(__name__)
class AqaraLockData(t.LVBytes):
"""Raw byte payload reported by Aqara manufacturer-specific attributes.
The lock sends length-prefixed binary data using either the OctetString
(0x41) or CharacterString (0x42) ZCL data type. zigpy decodes
CharacterString as UTF-8 and keeps the original bytes in ``.raw``; this
type restores those raw bytes so the quirk can decode the Aqara protocol.
"""
def __new__(cls, value: Any) -> "AqaraLockData":
if isinstance(value, (bytes, bytearray)):
raw = bytes(value)
elif isinstance(value, str):
# zigpy's CharacterString keeps the raw bytes in `.raw`
raw = getattr(value, "raw", None)
if not isinstance(raw, bytes):
raw = value.encode("latin-1", errors="replace")
else:
raw = bytes(value)
return super().__new__(cls, raw)
def _to_hex(value: bytes | str | Any) -> str | None:
"""Convert an attribute value to a lowercase hex string.
The device sends binary data, but zigpy may present it as ``bytes``, a
``CharacterString`` instance with a ``.raw`` bytes attribute, or a plain
decoded ``str``. This helper recovers the original bytes in all cases.
"""
if isinstance(value, (bytes, bytearray)):
return bytes(value).hex()
if isinstance(value, str):
raw = getattr(value, "raw", None)
if isinstance(raw, bytes):
return raw.hex()
return value.encode("latin-1", errors="replace").hex()
return None
class AqaraDoorLockS2ProCluster(DoorLock):
"""DoorLock cluster for Aqara ZNMS13LM.
Parses Aqara manufacturer-specific attributes and updates the standard
``lock_state`` attribute so Home Assistant sees the lock state. Every
recognised event is also emitted as a ``zha_event``.
"""
class AttributeDefs(DoorLock.AttributeDefs):
"""Manufacturer-specific attributes reported by the lock."""
finger_password_success: Final = ZCLAttributeDef(
id=ATTR_FINGER_PW_SUCCESS,
type=AqaraLockData,
is_manufacturer_specific=True,
)
finger_password_fail_bell: Final = ZCLAttributeDef(
id=ATTR_FINGER_PW_FAIL_BELL,
type=AqaraLockData,
is_manufacturer_specific=True,
)
password_add_delete: Final = ZCLAttributeDef(
id=ATTR_PW_ADD_DEL,
type=AqaraLockData,
is_manufacturer_specific=True,
)
set_language: Final = ZCLAttributeDef(
id=ATTR_SET_LANGUAGE,
type=AqaraLockData,
is_manufacturer_specific=True,
)
lock_status: Final = ZCLAttributeDef(
id=ATTR_LOCK_STATUS,
type=AqaraLockData,
is_manufacturer_specific=True,
)
def _update_attribute(self, attrid: int, value: t.TypeValue) -> None:
"""Intercept manufacturer reports and parse them."""
hex_value = _to_hex(value)
_LOGGER.debug(
"[%s] DoorLock attribute report: attrid=0x%04x type=%s hex=%s",
self.endpoint.device.ieee,
attrid,
type(value).__name__,
hex_value,
)
if attrid == self.AttributeDefs.lock_status.id:
self._parse_lock_status(value)
elif attrid == self.AttributeDefs.finger_password_success.id:
self._parse_finger_password_success(value)
elif attrid == self.AttributeDefs.finger_password_fail_bell.id:
self._parse_finger_password_fail_bell(value)
elif attrid == self.AttributeDefs.password_add_delete.id:
self._parse_password_add_delete(value)
elif attrid == self.AttributeDefs.set_language.id:
self._parse_set_language(value)
# Always store the raw value as well, so it remains visible in the
# cluster management UI for debugging.
super()._update_attribute(attrid, value)
def _update_lock_state(self, state: str) -> None:
"""Update the standard DoorLock ``lock_state`` attribute."""
lock_state = LOCK_STATE_MAP.get(state)
if lock_state is not None:
super()._update_attribute(
DoorLock.AttributeDefs.lock_state.id, lock_state
)
def _parse_lock_status(self, value: bytes | str) -> None:
"""Parse attribute 0xFFE6 (final lock status)."""
data = _to_hex(value)
if not data or len(data) < 10:
return
command = data[6:10]
result = LOCK_STATUS_COMMANDS.get(command)
if result is None:
return
action, state, reverse = result
self._update_lock_state(state)
self.listener_event(
ZHA_SEND_EVENT,
action,
{"state": state, "reverse": reverse, "command": command},
)
def _parse_finger_password_success(self, value: bytes | str) -> None:
"""Parse attribute 0xFF10 (successful finger/password unlock)."""
data = _to_hex(value)
if not data or len(data) < 14:
return
command = data[6:8]
user_type = data[8:9]
user_id = data[12:14]
base_action = {"01": "finger_open", "02": "password_open"}.get(command)
if base_action is None:
return
suffix = "_admin" if user_type == "1" else "_user"
action = f"{base_action}{suffix}_id{int(user_id, 16)}"
action_user = int(user_id, 16)
self.listener_event(
ZHA_SEND_EVENT,
action,
{
"action_user": action_user,
"user_type": "admin" if user_type == "1" else "user",
},
)
def _parse_finger_password_fail_bell(self, value: bytes | str) -> None:
"""Parse attribute 0xFF11 (failed auth or doorbell)."""
data = _to_hex(value)
if not data or len(data) < 14:
return
times = data[6:8]
type_ = data[12:14]
if type_ == "40":
action = "finger_not_match"
repeat = int(times, 16)
elif type_ == "02":
action = "password_not_match"
repeat = int(times, 16)
elif type_ == "00":
action = "ring_bell"
repeat = None
else:
return
event_args: dict[str, int | None] = {}
if repeat is not None:
event_args["action_repeat"] = repeat
self.listener_event(ZHA_SEND_EVENT, action, event_args)
def _parse_password_add_delete(self, value: bytes | str) -> None:
"""Parse attribute 0xFF01 (password/finger add or delete)."""
data = _to_hex(value)
if not data or len(data) < 20:
return
command = data[18:20]
user_id = data[12:14]
action = {"01": "finger_add", "02": "finger_delete"}.get(command)
if action is None:
return
action_user = int(user_id, 16)
self.listener_event(
ZHA_SEND_EVENT,
action,
{"action_user": action_user},
)
def _parse_set_language(self, value: bytes | str) -> None:
"""Parse attribute 0xFFE2 (language change)."""
data = _to_hex(value)
if not data or len(data) < 8:
return
lang_id = data[6:8]
action = (
"change_language_to_english"
if lang_id == "2"
else "change_language_to_chinese"
)
self.listener_event(ZHA_SEND_EVENT, action, {})
class AqaraLockS2Pro(XiaomiCustomDevice):
"""Aqara Smart Door Lock S2 Pro (ZNMS13LM / lumi.lock.acn03)."""
signature = {
MODELS_INFO: [(LUMI, "lumi.lock.acn03")],
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.DOOR_LOCK,
INPUT_CLUSTERS: [
Basic.cluster_id,
Identify.cluster_id,
PowerConfiguration.cluster_id,
DeviceTemperature.cluster_id,
Scenes.cluster_id,
Groups.cluster_id,
Time.cluster_id,
PollControl.cluster_id,
DoorLock.cluster_id,
Ota.cluster_id,
],
OUTPUT_CLUSTERS: [
Ota.cluster_id,
Time.cluster_id,
],
}
},
}
replacement = {
# The device reports itself as a Router, but it is a battery-powered
# EndDevice. Force the correct node descriptor so ZHA uses proper
# check-in intervals and keeps the device available.
NODE_DESCRIPTOR: XIAOMI_NODE_DESC,
ENDPOINTS: {
1: {
INPUT_CLUSTERS: [
BasicCluster,
XiaomiPowerConfiguration,
Identify.cluster_id,
DeviceTemperature.cluster_id,
Scenes.cluster_id,
Groups.cluster_id,
Time.cluster_id,
PollControl.cluster_id,
AqaraDoorLockS2ProCluster,
Ota.cluster_id,
],
OUTPUT_CLUSTERS: [
Ota.cluster_id,
Time.cluster_id,
],
}
},
}