feat(zha): add specific event parse for devices

related to #97

apps/controllerx/cx_core/controller.py

@@ -215,6 +215,13 @@ def get_zha_actions_mapping(self) -> Optional[TypeActionsMapping]:
         """
         return None
 
+    def get_zha_action(self, command: str, args) -> Optional[str]:
+        """
+        This method can be override for controllers that do not support
+        the standard extraction of the actions on cx_core/integration/zha.py
+        """
+        return None
+
     def get_type_actions_mapping(self) -> TypeActionsMapping:
         return {}
 

apps/controllerx/cx_core/integration/zha.py

@@ -18,13 +18,22 @@ def listen_changes(self, controller_id: str) -> None:
             self.controller, self.callback, "zha_event", device_ieee=controller_id
         )
 
-    async def callback(self, event_name: str, data: dict, kwargs: dict) -> None:
-        action = data["command"]
-        args = data["args"]
-        if type(args) == dict:
+    def get_action(self, command: str, args):
+        if isinstance(args, dict):
             args = args["args"]
         args = list(map(str, args))
-        if not (action == "stop" or action == "release"):
+        action = command
+        if not (command == "stop" or command == "release"):
             if len(args) > 0:
                 action += "_" + "_".join(args)
+        return action
+
+    async def callback(self, event_name: str, data: dict, kwargs: dict) -> None:
+        command = data["command"]
+        args = data["args"]
+        action = self.controller.get_zha_action(command, args)
+        if action is None:
+            # If there is no action extracted from the controller then
+            # we extract with the standard function
+            action = self.get_action(command, args)
         await self.controller.handle_action(action)

apps/controllerx/cx_devices/aqara.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from cx_const import Light, TypeActionsMapping
 from cx_core import LightController
 
@@ -128,6 +129,23 @@ def get_deconz_actions_mapping(self) -> TypeActionsMapping:
             7: Light.CLICK_BRIGHTNESS_UP,
         }
 
+    def get_zha_actions_mapping(self) -> TypeActionsMapping:
+        return {
+            "shake": Light.ON_MIN_BRIGHTNESS,
+            "knock": Light.TOGGLE,
+            "slide": Light.ON_FULL_BRIGHTNESS,
+            "flip180": Light.CLICK_COLOR_UP,
+            "flip90": Light.CLICK_COLOR_DOWN,
+            "rotate_left": Light.CLICK_BRIGHTNESS_DOWN,
+            "rotate_right": Light.CLICK_BRIGHTNESS_UP,
+        }
+
+    def get_zha_action(self, command: str, args) -> Optional[str]:
+        action = command
+        if command == "flip":
+            action = command + str(args["flip_degrees"])
+        return action
+
 
 class WXCJKG11LMLightController(LightController):
     def get_z2m_actions_mapping(self) -> TypeActionsMapping:

docs/others/extract-controller-id.md

@@ -17,6 +17,6 @@ In case of deCONZ, you can go to `Developer Tools > Events` then down the bottom
 
 In case of ZHA, you can go to `Developer Tools > Events` then down the bottom you can subscribe for `zha_event` and start listening. Then press any button and you will see event of the button, you will need to copy the `device_ieee` inside the `data` object. It is a number like the following 00:67:88:56:06:78:9b:3f.
 
-#### MQTT
+### MQTT
 
 In case of using MQTT integration, the `controller` attribute must have the MQTT topic to listen from. It is important that the topic payload contains directly the action name and not a JSON. This means that in case of using the MQTT integration in fvour of z2m, then the topic to listen to must be `zigbee2mqtt/<friendly name>/action` or `zigbee2mqtt/<friendly name>/click`.

tests/cx_devices/aqara_test.py

@@ -0,0 +1,22 @@
+import pytest
+from cx_devices.aqara import MFKZQ01LMLightController
+
+
+@pytest.mark.parametrize(
+    "command, args, expected_action",
+    [
+        ("shake", {}, "shake"),
+        ("knock", {}, "knock"),
+        ("slide", {}, "slide"),
+        ("flip", {"flip_degrees": 90}, "flip90"),
+        ("flip", {"flip_degrees": 180}, "flip180"),
+        ("rotate_left", {}, "rotate_left"),
+        ("rotate_right", {}, "rotate_right"),
+    ],
+)
+def test_zha_action_MFKZQ01LMLightController(command, args, expected_action):
+    sut = MFKZQ01LMLightController()
+
+    action = sut.get_zha_action(command, args)
+
+    assert action == expected_action