feat(core): add Z2MLightController support to existing light controllers

related to #168

RELEASE_NOTES.md

@@ -8,6 +8,7 @@ PRERELEASE_NOTE
 
 ## :pencil2: Features
 
+- Add Zigbee2MQTT Light Controller (`Z2MLightController`). Until now we had an option to listen from MQTT, but light commands will always go through HA Light integration. This new controller allows you to interact directly with Zigbe2MQTT commands to interact with your lights. This means that you can leverage the `hold` actions that Zigbee2MQTT offers with barely no lag and much more smoother than `Light Controller` hold actions. However, it is not as flexible and does not offer as many options as `Light Controller` does. Many of the existing devices now have support to `Z2MLightController`, and you can use it in the `class` as you can now use `LightController` as well. You can read more about it [here](https://BASE_URL/controllerx/others/zigbee2mqtt-light-controller). [ #118, #168 ]
 - Allow passing the delay time (in seconds) to `release_delay` attribute. [ #497 ]
 
 <!--

apps/controllerx/cx_const.py

@@ -92,8 +92,14 @@ class Z2MLight:
     HOLD = "hold"
     HOLD_BRIGHTNESS_UP = "hold_brightness_up"
     HOLD_BRIGHTNESS_DOWN = "hold_brightness_down"
+    HOLD_BRIGHTNESS_TOGGLE = "hold_brightness_toggle"
     HOLD_COLOR_TEMP_UP = "hold_colortemp_up"
     HOLD_COLOR_TEMP_DOWN = "hold_colortemp_down"
+    HOLD_COLOR_TEMP_TOGGLE = "hold_colortemp_toggle"
+    XYCOLOR_FROM_CONTROLLER = "xycolor_from_controller"
+    COLORTEMP_FROM_CONTROLLER = "colortemp_from_controller"
+    BRIGHTNESS_FROM_CONTROLLER_LEVEL = "brightness_from_controller_level"
+    BRIGHTNESS_FROM_CONTROLLER_ANGLE = "brightness_from_controller_angle"
 
 
 class MediaPlayer:

apps/controllerx/cx_core/stepper/__init__.py

@@ -92,3 +92,8 @@ def step(self, value: Number, direction: str) -> StepperOutput:
         None, the loop will stop executing.
         """
         raise NotImplementedError
+
+
+class InvertStepper(Stepper):
+    def step(self, value: Number, direction: str) -> StepperOutput:
+        return StepperOutput(self.apply_sign(value, direction), next_direction=None)

apps/controllerx/cx_core/type/light_controller.py

@@ -646,7 +646,9 @@ async def is_colortemp_supported(self) -> bool:
         return "color_temp" in await self.supported_color_modes
 
     @lru_cache(maxsize=None)
-    def get_stepper(self, attribute: str, steps: Number, mode: str) -> Stepper:
+    def get_stepper(
+        self, attribute: str, steps: Number, mode: str, *, tag: str
+    ) -> Stepper:
         previous_direction = Stepper.invert_direction(self.hold_toggle_direction_init)
         if attribute == LightController.ATTRIBUTE_XY_COLOR:
             return IndexLoopStepper(len(self.color_wheel), previous_direction)
@@ -764,7 +766,7 @@ async def click(
             self.value_attribute,
             attribute,
             direction,
-            self.get_stepper(attribute, steps or self.manual_steps, mode),
+            self.get_stepper(attribute, steps or self.manual_steps, mode, tag="click"),
             "click",
         )
 
@@ -804,7 +806,9 @@ async def _hold(
             f"Attribute value before running the hold action: {self.value_attribute}",
             level="DEBUG",
         )
-        stepper = self.get_stepper(attribute, steps or self.automatic_steps, mode)
+        stepper = self.get_stepper(
+            attribute, steps or self.automatic_steps, mode, tag="hold"
+        )
         if direction == StepperDir.TOGGLE:
             self.log(
                 f"Previous direction: {stepper.previous_direction}",

apps/controllerx/cx_core/type/z2m_light_controller.py

@@ -1,10 +1,13 @@
 import asyncio
 import json
+from functools import lru_cache
 from typing import Any, Dict, List, Optional, Set, Type
 
 from cx_const import PredefinedActionsMapping, StepperDir, Z2MLight
 from cx_core.controller import action
-from cx_core.stepper import MinMax, Stepper
+from cx_core.integration import EventData
+from cx_core.integration.z2m import Z2MIntegration
+from cx_core.stepper import InvertStepper, MinMax
 from cx_core.type_controller import Entity, TypeController
 
 DEFAULT_CLICK_STEPS = 70
@@ -159,13 +162,31 @@ def get_predefined_actions_mapping(self) -> PredefinedActionsMapping:
                     StepperDir.DOWN,
                 ),
             ),
+            Z2MLight.HOLD_BRIGHTNESS_TOGGLE: (
+                self.hold,
+                (
+                    Z2MLightController.ATTRIBUTE_BRIGHTNESS,
+                    StepperDir.TOGGLE,
+                ),
+            ),
             Z2MLight.HOLD_COLOR_TEMP_DOWN: (
                 self.hold,
                 (
                     Z2MLightController.ATTRIBUTE_COLOR_TEMP,
                     StepperDir.DOWN,
                 ),
             ),
+            Z2MLight.HOLD_COLOR_TEMP_TOGGLE: (
+                self.hold,
+                (
+                    Z2MLightController.ATTRIBUTE_COLOR_TEMP,
+                    StepperDir.TOGGLE,
+                ),
+            ),
+            Z2MLight.XYCOLOR_FROM_CONTROLLER: self.xycolor_from_controller,
+            Z2MLight.COLORTEMP_FROM_CONTROLLER: self.colortemp_from_controller,
+            Z2MLight.BRIGHTNESS_FROM_CONTROLLER_LEVEL: self.brightness_from_controller_level,
+            Z2MLight.BRIGHTNESS_FROM_CONTROLLER_ANGLE: self.brightness_from_controller_angle,
         }
 
     async def _mqtt_call(self, payload: Dict[str, Any]) -> None:
@@ -222,27 +243,28 @@ async def _on_min(self, attribute: str) -> None:
     async def on_min(self, attribute: str) -> None:
         await self._on_min(attribute)
 
+    @lru_cache(maxsize=None)
+    def get_stepper(self, attribute: str, steps: float, *, tag: str) -> InvertStepper:
+        previous_direction = StepperDir.DOWN
+        return InvertStepper(self.MIN_MAX_ATTR[attribute], steps, previous_direction)
+
     async def _change_light_state(
         self,
         *,
         attribute: str,
         direction: str,
-        steps: float,
+        stepper: InvertStepper,
         transition: float,
         use_onoff: bool,
         mode: str,
     ) -> None:
-        attribute = self.get_option(attribute, self.ATTRIBUTES_LIST, "`click` action")
-        direction = self.get_option(
-            direction, [StepperDir.UP, StepperDir.DOWN], "`click` action"
-        )
-
         onoff_cmd = (
             "_onoff" if use_onoff and attribute == self.ATTRIBUTE_BRIGHTNESS else ""
         )
+        stepper_output = stepper.step(stepper.steps, direction)
         await self._mqtt_call(
             {
-                f"{attribute}_{mode}{onoff_cmd}": Stepper.apply_sign(steps, direction),
+                f"{attribute}_{mode}{onoff_cmd}": stepper_output.next_value,
                 "transition": transition,
             }
         )
@@ -256,32 +278,56 @@ async def click(
         transition: Optional[float] = None,
         use_onoff: Optional[bool] = None,
     ) -> None:
+        attribute = self.get_option(attribute, self.ATTRIBUTES_LIST, "`click` action")
+        direction = self.get_option(
+            direction, [StepperDir.UP, StepperDir.DOWN], "`click` action"
+        )
+        steps = steps if steps is not None else self.click_steps
+        stepper = self.get_stepper(attribute, steps, tag="click")
         await self._change_light_state(
             attribute=attribute,
             direction=direction,
-            steps=steps if steps is not None else self.click_steps,
+            stepper=stepper,
             transition=transition if transition is not None else self.transition,
             use_onoff=use_onoff if use_onoff is not None else self.use_onoff,
             mode="step",
         )
 
-    @action
-    async def hold(
+    async def _hold(
         self,
         attribute: str,
         direction: str,
         steps: Optional[float] = None,
         use_onoff: Optional[bool] = None,
     ) -> None:
+        attribute = self.get_option(attribute, self.ATTRIBUTES_LIST, "`hold` action")
+        direction = self.get_option(
+            direction,
+            [StepperDir.UP, StepperDir.DOWN, StepperDir.TOGGLE],
+            "`hold` action",
+        )
+        steps = steps if steps is not None else self.hold_steps
+        stepper = self.get_stepper(attribute, steps, tag="hold")
+        direction = stepper.get_direction(steps, direction)
         await self._change_light_state(
             attribute=attribute,
             direction=direction,
-            steps=steps if steps is not None else self.click_steps,
+            stepper=stepper,
             transition=self.transition,
             use_onoff=use_onoff if use_onoff is not None else self.use_onoff,
             mode="move",
         )
 
+    @action
+    async def hold(
+        self,
+        attribute: str,
+        direction: str,
+        steps: Optional[float] = None,
+        use_onoff: Optional[bool] = None,
+    ) -> None:
+        await self._hold(attribute, direction, steps, use_onoff)
+
     @action
     async def release(self) -> None:
         await asyncio.gather(
@@ -290,3 +336,72 @@ async def release(self) -> None:
                 for attribute in self.ATTRIBUTES_LIST
             ]
         )
+
+    @action
+    async def xycolor_from_controller(self, extra: Optional[EventData] = None) -> None:
+        if extra is None:
+            self.log("No event data present", level="WARNING")
+            return
+        if isinstance(self.integration, Z2MIntegration):
+            if "action_color" not in extra:
+                self.log(
+                    "`action_color` is not present in the MQTT payload", level="WARNING"
+                )
+                return
+            xy_color = extra["action_color"]
+            await self._on(color={"x": xy_color["x"], "y": xy_color["y"]})
+
+    @action
+    async def colortemp_from_controller(
+        self, extra: Optional[EventData] = None
+    ) -> None:
+        if extra is None:
+            self.log("No event data present", level="WARNING")
+            return
+        if isinstance(self.integration, Z2MIntegration):
+            if "action_color_temperature" not in extra:
+                self.log(
+                    "`action_color_temperature` is not present in the MQTT payload",
+                    level="WARNING",
+                )
+                return
+            await self._on(color_temp=extra["action_color_temperature"])
+
+    @action
+    async def brightness_from_controller_level(
+        self, extra: Optional[EventData] = None
+    ) -> None:
+        if extra is None:
+            self.log("No event data present", level="WARNING")
+            return
+        if isinstance(self.integration, Z2MIntegration):
+            if "action_level" not in extra:
+                self.log(
+                    "`action_level` is not present in the MQTT payload",
+                    level="WARNING",
+                )
+                return
+            await self._on(brightness=extra["action_level"])
+
+    @action
+    async def brightness_from_controller_angle(
+        self,
+        steps: Optional[float] = None,
+        use_onoff: Optional[bool] = None,
+        extra: Optional[EventData] = None,
+    ) -> None:
+        if extra is None:
+            self.log("No event data present", level="WARNING")
+            return
+        if isinstance(self.integration, Z2MIntegration):
+            if "action_rotation_angle" not in extra:
+                self.log(
+                    "`action_rotation_angle` is not present in the MQTT payload",
+                    level="WARNING",
+                )
+                return
+            angle = extra["action_rotation_angle"]
+            direction = StepperDir.UP if angle > 0 else StepperDir.DOWN
+            await self._hold(
+                self.ATTRIBUTE_BRIGHTNESS, direction, steps=steps, use_onoff=use_onoff
+            )