| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| """Module for light effects.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import base64 | ||
| import copy | ||
| from typing import TYPE_CHECKING, Any | ||
|
|
||
| from ...feature import Feature | ||
| from ..smartmodule import SmartModule | ||
|
|
||
| if TYPE_CHECKING: | ||
| from ..smartdevice import SmartDevice | ||
|
|
||
|
|
||
| class LightEffectModule(SmartModule): | ||
| """Implementation of dynamic light effects.""" | ||
|
|
||
| REQUIRED_COMPONENT = "light_effect" | ||
| QUERY_GETTER_NAME = "get_dynamic_light_effect_rules" | ||
| AVAILABLE_BULB_EFFECTS = { | ||
| "L1": "Party", | ||
| "L2": "Relax", | ||
| } | ||
| LIGHT_EFFECTS_OFF = "Off" | ||
|
|
||
| def __init__(self, device: SmartDevice, module: str): | ||
| super().__init__(device, module) | ||
| self._scenes_names_to_id: dict[str, str] = {} | ||
|
|
||
| def _initialize_features(self): | ||
| """Initialize features.""" | ||
| device = self._device | ||
| self._add_feature( | ||
| Feature( | ||
| device, | ||
| "Light effect", | ||
| container=self, | ||
| attribute_getter="effect", | ||
| attribute_setter="set_effect", | ||
| category=Feature.Category.Config, | ||
| type=Feature.Type.Choice, | ||
| choices_getter="effect_list", | ||
| ) | ||
| ) | ||
|
|
||
| def _initialize_effects(self) -> dict[str, dict[str, Any]]: | ||
| """Return built-in effects.""" | ||
| # Copy the effects so scene name updates do not update the underlying dict. | ||
| effects = copy.deepcopy( | ||
| {effect["id"]: effect for effect in self.data["rule_list"]} | ||
| ) | ||
| for effect in effects.values(): | ||
| if not effect["scene_name"]: | ||
| # 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() | ||
| self._scenes_names_to_id = { | ||
| effect["scene_name"]: effect["id"] for effect in effects.values() | ||
| } | ||
| return effects | ||
|
|
||
| @property | ||
| def effect_list(self) -> list[str] | None: | ||
| """Return built-in effects list. | ||
|
|
||
| Example: | ||
| ['Party', 'Relax', ...] | ||
| """ | ||
| effects = [self.LIGHT_EFFECTS_OFF] | ||
| effects.extend( | ||
| [effect["scene_name"] for effect in self._initialize_effects().values()] | ||
| ) | ||
| return effects | ||
|
|
||
| @property | ||
| def effect(self) -> str: | ||
| """Return effect name.""" | ||
| # get_dynamic_light_effect_rules also has an enable property and current_rule_id | ||
| # property that could be used here as an alternative | ||
| if self._device._info["dynamic_light_effect_enable"]: | ||
| return self._initialize_effects()[ | ||
| self._device._info["dynamic_light_effect_id"] | ||
| ]["scene_name"] | ||
| return self.LIGHT_EFFECTS_OFF | ||
|
|
||
| async def set_effect( | ||
| self, | ||
| effect: str, | ||
| ) -> None: | ||
| """Set an effect for 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: | ||
| raise ValueError( | ||
| f"Cannot set light effect to {effect}, possible values " | ||
| f"are: {self.LIGHT_EFFECTS_OFF} " | ||
| f"{' '.join(self._scenes_names_to_id.keys())}" | ||
| ) | ||
| enable = effect != self.LIGHT_EFFECTS_OFF | ||
| params: dict[str, bool | str] = {"enable": enable} | ||
| 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) | ||
|
|
||
| def query(self) -> dict: | ||
| """Query to execute during the update cycle.""" | ||
| return {self.QUERY_GETTER_NAME: {"start_index": 0}} | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| from __future__ import annotations | ||
|
|
||
| from itertools import chain | ||
| from typing import cast | ||
|
|
||
| import pytest | ||
| from pytest_mock import MockerFixture | ||
|
|
||
| from kasa import Device, Feature | ||
| from kasa.smart.modules import LightEffectModule | ||
| from kasa.tests.device_fixtures import parametrize | ||
|
|
||
| light_effect = parametrize( | ||
| "has light effect", component_filter="light_effect", protocol_filter={"SMART"} | ||
| ) | ||
|
|
||
|
|
||
| @light_effect | ||
| async def test_light_effect(dev: Device, mocker: MockerFixture): | ||
| """Test light effect.""" | ||
| light_effect = cast(LightEffectModule, dev.modules.get("LightEffectModule")) | ||
| assert light_effect | ||
|
|
||
| feature = light_effect._module_features["light_effect"] | ||
| assert feature.type == Feature.Type.Choice | ||
|
|
||
| call = mocker.spy(light_effect, "call") | ||
| assert feature.choices == light_effect.effect_list | ||
| assert feature.choices | ||
| for effect in chain(reversed(feature.choices), feature.choices): | ||
| await light_effect.set_effect(effect) | ||
| enable = effect != LightEffectModule.LIGHT_EFFECTS_OFF | ||
| params: dict[str, bool | str] = {"enable": enable} | ||
| if enable: | ||
| params["id"] = light_effect._scenes_names_to_id[effect] | ||
| call.assert_called_with("set_dynamic_light_effect_rule_enable", params) | ||
| await dev.update() | ||
| assert light_effect.effect == effect | ||
| assert feature.value == effect | ||
|
|
||
| with pytest.raises(ValueError): | ||
| await light_effect.set_effect("foobar") |
| Back | FazBrowse Home | New Git URL |
Uh oh!
There was an error while loading. Please reload this page.