blacklight/platypush
1
0
Fork 0
forked from platypush/platypush
Code Issues Pull requests Projects Releases Packages Wiki Activity

Major refactor for the light.hue plugin.

Browse source 
- Added support for lights as native platform entities.
- Improved performance by using the JSON API objects whenever possible
  to interact with the bridge instead of the native Python objects,
  which perform a bunch of lazy API calls under the hood resulting in
  degraded performance.
- Fixed lights animation attributes by setting only the ones actually
  supported by a light.
- Several LINT fixes.
This commit is contained in:
Fabio Manganiello 2022-04-30 01:07:00 +02:00
parent 975d37c562
commit 8d57cf06c2
Signed by: blacklight
GPG key ID: D90FBA7F76362774
2 changed files with 366 additions and 198 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
platypush/plugins/light/__init__.py
View file

@ -1,10 +1,12 @@
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from platypush.plugins import action from platypush.entities import manages
from platypush.plugins.switch import SwitchPlugin from platypush.entities.lights import Light
from platypush.plugins import Plugin, action
class LightPlugin(SwitchPlugin, ABC): @manages(Light)
class LightPlugin(Plugin, ABC):
""" """
Abstract plugin to interface your logic with lights/bulbs. Abstract plugin to interface your logic with lights/bulbs.
""" """
@ -27,5 +29,13 @@ class LightPlugin(SwitchPlugin, ABC):
"""Toggle the light status (on/off)""" """Toggle the light status (on/off)"""
raise NotImplementedError() raise NotImplementedError()
@action
@abstractmethod
def status(self):
"""
Get the current status of the lights.
"""
raise NotImplementedError()
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et:

498
platypush/plugins/light/hue/__init__.py
View file

@ -4,10 +4,15 @@ import time
from enum import Enum from enum import Enum
from threading import Thread, Event from threading import Thread, Event
from typing import List from typing import Iterable, Union, Mapping, Any, Set
from platypush.context import get_bus from platypush.context import get_bus
from platypush.message.event.light import LightAnimationStartedEvent, LightAnimationStoppedEvent from platypush.entities import Entity
from platypush.entities.lights import Light as LightEntity
from platypush.message.event.light import (
LightAnimationStartedEvent,
LightAnimationStoppedEvent,
)
from platypush.plugins import action from platypush.plugins import action
from platypush.plugins.light import LightPlugin from platypush.plugins.light import LightPlugin
from platypush.utils import set_thread_name from platypush.utils import set_thread_name
@ -34,6 +39,7 @@ class LightHuePlugin(LightPlugin):
ANIMATION_CTRL_QUEUE_NAME = 'platypush/light/hue/AnimationCtrl' ANIMATION_CTRL_QUEUE_NAME = 'platypush/light/hue/AnimationCtrl'
_BRIDGE_RECONNECT_SECONDS = 5 _BRIDGE_RECONNECT_SECONDS = 5
_MAX_RECONNECT_TRIES = 5 _MAX_RECONNECT_TRIES = 5
_UNINITIALIZED_BRIDGE_ERR = 'The Hue bridge is not initialized'
class Animation(Enum): class Animation(Enum):
COLOR_TRANSITION = 'color_transition' COLOR_TRANSITION = 'color_transition'
@ -61,32 +67,43 @@ class LightHuePlugin(LightPlugin):
self.bridge_address = bridge self.bridge_address = bridge
self.bridge = None self.bridge = None
self.logger.info('Initializing Hue lights plugin - bridge: "{}"'.format(self.bridge_address)) self.logger.info(
'Initializing Hue lights plugin - bridge: "{}"'.format(self.bridge_address)
)
self.connect() self.connect()
self.lights = [] self.lights = set()
self.groups = [] self.groups = set()
if lights: if lights:
self.lights = lights self.lights = set(lights)
elif groups: elif groups:
self.groups = groups self.groups = set(groups)
self._expand_groups() self.lights.update(self._expand_groups(self.groups))
else: else:
# noinspection PyUnresolvedReferences self.lights = {light['name'] for light in self._get_lights().values()}
self.lights = [light.name for light in self.bridge.lights]
self.animation_thread = None self.animation_thread = None
self.animations = {} self.animations = {}
self._animation_stop = Event() self._animation_stop = Event()
self._init_animations() self._init_animations()
self.logger.info('Configured lights: "{}"'.format(self.lights)) self.logger.info(f'Configured lights: {self.lights}')
def _expand_groups(self): def _expand_groups(self, groups: Iterable[str]) -> Set[str]:
groups = [g for g in self.bridge.groups if g.name in self.groups] lights = set()
for group in groups: light_id_to_name = {
for light in group.lights: light_id: light['name'] for light_id, light in self._get_lights().items()
self.lights += [light.name] }
groups_ = [g for g in self._get_groups().values() if g.get('name') in groups]
for group in groups_:
for light_id in group.get('lights', []):
light_name = light_id_to_name.get(light_id)
if light_name:
lights.add(light_name)
return lights
def _init_animations(self): def _init_animations(self):
self.animations = { self.animations = {
@ -94,10 +111,10 @@ class LightHuePlugin(LightPlugin):
'lights': {}, 'lights': {},
} }
for group in self.bridge.groups: for group_id in self._get_groups():
self.animations['groups'][group.group_id] = None self.animations['groups'][group_id] = None
for light in self.bridge.lights: for light_id in self._get_lights():
self.animations['lights'][light.light_id] = None self.animations['lights'][light_id] = None
@action @action
def connect(self): def connect(self):
@ -110,6 +127,7 @@ class LightHuePlugin(LightPlugin):
# Lazy init # Lazy init
if not self.bridge: if not self.bridge:
from phue import Bridge, PhueRegistrationException from phue import Bridge, PhueRegistrationException
success = False success = False
n_tries = 0 n_tries = 0
@ -119,12 +137,14 @@ class LightHuePlugin(LightPlugin):
self.bridge = Bridge(self.bridge_address) self.bridge = Bridge(self.bridge_address)
success = True success = True
except PhueRegistrationException as e: except PhueRegistrationException as e:
self.logger.warning('Bridge registration error: {}'. self.logger.warning('Bridge registration error: {}'.format(str(e)))
format(str(e)))
if n_tries >= self._MAX_RECONNECT_TRIES: if n_tries >= self._MAX_RECONNECT_TRIES:
self.logger.error(('Bridge registration failed after ' + self.logger.error(
'{} attempts').format(n_tries)) (
'Bridge registration failed after ' + '{} attempts'
).format(n_tries)
)
break break
time.sleep(self._BRIDGE_RECONNECT_SECONDS) time.sleep(self._BRIDGE_RECONNECT_SECONDS)
@ -168,7 +188,7 @@ class LightHuePlugin(LightPlugin):
'id': id, 'id': id,
**scene, **scene,
} }
for id, scene in self.bridge.get_scene().items() for id, scene in self._get_scenes().items()
} }
@action @action
@ -215,7 +235,7 @@ class LightHuePlugin(LightPlugin):
'id': id, 'id': id,
**light, **light,
} }
for id, light in self.bridge.get_light().items() for id, light in self._get_lights().items()
} }
@action @action
@ -273,7 +293,7 @@ class LightHuePlugin(LightPlugin):
'id': id, 'id': id,
**group, **group,
} }
for id, group in self.bridge.get_group().items() for id, group in self._get_groups().items()
} }
@action @action
@ -321,15 +341,22 @@ class LightHuePlugin(LightPlugin):
self.bridge = None self.bridge = None
raise e raise e
assert self.bridge, self._UNINITIALIZED_BRIDGE_ERR
lights = [] lights = []
groups = [] groups = []
if 'lights' in kwargs: if 'lights' in kwargs:
lights = kwargs.pop('lights').split(',').strip() \ lights = (
if isinstance(lights, str) else kwargs.pop('lights') kwargs.pop('lights').split(',').strip()
if isinstance(lights, str)
else kwargs.pop('lights')
)
if 'groups' in kwargs: if 'groups' in kwargs:
groups = kwargs.pop('groups').split(',').strip() \ groups = (
if isinstance(groups, str) else kwargs.pop('groups') kwargs.pop('groups').split(',').strip()
if isinstance(groups, str)
else kwargs.pop('groups')
)
if not lights and not groups: if not lights and not groups:
lights = self.lights lights = self.lights
@ -340,12 +367,13 @@ class LightHuePlugin(LightPlugin):
try: try:
if attr == 'scene': if attr == 'scene':
self.bridge.run_scene(groups[0], kwargs.pop('name')) assert groups, 'No groups specified'
self.bridge.run_scene(list(groups)[0], kwargs.pop('name'))
else: else:
if groups: if groups:
self.bridge.set_group(groups, attr, *args, **kwargs) self.bridge.set_group(list(groups), attr, *args, **kwargs)
if lights: if lights:
self.bridge.set_light(lights, attr, *args, **kwargs) self.bridge.set_light(list(lights), attr, *args, **kwargs)
except Exception as e: except Exception as e:
# Reset bridge connection # Reset bridge connection
self.bridge = None self.bridge = None
@ -375,6 +403,7 @@ class LightHuePlugin(LightPlugin):
""" """
self.connect() self.connect()
assert self.bridge, self._UNINITIALIZED_BRIDGE_ERR
self.bridge.set_light(light, **kwargs) self.bridge.set_light(light, **kwargs)
@action @action
@ -382,7 +411,8 @@ class LightHuePlugin(LightPlugin):
""" """
Set a group (or groups) property. Set a group (or groups) property.
:param group: Group or groups to set. Can be a string representing the group name, a group object, a list of strings, or a list of group objects. :param group: Group or groups to set. It can be a string representing the
group name, a group object, a list of strings, or a list of group objects.
:param kwargs: key-value list of parameters to set. :param kwargs: key-value list of parameters to set.
Example call:: Example call::
@ -400,6 +430,7 @@ class LightHuePlugin(LightPlugin):
""" """
self.connect() self.connect()
assert self.bridge, self._UNINITIALIZED_BRIDGE_ERR
self.bridge.set_group(group, **kwargs) self.bridge.set_group(group, **kwargs)
@action @action
@ -451,15 +482,16 @@ class LightHuePlugin(LightPlugin):
groups_off = [] groups_off = []
if groups: if groups:
all_groups = self.bridge.get_group().values() all_groups = self._get_groups().values()
groups_on = [ groups_on = [
group['name'] for group in all_groups group['name']
for group in all_groups
if group['name'] in groups and group['state']['any_on'] is True if group['name'] in groups and group['state']['any_on'] is True
] ]
groups_off = [ groups_off = [
group['name'] for group in all_groups group['name']
for group in all_groups
if group['name'] in groups and group['state']['any_on'] is False if group['name'] in groups and group['state']['any_on'] is False
] ]
@ -467,15 +499,17 @@ class LightHuePlugin(LightPlugin):
lights = self.lights lights = self.lights
if lights: if lights:
all_lights = self.bridge.get_light().values() all_lights = self._get_lights().values()
lights_on = [ lights_on = [
light['name'] for light in all_lights light['name']
for light in all_lights
if light['name'] in lights and light['state']['on'] is True if light['name'] in lights and light['state']['on'] is True
] ]
lights_off = [ lights_off = [
light['name'] for light in all_lights light['name']
for light in all_lights
if light['name'] in lights and light['state']['on'] is False if light['name'] in lights and light['state']['on'] is False
] ]
@ -499,8 +533,13 @@ class LightHuePlugin(LightPlugin):
groups = [] groups = []
if lights is None: if lights is None:
lights = [] lights = []
return self._exec('bri', int(value) % (self.MAX_BRI + 1), return self._exec(
lights=lights, groups=groups, **kwargs) 'bri',
int(value) % (self.MAX_BRI + 1),
lights=lights,
groups=groups,
**kwargs,
)
@action @action
def sat(self, value, lights=None, groups=None, **kwargs): def sat(self, value, lights=None, groups=None, **kwargs):
@ -516,8 +555,13 @@ class LightHuePlugin(LightPlugin):
groups = [] groups = []
if lights is None: if lights is None:
lights = [] lights = []
return self._exec('sat', int(value) % (self.MAX_SAT + 1), return self._exec(
lights=lights, groups=groups, **kwargs) 'sat',
int(value) % (self.MAX_SAT + 1),
lights=lights,
groups=groups,
**kwargs,
)
@action @action
def hue(self, value, lights=None, groups=None, **kwargs): def hue(self, value, lights=None, groups=None, **kwargs):
@ -533,8 +577,13 @@ class LightHuePlugin(LightPlugin):
groups = [] groups = []
if lights is None: if lights is None:
lights = [] lights = []
return self._exec('hue', int(value) % (self.MAX_HUE + 1), return self._exec(
lights=lights, groups=groups, **kwargs) 'hue',
int(value) % (self.MAX_HUE + 1),
lights=lights,
groups=groups,
**kwargs,
)
@action @action
def xy(self, value, lights=None, groups=None, **kwargs): def xy(self, value, lights=None, groups=None, **kwargs):
@ -584,25 +633,31 @@ class LightHuePlugin(LightPlugin):
lights = [] lights = []
if lights: if lights:
bri = statistics.mean([ bri = statistics.mean(
[
light['state']['bri'] light['state']['bri']
for light in self.bridge.get_light().values() for light in self._get_lights().values()
if light['name'] in lights if light['name'] in lights
]) ]
)
elif groups: elif groups:
bri = statistics.mean([ bri = statistics.mean(
[
group['action']['bri'] group['action']['bri']
for group in self.bridge.get_group().values() for group in self._get_groups().values()
if group['name'] in groups if group['name'] in groups
]) ]
)
else: else:
bri = statistics.mean([ bri = statistics.mean(
[
light['state']['bri'] light['state']['bri']
for light in self.bridge.get_light().values() for light in self._get_lights().values()
if light['name'] in self.lights if light['name'] in self.lights
]) ]
)
delta *= (self.MAX_BRI / 100) delta *= self.MAX_BRI / 100
if bri + delta < 0: if bri + delta < 0:
bri = 0 bri = 0
elif bri + delta > self.MAX_BRI: elif bri + delta > self.MAX_BRI:
@ -628,25 +683,31 @@ class LightHuePlugin(LightPlugin):
lights = [] lights = []
if lights: if lights:
sat = statistics.mean([ sat = statistics.mean(
[
light['state']['sat'] light['state']['sat']
for light in self.bridge.get_light().values() for light in self._get_lights().values()
if light['name'] in lights if light['name'] in lights
]) ]
)
elif groups: elif groups:
sat = statistics.mean([ sat = statistics.mean(
[
group['action']['sat'] group['action']['sat']
for group in self.bridge.get_group().values() for group in self._get_groups().values()
if group['name'] in groups if group['name'] in groups
]) ]
)
else: else:
sat = statistics.mean([ sat = statistics.mean(
[
light['state']['sat'] light['state']['sat']
for light in self.bridge.get_light().values() for light in self._get_lights().values()
if light['name'] in self.lights if light['name'] in self.lights
]) ]
)
delta *= (self.MAX_SAT / 100) delta *= self.MAX_SAT / 100
if sat + delta < 0: if sat + delta < 0:
sat = 0 sat = 0
elif sat + delta > self.MAX_SAT: elif sat + delta > self.MAX_SAT:
@ -672,25 +733,31 @@ class LightHuePlugin(LightPlugin):
lights = [] lights = []
if lights: if lights:
hue = statistics.mean([ hue = statistics.mean(
[
light['state']['hue'] light['state']['hue']
for light in self.bridge.get_light().values() for light in self._get_lights().values()
if light['name'] in lights if light['name'] in lights
]) ]
)
elif groups: elif groups:
hue = statistics.mean([ hue = statistics.mean(
[
group['action']['hue'] group['action']['hue']
for group in self.bridge.get_group().values() for group in self._get_groups().values()
if group['name'] in groups if group['name'] in groups
]) ]
)
else: else:
hue = statistics.mean([ hue = statistics.mean(
[
light['state']['hue'] light['state']['hue']
for light in self.bridge.get_light().values() for light in self._get_lights().values()
if light['name'] in self.lights if light['name'] in self.lights
]) ]
)
delta *= (self.MAX_HUE / 100) delta *= self.MAX_HUE / 100
if hue + delta < 0: if hue + delta < 0:
hue = 0 hue = 0
elif hue + delta > self.MAX_HUE: elif hue + delta > self.MAX_HUE:
@ -734,10 +801,20 @@ class LightHuePlugin(LightPlugin):
self._init_animations() self._init_animations()
@action @action
def animate(self, animation, duration=None, def animate(
hue_range=None, sat_range=None, self,
bri_range=None, lights=None, groups=None, animation,
hue_step=1000, sat_step=2, bri_step=1, transition_seconds=1.0): duration=None,
hue_range=None,
sat_range=None,
bri_range=None,
lights=None,
groups=None,
hue_step=1000,
sat_step=2,
bri_step=1,
transition_seconds=1.0,
):
""" """
Run a lights animation. Run a lights animation.
@ -747,28 +824,33 @@ class LightHuePlugin(LightPlugin):
:param duration: Animation duration in seconds (default: None, i.e. continue until stop) :param duration: Animation duration in seconds (default: None, i.e. continue until stop)
:type duration: float :type duration: float
:param hue_range: If you selected a ``color_transition``, this will specify the hue range of your color ``color_transition``. :param hue_range: If you selected a ``color_transition``, this will
Default: [0, 65535] specify the hue range of your color ``color_transition``. Default: [0, 65535]
:type hue_range: list[int] :type hue_range: list[int]
:param sat_range: If you selected a color ``color_transition``, this will specify the saturation range of your color :param sat_range: If you selected a color ``color_transition``, this
``color_transition``. Default: [0, 255] will specify the saturation range of your color ``color_transition``.
Default: [0, 255]
:type sat_range: list[int] :type sat_range: list[int]
:param bri_range: If you selected a color ``color_transition``, this will specify the brightness range of your color :param bri_range: If you selected a color ``color_transition``, this
``color_transition``. Default: [254, 255] :type bri_range: list[int] will specify the brightness range of your color ``color_transition``.
Default: [254, 255] :type bri_range: list[int]
:param lights: Lights to control (names, IDs or light objects). Default: plugin default lights :param lights: Lights to control (names, IDs or light objects). Default: plugin default lights
:param groups: Groups to control (names, IDs or group objects). Default: plugin default groups :param groups: Groups to control (names, IDs or group objects). Default: plugin default groups
:param hue_step: If you selected a color ``color_transition``, this will specify by how much the color hue will change :param hue_step: If you selected a color ``color_transition``, this
between iterations. Default: 1000 :type hue_step: int will specify by how much the color hue will change between iterations.
Default: 1000 :type hue_step: int
:param sat_step: If you selected a color ``color_transition``, this will specify by how much the saturation will change :param sat_step: If you selected a color ``color_transition``, this
between iterations. Default: 2 :type sat_step: int will specify by how much the saturation will change between iterations.
Default: 2 :type sat_step: int
:param bri_step: If you selected a color ``color_transition``, this will specify by how much the brightness will change :param bri_step: If you selected a color ``color_transition``, this
between iterations. Default: 1 :type bri_step: int will specify by how much the brightness will change between iterations.
Default: 1 :type bri_step: int
:param transition_seconds: Time between two transitions or blinks in seconds. Default: 1.0 :param transition_seconds: Time between two transitions or blinks in seconds. Default: 1.0
:type transition_seconds: float :type transition_seconds: float
@ -776,20 +858,26 @@ class LightHuePlugin(LightPlugin):
self.stop_animation() self.stop_animation()
self._animation_stop.clear() self._animation_stop.clear()
all_lights = self._get_lights()
bri_range = bri_range or [self.MAX_BRI - 1, self.MAX_BRI]
sat_range = sat_range or [0, self.MAX_SAT]
hue_range = hue_range or [0, self.MAX_HUE]
if bri_range is None:
bri_range = [self.MAX_BRI - 1, self.MAX_BRI]
if sat_range is None:
sat_range = [0, self.MAX_SAT]
if hue_range is None:
hue_range = [0, self.MAX_HUE]
if groups: if groups:
groups = [g for g in self.bridge.groups if g.name in groups or g.group_id in groups] groups = {
lights = lights or [] group_id: group
for group in groups: for group_id, group in self._get_groups().items()
lights.extend([light.name for light in group.lights]) if group.get('name') in groups or group_id in groups
}
lights = set(lights or [])
lights.update(self._expand_groups([g['name'] for g in groups.values()]))
elif lights: elif lights:
lights = [light.name for light in self.bridge.lights if light.name in lights or light.light_id in lights] lights = {
light['name']
for light_id, light in all_lights.items()
if light['name'] in lights or int(light_id) in lights
}
else: else:
lights = self.lights lights = self.lights
@ -806,26 +894,50 @@ class LightHuePlugin(LightPlugin):
} }
if groups: if groups:
for group in groups: for group_id in groups:
self.animations['groups'][group.group_id] = info self.animations['groups'][group_id] = info
for light in self.bridge.lights: for light_id, light in all_lights.items():
if light.name in lights: if light['name'] in lights:
self.animations['lights'][light.light_id] = info self.animations['lights'][light_id] = info
def _initialize_light_attrs(lights): def _initialize_light_attrs(lights):
lights_by_name = {
light['name']: light for light in self._get_lights().values()
}
if animation == self.Animation.COLOR_TRANSITION: if animation == self.Animation.COLOR_TRANSITION:
return {light: { return {
'hue': random.randint(hue_range[0], hue_range[1]), light: {
'sat': random.randint(sat_range[0], sat_range[1]), **(
'bri': random.randint(bri_range[0], bri_range[1]), {'hue': random.randint(hue_range[0], hue_range[1])} # type: ignore
} for light in lights} if 'hue' in lights_by_name.get(light, {}).get('state', {})
else {}
),
**(
{'sat': random.randint(sat_range[0], sat_range[1])} # type: ignore
if 'sat' in lights_by_name.get(light, {}).get('state', {})
else {}
),
**(
{'bri': random.randint(bri_range[0], bri_range[1])} # type: ignore
if 'bri' in lights_by_name.get(light, {}).get('state', {})
else {}
),
}
for light in lights
}
elif animation == self.Animation.BLINK: elif animation == self.Animation.BLINK:
return {light: { return {
light: {
'on': True, 'on': True,
'bri': self.MAX_BRI, **({'bri': self.MAX_BRI} if 'bri' in light else {}),
'transitiontime': 0, 'transitiontime': 0,
} for light in lights} }
for light in lights
}
raise AssertionError(f'Unknown animation type: {animation}')
def _next_light_attrs(lights): def _next_light_attrs(lights):
if animation == self.Animation.COLOR_TRANSITION: if animation == self.Animation.COLOR_TRANSITION:
@ -843,15 +955,19 @@ class LightHuePlugin(LightPlugin):
else: else:
continue continue
lights[light][attr] = ((value - attr_range[0] + attr_step) % lights[light][attr] = (
(attr_range[1] - attr_range[0] + 1)) + \ (value - attr_range[0] + attr_step)
attr_range[0] % (attr_range[1] - attr_range[0] + 1)
) + attr_range[0]
elif animation == self.Animation.BLINK: elif animation == self.Animation.BLINK:
lights = {light: { lights = {
'on': False if attrs['on'] else True, light: {
'on': not attrs['on'],
'bri': self.MAX_BRI, 'bri': self.MAX_BRI,
'transitiontime': 0, 'transitiontime': 0,
} for (light, attrs) in lights.items()} }
for (light, attrs) in lights.items()
}
return lights return lights
@ -860,13 +976,23 @@ class LightHuePlugin(LightPlugin):
def _animate_thread(lights): def _animate_thread(lights):
set_thread_name('HueAnimate') set_thread_name('HueAnimate')
get_bus().post(LightAnimationStartedEvent(lights=lights, groups=groups, animation=animation)) get_bus().post(
LightAnimationStartedEvent(
lights=lights,
groups=list((groups or {}).keys()),
animation=animation,
)
)
lights = _initialize_light_attrs(lights) lights = _initialize_light_attrs(lights)
animation_start_time = time.time() animation_start_time = time.time()
stop_animation = False stop_animation = False
while not stop_animation and not (duration and time.time() - animation_start_time > duration): while not stop_animation and not (
duration and time.time() - animation_start_time > duration
):
assert self.bridge, self._UNINITIALIZED_BRIDGE_ERR
try: try:
if animation == self.Animation.COLOR_TRANSITION: if animation == self.Animation.COLOR_TRANSITION:
for (light, attrs) in lights.items(): for (light, attrs) in lights.items():
@ -877,7 +1003,9 @@ class LightHuePlugin(LightPlugin):
self.logger.debug('Setting lights to {}'.format(conf)) self.logger.debug('Setting lights to {}'.format(conf))
if groups: if groups:
self.bridge.set_group([g.name for g in groups], conf) self.bridge.set_group(
[g['name'] for g in groups.values()], conf
)
else: else:
self.bridge.set_light(lights.keys(), conf) self.bridge.set_light(lights.keys(), conf)
@ -891,57 +1019,87 @@ class LightHuePlugin(LightPlugin):
lights = _next_light_attrs(lights) lights = _next_light_attrs(lights)
get_bus().post(LightAnimationStoppedEvent(lights=lights, groups=groups, animation=animation)) get_bus().post(
LightAnimationStoppedEvent(
lights=list(lights.keys()),
groups=list((groups or {}).keys()),
animation=animation,
)
)
self.animation_thread = None self.animation_thread = None
self.animation_thread = Thread(target=_animate_thread, self.animation_thread = Thread(
name='HueAnimate', target=_animate_thread, name='HueAnimate', args=(lights,)
args=(lights,)) )
self.animation_thread.start() self.animation_thread.start()
@property def _get_light_attr(self, light, attr: str):
def switches(self) -> List[dict]: try:
""" return getattr(light, attr, None)
:returns: Implements :meth:`platypush.plugins.switch.SwitchPlugin.switches` and returns the status of the except KeyError:
configured lights. Example: return None
.. code-block:: json def transform_entities(
self, entities: Union[Iterable[Union[dict, Entity]], Mapping[Any, dict]]
) -> Iterable[Entity]:
new_entities = []
if isinstance(entities, dict):
entities = [{'id': id, **e} for id, e in entities.items()]
[ for entity in entities:
{ if isinstance(entity, Entity):
"id": "3", new_entities.append(entity)
"name": "Lightbulb 1", elif isinstance(entity, dict):
"on": true, new_entities.append(
"bri": 254, LightEntity(
"hue": 1532, id=entity['id'],
"sat": 215, name=entity['name'],
"effect": "none", description=entity['type'],
"xy": [ on=entity.get('state', {}).get('on', False),
0.6163, brightness=entity.get('state', {}).get('bri'),
0.3403 saturation=entity.get('state', {}).get('sat'),
], hue=entity.get('state', {}).get('hue'),
"ct": 153, temperature=entity.get('state', {}).get('ct'),
"alert": "none", colormode=entity.get('colormode'),
"colormode": "hs", reachable=entity.get('reachable'),
"reachable": true data={
"type": "Extended color light", 'effect': entity.get('state', {}).get('effect'),
"modelid": "LCT001", 'xy': entity.get('state', {}).get('xy'),
"manufacturername": "Philips", },
"uniqueid": "00:11:22:33:44:55:66:77-88", )
"swversion": "5.105.0.21169" )
}
]
""" return super().transform_entities(new_entities) # type: ignore
return [ def _get_lights(self) -> dict:
{ assert self.bridge, self._UNINITIALIZED_BRIDGE_ERR
'id': id, lights = self.bridge.get_light()
**light.pop('state', {}), self.publish_entities(lights) # type: ignore
**light, return lights
}
for id, light in self.bridge.get_light().items() def _get_groups(self) -> dict:
] assert self.bridge, self._UNINITIALIZED_BRIDGE_ERR
groups = self.bridge.get_group() or {}
return groups
def _get_scenes(self) -> dict:
assert self.bridge, self._UNINITIALIZED_BRIDGE_ERR
scenes = self.bridge.get_scene() or {}
return scenes
@action
def status(self) -> Iterable[LightEntity]:
lights = self.transform_entities(self._get_lights())
for light in lights:
light.id = light.external_id
for attr, value in (light.data or {}).items():
setattr(light, attr, value)
del light.external_id
del light.data
return lights
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et: