Add post update hook to module and use in smart LightEffect (#921)

Adds a post update hook to modules so they can calculate values and collections once rather than on each property access
This commit is contained in:
Steven B 2024-05-19 10:18:17 +01:00 committed by GitHub
parent 3490a1ef84
commit 9989d0f6ec
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 46 additions and 23 deletions

View File

@ -107,6 +107,18 @@ class Module(ABC):
"""Initialize features after the initial update.
This can be implemented if features depend on module query responses.
It will only be called once per module and will always be called
after *_post_update_hook* has been called for every device module and its
children's modules.
"""
def _post_update_hook(self): # noqa: B027
"""Perform actions after a device update.
This can be implemented if a module needs to perform actions each time
the device has updated like generating collections for property access.
It will be called after every update and will be called prior to
*_initialize_features* on the first update.
"""
def _add_feature(self, feature: Feature):

View File

@ -4,14 +4,11 @@ from __future__ import annotations
import base64
import copy
from typing import TYPE_CHECKING, Any
from typing import Any
from ...interfaces.lighteffect import LightEffect as LightEffectInterface
from ..smartmodule import SmartModule
if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class LightEffect(SmartModule, LightEffectInterface):
"""Implementation of dynamic light effects."""
@ -23,12 +20,13 @@ class LightEffect(SmartModule, LightEffectInterface):
"L2": "Relax",
}
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._scenes_names_to_id: dict[str, str] = {}
_effect: str
_effect_state_list: dict[str, dict[str, Any]]
_effect_list: list[str]
_scenes_names_to_id: dict[str, str]
def _initialize_effects(self) -> dict[str, dict[str, Any]]:
"""Return built-in effects."""
def _post_update_hook(self) -> None:
"""Update internal effect state."""
# 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"]}
@ -40,10 +38,21 @@ class LightEffect(SmartModule, LightEffectInterface):
else:
# Otherwise it will be b64 encoded
effect["scene_name"] = base64.b64decode(effect["scene_name"]).decode()
self._effect_state_list = effects
self._effect_list = [self.LIGHT_EFFECTS_OFF]
self._effect_list.extend([effect["scene_name"] for effect in effects.values()])
self._scenes_names_to_id = {
effect["scene_name"]: effect["id"] for effect in effects.values()
}
return effects
# 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"]:
self._effect = self._effect_state_list[
self._device._info["dynamic_light_effect_id"]
]["scene_name"]
else:
self._effect = self.LIGHT_EFFECTS_OFF
@property
def effect_list(self) -> list[str]:
@ -52,22 +61,12 @@ class LightEffect(SmartModule, LightEffectInterface):
Example:
['Party', 'Relax', ...]
"""
effects = [self.LIGHT_EFFECTS_OFF]
effects.extend(
[effect["scene_name"] for effect in self._initialize_effects().values()]
)
return effects
return self._effect_list
@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
return self._effect
async def set_effect(
self,

View File

@ -41,7 +41,6 @@ class SmartChildDevice(SmartDevice):
"""Create a child device based on device info and component listing."""
child: SmartChildDevice = cls(parent, child_info, child_components)
await child._initialize_modules()
await child._initialize_features()
return child
@property

View File

@ -184,6 +184,13 @@ class SmartDevice(Device):
for info in child_info["child_device_list"]:
self._children[info["device_id"]]._update_internal_state(info)
# Call handle update for modules that want to update internal data
for module in self._modules.values():
module._post_update_hook()
for child in self._children.values():
for child_module in child._modules.values():
child_module._post_update_hook()
# We can first initialize the features after the first update.
# We make here an assumption that every device has at least a single feature.
if not self._features:
@ -332,6 +339,9 @@ class SmartDevice(Device):
for feat in module._module_features.values():
self._add_feature(feat)
for child in self._children.values():
await child._initialize_features()
@property
def is_cloud_connected(self) -> bool:
"""Returns if the device is connected to the cloud."""

View File

@ -48,7 +48,10 @@ async def test_initial_update(dev: SmartDevice, mocker: MockerFixture):
"""Test the initial update cycle."""
# As the fixture data is already initialized, we reset the state for testing
dev._components_raw = None
dev._components = {}
dev._modules = {}
dev._features = {}
dev._children = {}
negotiate = mocker.spy(dev, "_negotiate")
initialize_modules = mocker.spy(dev, "_initialize_modules")