Source code for platypush.plugins.light.hue

import random
import statistics
import time

from enum import Enum
from threading import Thread, Event
from typing import List

from platypush.context import get_bus
from platypush.message.event.light import LightAnimationStartedEvent, LightAnimationStoppedEvent
from platypush.plugins import action
from platypush.plugins.light import LightPlugin
from platypush.utils import set_thread_name


[docs]class LightHuePlugin(LightPlugin): """ Philips Hue lights plugin. Requires: * **phue** (``pip install phue``) Triggers: - :class:`platypush.message.event.light.LightAnimationStartedEvent` when an animation is started. - :class:`platypush.message.event.light.LightAnimationStoppedEvent` when an animation is stopped. """ MAX_BRI = 255 MAX_SAT = 255 MAX_HUE = 65535 ANIMATION_CTRL_QUEUE_NAME = 'platypush/light/hue/AnimationCtrl' _BRIDGE_RECONNECT_SECONDS = 5 _MAX_RECONNECT_TRIES = 5
[docs] class Animation(Enum): COLOR_TRANSITION = 'color_transition' BLINK = 'blink' def __eq__(self, other): if isinstance(other, str): return self.value == other elif isinstance(other, self.__class__): return self == other
[docs] def __init__(self, bridge, lights=None, groups=None): """ :param bridge: Bridge address or hostname :type bridge: str :param lights: Default lights to be controlled (default: all) :type lights: list[str] :param groups Default groups to be controlled (default: all) :type groups: list[str] """ super().__init__() self.bridge_address = bridge self.bridge = None self.logger.info('Initializing Hue lights plugin - bridge: "{}"'.format(self.bridge_address)) self.connect() self.lights = [] self.groups = [] if lights: self.lights = lights elif groups: self.groups = groups self._expand_groups() else: # noinspection PyUnresolvedReferences self.lights = [light.name for light in self.bridge.lights] self.animation_thread = None self.animations = {} self._animation_stop = Event() self._init_animations() self.logger.info('Configured lights: "{}"'.format(self.lights))
def _expand_groups(self): groups = [g for g in self.bridge.groups if g.name in self.groups] for group in groups: for light in group.lights: self.lights += [light.name] def _init_animations(self): self.animations = { 'groups': {}, 'lights': {}, } for group in self.bridge.groups: self.animations['groups'][group.group_id] = None for light in self.bridge.lights: self.animations['lights'][light.light_id] = None
[docs] @action def connect(self): """ Connect to the configured Hue bridge. If the device hasn't been paired yet, uncomment the ``.connect()`` and ``.get_api()`` lines and retry after clicking the pairing button on your bridge. """ # Lazy init if not self.bridge: from phue import Bridge, PhueRegistrationException success = False n_tries = 0 while not success: try: n_tries += 1 self.bridge = Bridge(self.bridge_address) success = True except PhueRegistrationException as e: self.logger.warning('Bridge registration error: {}'. format(str(e))) if n_tries >= self._MAX_RECONNECT_TRIES: self.logger.error(('Bridge registration failed after ' + '{} attempts').format(n_tries)) break time.sleep(self._BRIDGE_RECONNECT_SECONDS) self.logger.info('Bridge connected') self.get_scenes() else: self.logger.info('Bridge already connected')
[docs] @action def get_scenes(self): """ Get the available scenes on the devices. :returns: The scenes configured on the bridge. Example output:: { "scene-id-1": { "name": "Scene 1", "lights": [ "1", "3" ], "owner": "owner-id", "recycle": true, "locked": false, "appdata": {}, "picture": "", "lastupdated": "2018-06-01T00:00:00", "version": 1 } } """ return { id: { 'id': id, **scene, } for id, scene in self.bridge.get_scene().items() }
[docs] @action def get_lights(self): """ Get the configured lights. :returns: List of available lights as id->dict. Example:: { "1": { "state": { "on": true, "bri": 254, "hue": 1532, "sat": 215, "effect": "none", "xy": [ 0.6163, 0.3403 ], "ct": 153, "alert": "none", "colormode": "hs", "reachable": true }, "type": "Extended color light", "name": "Lightbulb 1", "modelid": "LCT001", "manufacturername": "Philips", "uniqueid": "00:11:22:33:44:55:66:77-88", "swversion": "5.105.0.21169" } } """ return { id: { 'id': id, **light, } for id, light in self.bridge.get_light().items() }
[docs] @action def get_groups(self): """ Get the list of configured light groups. :returns: List of configured light groups as id->dict. Example:: { "1": { "name": "Living Room", "lights": [ "16", "13", "12", "11", "10", "9", "1", "3" ], "type": "Room", "state": { "all_on": true, "any_on": true }, "class": "Living room", "action": { "on": true, "bri": 241, "hue": 37947, "sat": 221, "effect": "none", "xy": [ 0.2844, 0.2609 ], "ct": 153, "alert": "none", "colormode": "hs" } } } """ return { id: { 'id': id, **group, } for id, group in self.bridge.get_group().items() }
[docs] @action def get_animations(self): """ Get the list of running light animations. :returns: dict. Structure:: { "groups": { "id_1": { "type": "color_transition", "hue_range": [0,65535], "sat_range": [0,255], "bri_range": [0,255], "hue_step": 10, "sat_step": 10, "bri_step": 2, "transition_seconds": 2 } }, "lights": { "id_1": {} } } """ return self.animations
def _exec(self, attr, *args, **kwargs): try: self.connect() self.stop_animation() except Exception as e: # Reset bridge connection self.bridge = None raise e lights = [] groups = [] if 'lights' in kwargs: lights = kwargs.pop('lights').split(',').strip() \ if isinstance(lights, str) else kwargs.pop('lights') if 'groups' in kwargs: groups = kwargs.pop('groups').split(',').strip() \ if isinstance(groups, str) else kwargs.pop('groups') if not lights and not groups: lights = self.lights groups = self.groups if not self.bridge: self.connect() try: if attr == 'scene': self.bridge.run_scene(groups[0], kwargs.pop('name')) else: if groups: self.bridge.set_group(groups, attr, *args, **kwargs) if lights: self.bridge.set_light(lights, attr, *args, **kwargs) except Exception as e: # Reset bridge connection self.bridge = None raise e
[docs] @action def set_light(self, light, **kwargs): """ Set a light (or lights) property. :param light: Light or lights to set. Can be a string representing the light name, a light object, a list of string, or a list of light objects. :param kwargs: key-value list of parameters to set. Example call:: { "type": "request", "target": "hostname", "action": "light.hue.set_light", "args": { "light": "Bulb 1", "sat": 255 } } """ self.connect() self.bridge.set_light(light, **kwargs)
[docs] @action def set_group(self, group, **kwargs): """ 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 kwargs: key-value list of parameters to set. Example call:: { "type": "request", "target": "hostname", "action": "light.hue.set_group", "args": { "light": "Living Room", "sat": 255 } } """ self.connect() self.bridge.set_group(group, **kwargs)
[docs] @action def on(self, lights=None, groups=None, **kwargs): """ Turn lights/groups on. :param lights: Lights to turn on (names or light objects). Default: plugin default lights :param groups: Groups to turn on (names or group objects). Default: plugin default groups """ if groups is None: groups = [] if lights is None: lights = [] return self._exec('on', True, lights=lights, groups=groups, **kwargs)
[docs] @action def off(self, lights=None, groups=None, **kwargs): """ Turn lights/groups off. :param lights: Lights to turn off (names or light objects). Default: plugin default lights :param groups: Groups to turn off (names or group objects). Default: plugin default groups """ if groups is None: groups = [] if lights is None: lights = [] return self._exec('on', False, lights=lights, groups=groups, **kwargs)
[docs] @action def toggle(self, lights=None, groups=None, **kwargs): """ Toggle lights/groups on/off. :param lights: Lights to turn off (names or light objects). Default: plugin default lights :param groups: Groups to turn off (names or group objects). Default: plugin default groups """ if groups is None: groups = [] if lights is None: lights = [] lights_on = [] lights_off = [] groups_on = [] groups_off = [] if groups: all_groups = self.bridge.get_group().values() groups_on = [ group['name'] for group in all_groups if group['name'] in groups and group['state']['any_on'] is True ] groups_off = [ group['name'] for group in all_groups if group['name'] in groups and group['state']['any_on'] is False ] if not groups and not lights: lights = self.lights if lights: all_lights = self.bridge.get_light().values() lights_on = [ light['name'] for light in all_lights if light['name'] in lights and light['state']['on'] is True ] lights_off = [ light['name'] for light in all_lights if light['name'] in lights and light['state']['on'] is False ] if lights_on or groups_on: self._exec('on', False, lights=lights_on, groups=groups_on, **kwargs) if lights_off or groups_off: self._exec('on', True, lights=lights_off, groups=groups_off, **kwargs)
[docs] @action def bri(self, value, lights=None, groups=None, **kwargs): """ Set lights/groups brightness. :param lights: Lights to control (names or light objects). Default: plugin default lights :param groups: Groups to control (names or group objects). Default: plugin default groups :param value: Brightness value (range: 0-255) """ if groups is None: groups = [] if lights is None: lights = [] return self._exec('bri', int(value) % (self.MAX_BRI + 1), lights=lights, groups=groups, **kwargs)
[docs] @action def sat(self, value, lights=None, groups=None, **kwargs): """ Set lights/groups saturation. :param lights: Lights to control (names or light objects). Default: plugin default lights :param groups: Groups to control (names or group objects). Default: plugin default groups :param value: Saturation value (range: 0-255) """ if groups is None: groups = [] if lights is None: lights = [] return self._exec('sat', int(value) % (self.MAX_SAT + 1), lights=lights, groups=groups, **kwargs)
[docs] @action def hue(self, value, lights=None, groups=None, **kwargs): """ Set lights/groups color hue. :param lights: Lights to control (names or light objects). Default: plugin default lights :param groups: Groups to control (names or group objects). Default: plugin default groups :param value: Hue value (range: 0-65535) """ if groups is None: groups = [] if lights is None: lights = [] return self._exec('hue', int(value) % (self.MAX_HUE + 1), lights=lights, groups=groups, **kwargs)
[docs] @action def xy(self, value, lights=None, groups=None, **kwargs): """ Set lights/groups XY colors. :param value: xY value :type value: list[float] containing the two values :param lights: List of lights. :param groups: List of groups. """ if groups is None: groups = [] if lights is None: lights = [] return self._exec('xy', value, lights=lights, groups=groups, **kwargs)
[docs] @action def ct(self, value, lights=None, groups=None, **kwargs): """ Set lights/groups color temperature. :param value: Temperature value (range: 0-255) :type value: int :param lights: List of lights. :param groups: List of groups. """ if groups is None: groups = [] if lights is None: lights = [] return self._exec('ct', value, lights=lights, groups=groups, **kwargs)
[docs] @action def delta_bri(self, delta, lights=None, groups=None, **kwargs): """ Change lights/groups brightness by a delta [-100, 100] compared to the current state. :param lights: Lights to control (names or light objects). Default: plugin default lights :param groups: Groups to control (names or group objects). Default: plugin default groups :param delta: Brightness delta value (range: -100, 100) """ if groups is None: groups = [] if lights is None: lights = [] if lights: bri = statistics.mean([ light['state']['bri'] for light in self.bridge.get_light().values() if light['name'] in lights ]) elif groups: bri = statistics.mean([ group['action']['bri'] for group in self.bridge.get_group().values() if group['name'] in groups ]) else: bri = statistics.mean([ light['state']['bri'] for light in self.bridge.get_light().values() if light['name'] in self.lights ]) delta *= (self.MAX_BRI / 100) if bri + delta < 0: bri = 0 elif bri + delta > self.MAX_BRI: bri = self.MAX_BRI else: bri += delta return self._exec('bri', int(bri), lights=lights, groups=groups, **kwargs)
[docs] @action def delta_sat(self, delta, lights=None, groups=None, **kwargs): """ Change lights/groups saturation by a delta [-100, 100] compared to the current state. :param lights: Lights to control (names or light objects). Default: plugin default lights :param groups: Groups to control (names or group objects). Default: plugin default groups :param delta: Saturation delta value (range: -100, 100) """ if groups is None: groups = [] if lights is None: lights = [] if lights: sat = statistics.mean([ light['state']['sat'] for light in self.bridge.get_light().values() if light['name'] in lights ]) elif groups: sat = statistics.mean([ group['action']['sat'] for group in self.bridge.get_group().values() if group['name'] in groups ]) else: sat = statistics.mean([ light['state']['sat'] for light in self.bridge.get_light().values() if light['name'] in self.lights ]) delta *= (self.MAX_SAT / 100) if sat + delta < 0: sat = 0 elif sat + delta > self.MAX_SAT: sat = self.MAX_SAT else: sat += delta return self._exec('sat', int(sat), lights=lights, groups=groups, **kwargs)
[docs] @action def delta_hue(self, delta, lights=None, groups=None, **kwargs): """ Change lights/groups hue by a delta [-100, 100] compared to the current state. :param lights: Lights to control (names or light objects). Default: plugin default lights :param groups: Groups to control (names or group objects). Default: plugin default groups :param delta: Hue delta value (range: -100, 100) """ if groups is None: groups = [] if lights is None: lights = [] if lights: hue = statistics.mean([ light['state']['hue'] for light in self.bridge.get_light().values() if light['name'] in lights ]) elif groups: hue = statistics.mean([ group['action']['hue'] for group in self.bridge.get_group().values() if group['name'] in groups ]) else: hue = statistics.mean([ light['state']['hue'] for light in self.bridge.get_light().values() if light['name'] in self.lights ]) delta *= (self.MAX_HUE / 100) if hue + delta < 0: hue = 0 elif hue + delta > self.MAX_HUE: hue = self.MAX_HUE else: hue += delta return self._exec('hue', int(hue), lights=lights, groups=groups, **kwargs)
[docs] @action def scene(self, name, lights=None, groups=None, **kwargs): """ Set a scene by name. :param lights: Lights to control (names or light objects). Default: plugin default lights :param groups: Groups to control (names or group objects). Default: plugin default groups :param name: Name of the scene """ if groups is None: groups = [] if lights is None: lights = [] return self._exec('scene', name=name, lights=lights, groups=groups, **kwargs)
[docs] @action def is_animation_running(self): """ :returns: True if there is an animation running, false otherwise. """ return self.animation_thread is not None and self.animation_thread.is_alive()
[docs] @action def stop_animation(self): """ Stop a running animation. """ if self.is_animation_running(): self._animation_stop.set() self._init_animations()
[docs] @action def animate(self, animation, 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. :param animation: Animation name. Supported types: **color_transition** and **blink** :type animation: str :param duration: Animation duration in seconds (default: None, i.e. continue until stop) :type duration: float :param hue_range: If you selected a ``color_transition``, this will specify the hue range of your color ``color_transition``. Default: [0, 65535] :type hue_range: list[int] :param sat_range: If you selected a color ``color_transition``, this will specify the saturation range of your color ``color_transition``. Default: [0, 255] :type sat_range: list[int] :param bri_range: If you selected a color ``color_transition``, this 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 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 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 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 between iterations. Default: 1 :type bri_step: int :param transition_seconds: Time between two transitions or blinks in seconds. Default: 1.0 :type transition_seconds: float """ self.stop_animation() self._animation_stop.clear() 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: groups = [g for g in self.bridge.groups if g.name in groups or g.group_id in groups] lights = lights or [] for group in groups: lights.extend([light.name for light in group.lights]) elif lights: lights = [light.name for light in self.bridge.lights if light.name in lights or light.light_id in lights] else: lights = self.lights info = { 'type': animation, 'duration': duration, 'hue_range': hue_range, 'sat_range': sat_range, 'bri_range': bri_range, 'hue_step': hue_step, 'sat_step': sat_step, 'bri_step': bri_step, 'transition_seconds': transition_seconds, } if groups: for group in groups: self.animations['groups'][group.group_id] = info for light in self.bridge.lights: if light.name in lights: self.animations['lights'][light.light_id] = info def _initialize_light_attrs(lights): if animation == self.Animation.COLOR_TRANSITION: return {light: { 'hue': random.randint(hue_range[0], hue_range[1]), 'sat': random.randint(sat_range[0], sat_range[1]), 'bri': random.randint(bri_range[0], bri_range[1]), } for light in lights} elif animation == self.Animation.BLINK: return {light: { 'on': True, 'bri': self.MAX_BRI, 'transitiontime': 0, } for light in lights} def _next_light_attrs(lights): if animation == self.Animation.COLOR_TRANSITION: for (light, attrs) in lights.items(): for (attr, value) in attrs.items(): if attr == 'hue': attr_range = hue_range attr_step = hue_step elif attr == 'bri': attr_range = bri_range attr_step = bri_step elif attr == 'sat': attr_range = sat_range attr_step = sat_step else: continue lights[light][attr] = ((value - attr_range[0] + attr_step) % (attr_range[1] - attr_range[0] + 1)) + \ attr_range[0] elif animation == self.Animation.BLINK: lights = {light: { 'on': False if attrs['on'] else True, 'bri': self.MAX_BRI, 'transitiontime': 0, } for (light, attrs) in lights.items()} return lights def _should_stop(): return self._animation_stop.is_set() def _animate_thread(lights): set_thread_name('HueAnimate') get_bus().post(LightAnimationStartedEvent(lights=lights, groups=groups, animation=animation)) lights = _initialize_light_attrs(lights) animation_start_time = time.time() stop_animation = False while not stop_animation and not (duration and time.time() - animation_start_time > duration): try: if animation == self.Animation.COLOR_TRANSITION: for (light, attrs) in lights.items(): self.logger.debug('Setting {} to {}'.format(light, attrs)) self.bridge.set_light(light, attrs) elif animation == self.Animation.BLINK: conf = lights[list(lights.keys())[0]] self.logger.debug('Setting lights to {}'.format(conf)) if groups: self.bridge.set_group([g.name for g in groups], conf) else: self.bridge.set_light(lights.keys(), conf) if transition_seconds: time.sleep(transition_seconds) stop_animation = _should_stop() except Exception as e: self.logger.warning(e) time.sleep(2) lights = _next_light_attrs(lights) get_bus().post(LightAnimationStoppedEvent(lights=lights, groups=groups, animation=animation)) self.animation_thread = None self.animation_thread = Thread(target=_animate_thread, name='HueAnimate', args=(lights,)) self.animation_thread.start()
@property def switches(self) -> List[dict]: """ :returns: Implements :meth:`platypush.plugins.switch.SwitchPlugin.switches` and returns the status of the configured lights. Example: .. code-block:: json [ { "id": "3", "name": "Lightbulb 1", "on": true, "bri": 254, "hue": 1532, "sat": 215, "effect": "none", "xy": [ 0.6163, 0.3403 ], "ct": 153, "alert": "none", "colormode": "hs", "reachable": true "type": "Extended color light", "modelid": "LCT001", "manufacturername": "Philips", "uniqueid": "00:11:22:33:44:55:66:77-88", "swversion": "5.105.0.21169" } ] """ return [ { 'id': id, **light.pop('state', {}), **light, } for id, light in self.bridge.get_light().items() ]
# vim:sw=4:ts=4:et: