feat: add ZHA quirk for Sunricher SR-ZG9002KR12-Pro remote

Author: niracler

tests/test_sunricher.py

@@ -0,0 +1,154 @@
+"""Tests for Sunricher remote device."""
+
+from unittest import mock
+
+import pytest
+from zigpy.zcl.foundation import ZCLHeader
+
+import zhaquirks
+from zhaquirks.const import BUTTON, COMMAND, PRESS_TYPE
+from zhaquirks.sunricher.remote import ZG9002KR12ProRemoteCluster
+
+zhaquirks.setup()
+
+
+@pytest.mark.parametrize(
+    "button_id, press_type_id, expected_action",
+    [
+        (1, 1, "k1_short_press"),
+        (2, 1, "k2_short_press"),
+    ],
+)
+def test_button_press_events(
+    zigpy_device_from_v2_quirk, button_id, press_type_id, expected_action
+):
+    """Test button press events are correctly generated."""
+
+    device = zigpy_device_from_v2_quirk(
+        manufacturer="Sunricher",
+        model="HK-ZRC-K12&RS-E",
+    )
+
+    cluster = device.endpoints[1].sunricher_remote_cluster
+    listener = mock.MagicMock()
+    cluster.add_listener(listener)
+
+    button_mask = 1 << (button_id - 1)
+    high_byte = (button_mask >> 8) & 0xFF
+    low_byte = button_mask & 0xFF
+
+    args = [0x01, high_byte, low_byte, press_type_id]
+
+    cluster.handle_cluster_request(ZCLHeader(), args)
+
+    assert listener.zha_send_event.call_count == 1
+    assert listener.zha_send_event.call_args[0][0] == expected_action
+
+    event_data = listener.zha_send_event.call_args[0][1]
+    press_type_info = cluster.PRESS_TYPES.get(press_type_id)
+
+    assert event_data[BUTTON] == button_id
+    assert event_data[PRESS_TYPE] == press_type_info.action
+    assert event_data[COMMAND] == "button_press"
+
+
+def test_multiple_buttons_pressed(zigpy_device_from_v2_quirk):
+    """Test multiple buttons pressed at the same time."""
+
+    device = zigpy_device_from_v2_quirk(
+        manufacturer="Sunricher",
+        model="HK-ZRC-K12&RS-E",
+    )
+
+    cluster = device.endpoints[1].sunricher_remote_cluster
+    listener = mock.MagicMock()
+    cluster.add_listener(listener)
+
+    button_mask = 3
+    high_byte = (button_mask >> 8) & 0xFF
+    low_byte = button_mask & 0xFF
+
+    press_type_id = 1
+    args = [0x01, high_byte, low_byte, press_type_id]
+
+    cluster.handle_cluster_request(ZCLHeader(), args)
+
+    assert listener.zha_send_event.call_count == 2
+
+    event_names = [call[0][0] for call in listener.zha_send_event.call_args_list]
+    assert "k1_short_press" in event_names
+    assert "k2_short_press" in event_names
+
+
+@pytest.mark.parametrize(
+    "direction_id, speed, expected_action, expected_direction",
+    [
+        (1, 10, "clockwise_rotation", "clockwise"),
+        (2, 5, "anti_clockwise_rotation", "anti_clockwise"),
+    ],
+)
+def test_knob_rotation_events(
+    zigpy_device_from_v2_quirk, direction_id, speed, expected_action, expected_direction
+):
+    """Test knob rotation events are correctly generated."""
+
+    device = zigpy_device_from_v2_quirk(
+        manufacturer="Sunricher",
+        model="HK-ZRC-K12&RS-E",
+    )
+
+    cluster = device.endpoints[1].sunricher_remote_cluster
+    listener = mock.MagicMock()
+    cluster.add_listener(listener)
+
+    args = [0x03, direction_id, 0x00, speed]
+
+    cluster.handle_cluster_request(ZCLHeader(), args)
+
+    assert listener.zha_send_event.call_count == 1
+    assert listener.zha_send_event.call_args[0][0] == expected_action
+
+    event_data = listener.zha_send_event.call_args[0][1]
+
+    assert event_data[BUTTON] == 9
+    assert event_data[PRESS_TYPE] == expected_action
+    assert event_data["speed"] == speed
+
+
+def test_unknown_message_type(zigpy_device_from_v2_quirk):
+    """Test handling of unknown message types."""
+
+    device = zigpy_device_from_v2_quirk(
+        manufacturer="Sunricher",
+        model="HK-ZRC-K12&RS-E",
+    )
+
+    cluster = device.endpoints[1].sunricher_remote_cluster
+    listener = mock.MagicMock()
+    cluster.add_listener(listener)
+
+    args = [0x99, 0x00, 0x00, 0x00]
+
+    cluster.handle_cluster_request(ZCLHeader(), args)
+
+    assert listener.zha_send_event.call_count == 0
+
+
+def test_generate_device_automation_triggers():
+    """Test generation of device automation triggers."""
+
+    triggers = ZG9002KR12ProRemoteCluster.generate_device_automation_triggers()
+
+    for button in ZG9002KR12ProRemoteCluster.BUTTONS.values():
+        for press_type in ZG9002KR12ProRemoteCluster.PRESS_TYPES.values():
+            trigger_key = (press_type.trigger, button.trigger)
+            assert trigger_key in triggers
+            assert (
+                triggers[trigger_key][COMMAND] == f"{button.action}_{press_type.action}"
+            )
+
+    knob_button = ZG9002KR12ProRemoteCluster.BUTTONS.get(9)
+    for direction in ZG9002KR12ProRemoteCluster.KNOB_DIRECTIONS.values():
+        trigger_key = (direction.trigger, knob_button.trigger)
+        assert trigger_key in triggers
+        assert triggers[trigger_key][COMMAND] == direction.action

zhaquirks/sunricher/__init__.py

@@ -0,0 +1 @@
+"""Quirks for Sunricher devices."""

zhaquirks/sunricher/remote.py

@@ -0,0 +1,173 @@
+"""Sunricher remote device."""
+
+import logging
+from typing import Any, Final, NamedTuple, Optional, Union
+
+from zigpy.quirks.v2 import QuirkBuilder
+import zigpy.types as t
+from zigpy.zcl import foundation
+
+from zhaquirks import CustomCluster
+from zhaquirks.const import (
+    BUTTON,
+    COMMAND,
+    DOUBLE_PRESS,
+    LONG_PRESS,
+    LONG_RELEASE,
+    PRESS_TYPE,
+    SHORT_PRESS,
+    ZHA_SEND_EVENT,
+)
+
+_LOGGER = logging.getLogger(__name__)
+SUNRICHER = "Sunricher"
+
+
+class Button(NamedTuple):
+    """Button class."""
+
+    id: int
+    action: str
+    trigger: str
+
+
+class PressType(NamedTuple):
+    """Button press type."""
+
+    name: str
+    action: str
+    trigger: str = None
+
+
+class ZG9002KR12ProRemoteCluster(CustomCluster):
+    """Sunricher manufacturer specific cluster for remote."""
+
+    cluster_id: Final[t.uint16_t] = 0xFF03
+    name: Final = "Sunricher remote cluster"
+    ep_attribute: Final = "sunricher_remote_cluster"
+
+    # Button mapping
+    BUTTONS: dict[int, Button] = {
+        1: Button(1, "k1", "K1"),
+        2: Button(2, "k2", "K2"),
+        3: Button(3, "k3", "K3"),
+        4: Button(4, "k4", "K4"),
+        5: Button(5, "k5", "K5"),
+        6: Button(6, "k6", "K6"),
+        7: Button(7, "k7", "K7"),
+        8: Button(8, "k8", "K8"),
+        9: Button(9, "knob", "Knob"),
+        11: Button(11, "k9", "K9"),
+        12: Button(12, "k10", "K10"),
+        15: Button(15, "k11", "K11"),
+        16: Button(16, "k12", "K12"),
+    }
+
+    # Press types
+    PRESS_TYPES: dict[int, PressType] = {
+        1: PressType(SHORT_PRESS, "short_press", "Short Press"),
+        2: PressType(DOUBLE_PRESS, "double_press", "Double Press"),
+        3: PressType(LONG_PRESS, "hold", "Hold"),
+        4: PressType(LONG_RELEASE, "hold_released", "Hold Released"),
+    }
+
+    # Knob directions
+    KNOB_DIRECTIONS: dict[int, PressType] = {
+        1: PressType("clockwise", "clockwise_rotation", "Clockwise Rotation"),
+        2: PressType(
+            "anti_clockwise", "anti_clockwise_rotation", "Anti-clockwise Rotation"
+        ),
+    }
+
+    def handle_cluster_request(
+        self,
+        hdr: foundation.ZCLHeader,
+        args: list[Any],
+        *,
+        dst_addressing: Optional[
+            Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
+        ] = None,
+    ):
+        """Handle the cluster command."""
+
+        message_type = args[0]
+        if message_type == 0x01:
+            # Button press event
+            press_type_mask = args[3]
+            button_mask = (args[1] << 8) | args[2]
+
+            press_type = self.PRESS_TYPES.get(press_type_mask)
+            action_buttons = []
+            for i in range(16):
+                if (button_mask >> i) & 1:
+                    button_id = i + 1
+                    button = self.BUTTONS.get(button_id)
+                    action_buttons.append(button)
+
+            _LOGGER.debug(
+                "Button event: action=%s, action_buttons=%s",
+                press_type.action,
+                [b.action for b in action_buttons],
+            )
+
+            for button in action_buttons:
+                action = f"{button.action}_{press_type.action}"
+                event_data = {
+                    BUTTON: button.id,
+                    PRESS_TYPE: press_type.action,
+                    COMMAND: "button_press",
+                }
+                self.listener_event(ZHA_SEND_EVENT, action, event_data)
+
+        elif message_type == 0x03:
+            # Knob rotation event
+            direction_mask = args[1]
+            action_speed = args[3]
+            direction = self.KNOB_DIRECTIONS.get(direction_mask)
+
+            _LOGGER.debug(
+                "Knob event: action=%s, action_speed=%s", direction.action, action_speed
+            )
+
+            event_data = {
+                BUTTON: 9,  # knob
+                PRESS_TYPE: direction.action,
+                "speed": action_speed,
+            }
+            self.listener_event(ZHA_SEND_EVENT, direction.action, event_data)
+
+    @classmethod
+    def generate_device_automation_triggers(cls):
+        """Generate automation triggers based on device buttons and press-types."""
+        triggers = {}
+        # Generate button triggers
+        for button in cls.BUTTONS.values():
+            for press_type in cls.PRESS_TYPES.values():
+                triggers[(press_type.trigger, button.trigger)] = {
+                    COMMAND: f"{button.action}_{press_type.action}"
+                }
+
+        # Generate knob triggers
+        button = cls.BUTTONS.get(9)  # knob
+        if button:
+            for direction in cls.KNOB_DIRECTIONS.values():
+                triggers[(direction.trigger, button.trigger)] = {
+                    COMMAND: direction.action
+                }
+
+        _LOGGER.debug("Generated triggers: %s", triggers)
+        return triggers
+
+
+(
+    QuirkBuilder(SUNRICHER, "HK-ZRC-K12&RS-E")
+    .friendly_name(
+        model="SR-ZG9002KR12-Pro",
+        manufacturer=SUNRICHER,
+    )
+    .replaces(ZG9002KR12ProRemoteCluster, endpoint_id=1)
+    .device_automation_triggers(
+        ZG9002KR12ProRemoteCluster.generate_device_automation_triggers()
+    )
+    .add_to_registry()
+)