Deprecated the light.hue backend

The polling logic has been moved to the `light.hue` plugin itself
instead, so it's no longer required to have both a plugin and a backend
enabled in order to fully manage a Hue bridge.
This commit is contained in:
Fabio Manganiello 2022-05-01 21:55:35 +02:00
parent a5541c33b0
commit 117f92e5b4
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
4 changed files with 48 additions and 89 deletions

View file

@ -1,100 +1,24 @@
from threading import Thread import warnings
from platypush.backend import Backend from platypush.backend import Backend
from platypush.context import get_plugin
from platypush.message.event.light import LightStatusChangeEvent
class LightHueBackend(Backend): class LightHueBackend(Backend):
""" """
This backend will periodically check for the status of your configured **DEPRECATED**
Philips Hue light devices and trigger events when the status of a device
(power, saturation, brightness or hue) changes.
Triggers: The polling logic of this backend has been moved to the ``light.hue`` plugin itself.
* :class:`platypush.message.event.light.LightStatusChangeEvent` when the
status of a lightbulb changes
Requires:
* The :class:`platypush.plugins.light.hue.LightHuePlugin` plugin to be
active and configured.
""" """
_DEFAULT_POLL_SECONDS = 10 def __init__(self, *args, **kwargs):
def __init__(self, poll_seconds=_DEFAULT_POLL_SECONDS, *args, **kwargs):
"""
:param poll_seconds: How often the backend will poll the Hue plugin for
status updates. Default: 10 seconds
:type poll_seconds: float
"""
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
self.poll_seconds = poll_seconds warnings.warn(
'The light.hue backend is deprecated. All of its logic '
@staticmethod 'has been moved to the light.hue plugin itself.'
def _get_lights(): )
plugin = get_plugin('light.hue')
if not plugin:
plugin = get_plugin('light.hue', reload=True)
return plugin.get_lights().output
def _listener(self):
def _thread():
lights = self._get_lights()
while not self.should_stop():
try:
lights_new = self._get_lights()
for light_id, light in lights_new.items():
event_args = {}
state = light.get('state')
prev_state = lights.get(light_id, {}).get('state', {})
if 'on' in state and state.get('on') != prev_state.get('on'):
event_args['on'] = state.get('on')
if 'bri' in state and state.get('bri') != prev_state.get('bri'):
event_args['bri'] = state.get('bri')
if 'sat' in state and state.get('sat') != prev_state.get('sat'):
event_args['sat'] = state.get('sat')
if 'hue' in state and state.get('hue') != prev_state.get('hue'):
event_args['hue'] = state.get('hue')
if 'ct' in state and state.get('ct') != prev_state.get('ct'):
event_args['ct'] = state.get('ct')
if 'xy' in state and state.get('xy') != prev_state.get('xy'):
event_args['xy'] = state.get('xy')
if event_args:
event_args['plugin_name'] = 'light.hue'
event_args['light_id'] = light_id
event_args['light_name'] = light.get('name')
self.bus.post(LightStatusChangeEvent(**event_args))
lights = lights_new
except Exception as e:
self.logger.exception(e)
finally:
self.wait_stop(self.poll_seconds)
return _thread
def run(self): def run(self):
super().run() super().run()
self.logger.info('Starting Hue lights backend')
while not self.should_stop():
try:
poll_thread = Thread(target=self._listener())
poll_thread.start()
poll_thread.join()
except Exception as e:
self.logger.exception(e)
self.wait_stop(self.poll_seconds)
self.logger.info('Stopped Hue lights backend') self.logger.info('Stopped Hue lights backend')

View file

@ -1,7 +1,5 @@
manifest: manifest:
events: events:
platypush.message.event.light.LightStatusChangeEvent: when thestatus of a lightbulb
changes
install: install:
pip: [] pip: []
package: platypush.backend.light.hue package: platypush.backend.light.hue

View file

@ -12,13 +12,14 @@ from platypush.entities.lights import Light as LightEntity
from platypush.message.event.light import ( from platypush.message.event.light import (
LightAnimationStartedEvent, LightAnimationStartedEvent,
LightAnimationStoppedEvent, LightAnimationStoppedEvent,
LightStatusChangeEvent,
) )
from platypush.plugins import action from platypush.plugins import action, RunnablePlugin
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
class LightHuePlugin(LightPlugin): class LightHuePlugin(RunnablePlugin, LightPlugin):
""" """
Philips Hue lights plugin. Philips Hue lights plugin.
@ -30,6 +31,8 @@ class LightHuePlugin(LightPlugin):
- :class:`platypush.message.event.light.LightAnimationStartedEvent` when an animation is started. - :class:`platypush.message.event.light.LightAnimationStartedEvent` when an animation is started.
- :class:`platypush.message.event.light.LightAnimationStoppedEvent` when an animation is stopped. - :class:`platypush.message.event.light.LightAnimationStoppedEvent` when an animation is stopped.
- :class:`platypush.message.event.light.LightStatusChangeEvent` when the status of a lightbulb
changes.
""" """
@ -53,7 +56,7 @@ class LightHuePlugin(LightPlugin):
elif isinstance(other, self.__class__): elif isinstance(other, self.__class__):
return self == other return self == other
def __init__(self, bridge, lights=None, groups=None): def __init__(self, bridge, lights=None, groups=None, poll_seconds: float = 20.0):
""" """
:param bridge: Bridge address or hostname :param bridge: Bridge address or hostname
:type bridge: str :type bridge: str
@ -63,6 +66,9 @@ class LightHuePlugin(LightPlugin):
:param groups Default groups to be controlled (default: all) :param groups Default groups to be controlled (default: all)
:type groups: list[str] :type groups: list[str]
:param poll_seconds: How often the plugin should check the bridge for light
updates (default: 20 seconds).
""" """
super().__init__() super().__init__()
@ -76,6 +82,7 @@ class LightHuePlugin(LightPlugin):
self.connect() self.connect()
self.lights = set() self.lights = set()
self.groups = set() self.groups = set()
self.poll_seconds = poll_seconds
self._cached_lights = {} self._cached_lights = {}
if lights: if lights:
@ -1175,5 +1182,33 @@ class LightHuePlugin(LightPlugin):
return lights return lights
def main(self):
lights_prev = self._get_lights() # Initialize the lights
while not self.should_stop():
try:
lights_new = self._get_lights()
for light_id, light in lights_new.items():
event_args = {}
new_state = light.get('state', {})
prev_state = lights_prev.get(light_id, {}).get('state', {})
for attr in ['on', 'bri', 'sat', 'hue', 'ct', 'xy']:
if attr in new_state and new_state.get(attr) != prev_state.get(
attr
):
event_args[attr] = new_state.get(attr)
if event_args:
event_args['plugin_name'] = 'light.hue'
event_args['light_id'] = light_id
event_args['light_name'] = light.get('name')
get_bus().post(LightStatusChangeEvent(**event_args))
self.publish_entities([{'id': light_id, **light}]) # type: ignore
lights_prev = lights_new
finally:
self.wait_stop(self.poll_seconds)
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et:

View file

@ -4,6 +4,8 @@ manifest:
started. started.
platypush.message.event.light.LightAnimationStoppedEvent: when an animation is platypush.message.event.light.LightAnimationStoppedEvent: when an animation is
stopped. stopped.
platypush.message.event.light.LightStatusChangeEvent: when the status of a
lightbulb changes.
install: install:
pip: pip:
- phue - phue