Merge remote-tracking branch 'upstream/master' into feat/parent_child_updates

kasa/smart/effects.py

@@ -2,8 +2,33 @@
 
 from __future__ import annotations
 
+from abc import ABC, abstractmethod
 from typing import cast
 
+from ..interfaces.lighteffect import LightEffect as LightEffectInterface
+
+
+class SmartLightEffect(LightEffectInterface, ABC):
+    """Abstract interface for smart light effects.
+
+    This interface extends lighteffect interface to add brightness controls.
+    """
+
+    @abstractmethod
+    async def set_brightness(self, brightness: int, *, transition: int | None = None):
+        """Set effect brightness."""
+
+    @property
+    @abstractmethod
+    def brightness(self) -> int:
+        """Return effect brightness."""
+
+    @property
+    @abstractmethod
+    def is_active(self) -> bool:
+        """Return True if effect is active."""
+
+
 EFFECT_AURORA = {
     "custom": 0,
     "id": "TapoStrip_1MClvV18i15Jq3bvJVf0eP",

kasa/smart/modules/__init__.py

@@ -1,5 +1,6 @@
 """Modules for SMART devices."""
 
+from ..effects import SmartLightEffect
 from .alarm import Alarm
 from .autooff import AutoOff
 from .batterysensor import BatterySensor
@@ -54,4 +55,5 @@ __all__ = [
     "WaterleakSensor",
     "ContactSensor",
     "FrostProtection",
+    "SmartLightEffect",
 ]

kasa/smart/modules/brightness.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from ...feature import Feature
-from ..smartmodule import SmartModule
+from ..smartmodule import Module, SmartModule
 
 BRIGHTNESS_MIN = 0
 BRIGHTNESS_MAX = 100
@@ -42,6 +42,12 @@ class Brightness(SmartModule):
     @property
     def brightness(self):
         """Return current brightness."""
+        # If the device supports effects and one is active, use its brightness
+        if (
+            light_effect := self._device.modules.get(Module.SmartLightEffect)
+        ) is not None and light_effect.is_active:
+            return light_effect.brightness
+
         return self.data["brightness"]
 
     async def set_brightness(self, brightness: int, *, transition: int | None = None):
@@ -59,6 +65,13 @@ class Brightness(SmartModule):
 
         if brightness == 0:
             return await self._device.turn_off()
+
+        # If the device supports effects and one is active, we adjust its brightness
+        if (
+            light_effect := self._device.modules.get(Module.SmartLightEffect)
+        ) is not None and light_effect.is_active:
+            return await light_effect.set_brightness(brightness)
+
         return await self.call("set_device_info", {"brightness": brightness})
 
     async def _check_supported(self):

kasa/smart/modules/lighteffect.py

@@ -3,14 +3,16 @@
 from __future__ import annotations
 
 import base64
+import binascii
+import contextlib
 import copy
 from typing import Any
 
-from ...interfaces.lighteffect import LightEffect as LightEffectInterface
-from ..smartmodule import SmartModule
+from ..effects import SmartLightEffect
+from ..smartmodule import Module, SmartModule
 
 
-class LightEffect(SmartModule, LightEffectInterface):
+class LightEffect(SmartModule, SmartLightEffect):
     """Implementation of dynamic light effects."""
 
     REQUIRED_COMPONENT = "light_effect"
@@ -36,8 +38,11 @@ class LightEffect(SmartModule, LightEffectInterface):
                 # If the name has not been edited scene_name will be an empty string
                 effect["scene_name"] = self.AVAILABLE_BULB_EFFECTS[effect["id"]]
             else:
-                # Otherwise it will be b64 encoded
-                effect["scene_name"] = base64.b64decode(effect["scene_name"]).decode()
+                # Otherwise it might be b64 encoded or raw string
+                with contextlib.suppress(binascii.Error):
+                    effect["scene_name"] = base64.b64decode(
+                        effect["scene_name"]
+                    ).decode()
 
         self._effect_state_list = effects
         self._effect_list = [self.LIGHT_EFFECTS_OFF]
@@ -77,6 +82,8 @@ class LightEffect(SmartModule, LightEffectInterface):
     ) -> None:
         """Set an effect for the device.
 
+        Calling this will modify the brightness of the effect on the device.
+
         The device doesn't store an active effect while not enabled so store locally.
         """
         if effect != self.LIGHT_EFFECTS_OFF and effect not in self._scenes_names_to_id:
@@ -90,7 +97,64 @@ class LightEffect(SmartModule, LightEffectInterface):
         if enable:
             effect_id = self._scenes_names_to_id[effect]
             params["id"] = effect_id
-        return await self.call("set_dynamic_light_effect_rule_enable", params)
+
+            # We set the wanted brightness before activating the effect
+            brightness_module = self._device.modules[Module.Brightness]
+            brightness = (
+                brightness if brightness is not None else brightness_module.brightness
+            )
+            await self.set_brightness(brightness, effect_id=effect_id)
+
+        await self.call("set_dynamic_light_effect_rule_enable", params)
+
+    @property
+    def is_active(self) -> bool:
+        """Return True if effect is active."""
+        return bool(self._device._info["dynamic_light_effect_enable"])
+
+    def _get_effect_data(self, effect_id: str | None = None) -> dict[str, Any]:
+        """Return effect data for the *effect_id*.
+
+        If *effect_id* is None, return the data for active effect.
+        """
+        if effect_id is None:
+            effect_id = self.data["current_rule_id"]
+
+        return self._effect_state_list[effect_id]
+
+    @property
+    def brightness(self) -> int:
+        """Return effect brightness."""
+        first_color_status = self._get_effect_data()["color_status_list"][0]
+        brightness = first_color_status[0]
+
+        return brightness
+
+    async def set_brightness(
+        self,
+        brightness: int,
+        *,
+        transition: int | None = None,
+        effect_id: str | None = None,
+    ):
+        """Set effect brightness."""
+        new_effect = self._get_effect_data(effect_id=effect_id).copy()
+
+        def _replace_brightness(data, new_brightness):
+            """Replace brightness.
+
+            The first element is the brightness, the rest are unknown.
+            [[33, 0, 0, 2700], [33, 321, 99, 0], [33, 196, 99, 0], .. ]
+            """
+            return [new_brightness, data[1], data[2], data[3]]
+
+        new_color_status_list = [
+            _replace_brightness(state, brightness)
+            for state in new_effect["color_status_list"]
+        ]
+        new_effect["color_status_list"] = new_color_status_list
+
+        return await self.call("edit_dynamic_light_effect_rule", new_effect)
 
     async def set_custom_effect(
         self,

kasa/smart/modules/lightstripeffect.py

@@ -4,15 +4,14 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from ...interfaces.lighteffect import LightEffect as LightEffectInterface
-from ..effects import EFFECT_MAPPING, EFFECT_NAMES
-from ..smartmodule import SmartModule
+from ..effects import EFFECT_MAPPING, EFFECT_NAMES, SmartLightEffect
+from ..smartmodule import Module, SmartModule
 
 if TYPE_CHECKING:
     from ..smartdevice import SmartDevice
 
 
-class LightStripEffect(SmartModule, LightEffectInterface):
+class LightStripEffect(SmartModule, SmartLightEffect):
     """Implementation of dynamic light effects."""
 
     REQUIRED_COMPONENT = "light_strip_lighting_effect"
@@ -22,6 +21,7 @@ class LightStripEffect(SmartModule, LightEffectInterface):
         effect_list = [self.LIGHT_EFFECTS_OFF]
         effect_list.extend(EFFECT_NAMES)
         self._effect_list = effect_list
+        self._effect_mapping = EFFECT_MAPPING
 
     @property
     def name(self) -> str:
@@ -53,6 +53,28 @@ class LightStripEffect(SmartModule, LightEffectInterface):
             return name
         return self.LIGHT_EFFECTS_OFF
 
+    @property
+    def is_active(self) -> bool:
+        """Return if effect is active."""
+        eff = self.data["lighting_effect"]
+        # softAP has enable=1, but brightness 0 which fails on tests
+        return bool(eff["enable"]) and eff["name"] in self._effect_list
+
+    @property
+    def brightness(self) -> int:
+        """Return effect brightness."""
+        eff = self.data["lighting_effect"]
+        return eff["brightness"]
+
+    async def set_brightness(self, brightness: int, *, transition: int | None = None):
+        """Set effect brightness."""
+        if brightness <= 0:
+            return await self.set_effect(self.LIGHT_EFFECTS_OFF)
+
+        # Need to pass bAdjusted to keep the existing effect running
+        eff = {"brightness": brightness, "bAdjusted": True}
+        return await self.set_custom_effect(eff)
+
     @property
     def effect_list(self) -> list[str]:
         """Return built-in effects list.
@@ -81,16 +103,24 @@ class LightStripEffect(SmartModule, LightEffectInterface):
         :param int brightness: The wanted brightness
         :param int transition: The wanted transition time
         """
+        brightness_module = self._device.modules[Module.Brightness]
         if effect == self.LIGHT_EFFECTS_OFF:
-            effect_dict = dict(self.data["lighting_effect"])
-            effect_dict["enable"] = 0
-        elif effect not in EFFECT_MAPPING:
+            state = self._device.modules[Module.Light].state
+            await self._device.modules[Module.Light].set_state(state)
+            return
+
+        if effect not in self._effect_mapping:
             raise ValueError(f"The effect {effect} is not a built in effect.")
         else:
-            effect_dict = EFFECT_MAPPING[effect]
+            effect_dict = self._effect_mapping[effect]
 
+        # Use explicitly given brightness
         if brightness is not None:
             effect_dict["brightness"] = brightness
+        # Fall back to brightness reported by the brightness module
+        elif brightness_module.brightness:
+            effect_dict["brightness"] = brightness_module.brightness
+
         if transition is not None:
             effect_dict["transition"] = transition
 

kasa/smart/smartchilddevice.py

@@ -64,12 +64,13 @@ class SmartChildDevice(SmartDevice):
     async def _update(self):
         """Update child module info.
 
-        The parent updates our internal info so just update modules with
-        their own queries.
+        Internal implementation to allow patching of public update in the cli
+        or test framework.
         """
         # Hubs attached devices only update via the parent hub
-        if self._parent.device_type == DeviceType.Hub:
+        if self._parent.device_type is DeviceType.Hub:
             return
+
         req: dict[str, Any] = {}
         for module in self.modules.values():
             if mod_query := module.query():