wyze.py

"""Support for the Wyze lock."""
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, Final

from enum import Enum

from zigpy.typing import AddressingMode
from zigpy.profiles import zha
from zigpy.quirks import CustomCluster, CustomDevice
import zigpy.types as t
from zigpy.zcl import foundation
from zigpy.zcl.clusters.closures import DoorLock, LockState
from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster
from zigpy.zcl.clusters.general import (
    Basic,
    Identify,
    Ota,
    PollControl,
    PowerConfiguration,
    Time,
)
from zigpy.zcl.clusters.homeautomation import Diagnostic
from zigpy.zcl.clusters.security import IasZone

from zhaquirks import Bus, LocalDataCluster
from zhaquirks.const import (
    CLUSTER_COMMAND,
    DEVICE_TYPE,
    ENDPOINTS,
    INPUT_CLUSTERS,
    MODELS_INFO,
    OFF,
    ON,
    OUTPUT_CLUSTERS,
    PROFILE_ID,
    ZONE_STATE,
)


YUNDING = "Yunding"
WYZE_CLUSTER_ID = 0xFC00
ZONE_TYPE = 0x0001

_LOGGER = logging.getLogger(__name__)


class DoorLockCluster(CustomCluster, DoorLock):
    """DoorLockCluster cluster."""

    def __init__(self, *args, **kwargs):
        """Init."""
        super().__init__(*args, **kwargs)
        self.endpoint.device.lock_bus.add_listener(self)

    def lock_state(self, locked):
        """Lock state."""
        self._update_attribute(0x0000, locked)
    

class MotionCluster(LocalDataCluster, IasZone):
    """Motion cluster."""

    def __init__(self, *args, **kwargs):
        """Init."""
        super().__init__(*args, **kwargs)
        self.endpoint.device.motion_bus.add_listener(self)
        super()._update_attribute(ZONE_TYPE, IasZone.ZoneType.Contact_Switch)

    def motion_event(self, zone_state):
        """Motion event."""
        super().listener_event(CLUSTER_COMMAND, None, ZONE_STATE, [zone_state])
        _LOGGER.debug("%s - Received motion event message", self.endpoint.device.ieee)


class WyzeStatusArgs(t.List, item_type=t.uint8_t):
    """ Just a list of bytes until we figure out something better """

class Action(Enum):
    APP_LOCK = 1,
    APP_UNLOCK = 2,
    MANUAL_LOCK = 3,
    MANUAL_UNLOCK = 4,
    AUTO_LOCK = 5,
    DOOR_OPEN = 6,
    DOOR_CLOSE = 7

actions = {
    # TODO: add and entry for you lock here
    (t.EUI64.convert("60:a4:23:ff:fe:8b:ba:85"), 1) : {
        # TODO update below with the values in the logs corresponding to your lock for each action 
        (14, 27) : Action.APP_LOCK,
        (14, 28) : Action.APP_UNLOCK,
        (10, 27) : Action.MANUAL_LOCK,
        (10, 28) : Action.MANUAL_UNLOCK,
        (7, 27) : Action.AUTO_LOCK,
        (240, 8) : Action.DOOR_OPEN,
        (240, 11) : Action.DOOR_CLOSE,
    }
}


class WyzeCluster(CustomCluster, ManufacturerSpecificCluster):
    """Wyze manufacturer specific cluster implementation."""

    class ServerCommandDefs(foundation.BaseCommandDefs):
        status_update: Final = foundation.ZCLCommandDef(
            id=0x00, schema=WyzeStatusArgs, direction=foundation.Direction.Client_to_Server, is_manufacturer_specific = True    # pyright: ignore[reportArgumentType]
        )

    cluster_id = WYZE_CLUSTER_ID        # pyright: ignore[reportAssignmentType]
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.seq = 0

    def handle_cluster_request(
        self,
        hdr: foundation.ZCLHeader,
        args: list[Any],
        *,
        dst_addressing: AddressingMode | None = None,
    ):
        """Handle a cluster request."""

        seq = args[8]
        if seq <= self.seq and (self.seq - seq < 50):
            self.warning(f"[Wyze Lock] Ignoring duplicate message, seq: {seq}")
            return
        self.seq = seq

        self.warning(f"[Wyze Lock] Received command: {args}")

        if not (args[0] == 79 and args[1] == 171):
            self.warning(f"[Wyze Lock] Ignoring mesage of unknown type: {(args[0], args[1])}, len: {len(args)}")
            return


        lock_actions = actions.get(self.endpoint.unique_id, None)
        action = lock_actions.get((args[52], args[41]), None) if lock_actions is not None else None
        
        if action is None:
            self.warning(f"Cannot find action for lock {self.endpoint.unique_id} and values {(args[52], args[41])}, update the 'actions' dict.")
            return

        self.warning(f"Door action: {action}")

        if action == Action.APP_UNLOCK:
            self.endpoint.device.lock_bus.listener_event("lock_state", LockState.Unlocked)
        elif action == Action.APP_LOCK:
            self.endpoint.device.lock_bus.listener_event("lock_state", LockState.Locked)
        elif action == Action.MANUAL_UNLOCK:
            self.endpoint.device.lock_bus.listener_event("lock_state", LockState.Unlocked)
        elif action == Action.MANUAL_LOCK:
            self.endpoint.device.lock_bus.listener_event("lock_state", LockState.Locked)
        elif action == Action.AUTO_LOCK:
            self.endpoint.device.lock_bus.listener_event("lock_state", LockState.Locked)
        elif action == Action.DOOR_OPEN:
            self.endpoint.device.motion_bus.listener_event("motion_event", ON)
        elif action == Action.DOOR_CLOSE:
            self.endpoint.device.motion_bus.listener_event("motion_event", OFF)


class WyzeLock(CustomDevice):
    """Wyze lock device."""

    def __init__(self, *args, **kwargs):
        """Init."""
        self.lock_bus = Bus()
        self.motion_bus = Bus()
        super().__init__(*args, **kwargs)

    signature = {
        #  <SimpleDescriptor endpoint=1 profile=260 device_type=10
        # device_version=1
        # input_clusters=[0, 1, 3, 32, 257, 2821, 64512]
                 # "0x0000", "0x0001", "0x0003", "0x0020", "0x0101", "0x0b05", "0xfc00"
        # output_clusters=[10, 25, 64512]>
             #          "0x000a", "0x0019", "0xfc00"
        MODELS_INFO: [(YUNDING, "Ford")],
        ENDPOINTS: {
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.DOOR_LOCK,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    PowerConfiguration.cluster_id,
                    Identify.cluster_id,
                    PollControl.cluster_id,
                    DoorLock.cluster_id,
                    Diagnostic.cluster_id,
                    WYZE_CLUSTER_ID,
                ],
                OUTPUT_CLUSTERS: [Ota.cluster_id, Time.cluster_id, WYZE_CLUSTER_ID],
            }
        },
    }
    replacement = {
        ENDPOINTS: {
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.DOOR_LOCK,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    PowerConfiguration.cluster_id,
                    Identify.cluster_id,
                    PollControl.cluster_id,
                    DoorLockCluster,
                    Diagnostic.cluster_id,
                    WyzeCluster,
                    MotionCluster,
                ],
                OUTPUT_CLUSTERS: [Ota.cluster_id, Time.cluster_id, WyzeCluster],
            }
        }
    }

I had to change the batteries on my locks and noticed for awhile now i'm not getting the locked/unlock events anymore. not sure if it was before or after the battery change. Do i need to change the action values every recalibration? Sending the lock and unlock commands from HA still works, but the status isn't updating for manual or auto lock events.

I had to change the batteries on my locks and noticed for awhile now i'm not getting the locked/unlock events anymore. not sure if it was before or after the battery change. Do i need to change the action values every recalibration? Sending the lock and unlock commands from HA still works, but the status isn't updating for manual or auto lock events.

I changed the batteries and didn't have to change the values, but if you did a recalibration I'm guessing that yes, you'll have to update the action values because calibrations changes them