import colorsys
import json
import os
import threading
from typing import (
TYPE_CHECKING,
Collection,
Dict,
Iterable,
Mapping,
Optional,
Set,
Union,
)
if TYPE_CHECKING:
import tinytuya
from platypush.config import Config
from platypush.context import get_bus
from platypush.entities import Entity, LightEntityManager
from platypush.entities.lights import Light as LightEntity
from platypush.message.event.light import LightStatusChangeEvent
from platypush.plugins import RunnablePlugin, action
[docs]
class LightTuyaPlugin(RunnablePlugin, LightEntityManager):
"""
Controls Tuya-based Wi-Fi light bulbs (Smart Life / Tuya Smart compatible
devices) directly over the LAN, with no cloud dependency, through
`tinytuya <https://github.com/jasonacox/tinytuya>`_.
Each device needs its IP address, device ID and local key. The easiest
way to collect them is through the tinytuya wizard::
$ python -m tinytuya wizard
which requires the API credentials of a project created on the `Tuya IoT
platform <https://iot.tuya.com>`_ and linked to your Smart Life/Tuya
Smart app account. It generates a ``devices.json`` file that can be
passed directly to this plugin through the ``devices_file``
configuration option. The cloud credentials are only needed for the
initial keys extraction - all the control operations are performed
locally.
"""
MAX_BRI = 255
MIN_TEMP = 153
MAX_TEMP = 500
MAX_HUE = 360
MAX_SAT = 100
_DEFAULT_DEVICES_FILES = (
os.path.join(
os.path.expanduser('~'), '.config', 'platypush', 'tuya', 'devices.json'
),
)
[docs]
def __init__(
self,
devices: Union[Iterable[dict], Mapping[str, dict], None] = None,
groups: Optional[Mapping[str, Iterable[str]]] = None,
devices_file: Optional[str] = None,
poll_interval: Optional[float] = 20.0,
connect_timeout: float = 5.0,
**kwargs,
):
"""
:param devices: Static list of devices. It can either be a list of
dictionaries in the format
``{"name": "...", "id": "...", "ip": "...", "key": "...",
"version": 3.x}`` (the same format generated by the tinytuya
wizard), or a ``name -> device info`` mapping. If not specified,
the devices will be loaded from ``devices_file`` (if provided and
it exists), or from the first file that exists among the default
locations (``~/.config/platypush/devices.json``).
:param groups: Optional ``group name -> [light names/IDs]`` mapping,
used to control multiple lights at once.
:param devices_file: Path to a ``devices.json`` file generated by the
tinytuya wizard.
:param poll_interval: How often the plugin should poll the devices
for status updates and emit
:class:`platypush.message.event.light.LightStatusChangeEvent`
events (default: 20 seconds). Set to 0/None to disable polling.
:param connect_timeout: Timeout for the connection to the devices
(default: 5 seconds).
"""
super().__init__(**kwargs)
self.poll_interval = poll_interval
self.connect_timeout = connect_timeout
self._state_refresh_delay = 0.75
self._state_refresh_retries = 4
self.groups: Dict[str, Set[str]] = {
name: set(lights) for name, lights in (groups or {}).items()
}
self._dev_info: Dict[str, dict] = {}
self._name_to_id: Dict[str, str] = {}
self._ip_to_id: Dict[str, str] = {}
self._bulbs: Dict[str, 'tinytuya.BulbDevice'] = {}
self._bulb_locks: Dict[str, threading.Lock] = {}
self._cached_dps: Dict[str, dict] = {}
self._init_devices(devices=devices, devices_file=devices_file)
self.logger.info(
'Initialized Tuya light plugin - devices: %s',
[d.get('name') for d in self._dev_info.values()],
)
def _init_devices(
self,
devices: Union[Iterable[dict], Mapping[str, dict], None],
devices_file: Optional[str],
):
devs = self._load_devices_file(devices_file)
if isinstance(devices, Mapping):
devs += [{'name': name, **info} for name, info in devices.items()]
elif devices:
devs += list(devices)
for dev in devs:
dev_id = dev.get('id')
if not dev_id:
continue
info = {
'id': dev_id,
'name': dev.get('name', dev_id),
'ip': dev.get('ip') or dev.get('address'),
'key': dev.get('key') or dev.get('local_key'),
'version': float(dev.get('version', 3.5)),
}
self._dev_info[dev_id] = info
self._name_to_id[info['name']] = dev_id
if info['ip']:
self._ip_to_id[info['ip']] = dev_id
def _load_devices_file(self, devices_file: Optional[str]) -> list:
candidates = (
[devices_file] if devices_file else list(self._DEFAULT_DEVICES_FILES)
)
try:
candidates.append(
os.path.join(Config.get_workdir(), 'light.tuya', 'devices.json')
)
except AssertionError:
pass
for fname in candidates:
if fname and os.path.isfile(os.path.expanduser(fname)):
with open(os.path.expanduser(fname)) as f:
return json.load(f)
return []
def _resolve_dev_id(self, device) -> str:
if isinstance(device, Entity):
device = device.external_id or device.name
if isinstance(device, Mapping):
device = device.get('id') or device.get('name') or device.get('ip')
for idx in (self._dev_info, self._name_to_id, self._ip_to_id):
if device in idx:
return device if idx is self._dev_info else idx[device] # type: ignore
# Case-insensitive name match
for name, dev_id in self._name_to_id.items():
if name.lower() == str(device).lower():
return dev_id
raise AssertionError(f'No such Tuya device: {device}')
def _get_bulb(self, device) -> 'tinytuya.BulbDevice':
import tinytuya
dev_id = self._resolve_dev_id(device)
bulb = self._bulbs.get(dev_id)
if bulb:
return bulb
info = self._dev_info[dev_id]
assert info['ip'], f'Missing IP address for device {dev_id}'
bulb = tinytuya.BulbDevice(
dev_id=info['id'],
address=info['ip'],
local_key=info['key'],
version=info['version'],
)
bulb.set_socketPersistent(True)
bulb.set_socketTimeout(self.connect_timeout)
bulb.set_socketRetryLimit(2)
bulb.set_socketRetryDelay(1)
self._bulbs[dev_id] = bulb
self._bulb_locks.setdefault(dev_id, threading.Lock())
return bulb
def _reset_bulb(self, dev_id: str):
bulb = self._bulbs.pop(dev_id, None)
if bulb:
try:
bulb.close()
except Exception:
pass
def _bulb_call(self, device, method, *args, **kwargs):
import tinytuya
dev_id = self._resolve_dev_id(device)
bulb = self._get_bulb(device)
with self._bulb_locks.setdefault(dev_id, threading.Lock()):
try:
ret = getattr(bulb, method)(*args, **kwargs)
if isinstance(ret, dict) and ret.get('Error'):
raise RuntimeError(f'{dev_id}: {ret["Error"]} - {ret.get("Err")}')
return ret
except (OSError, RuntimeError, tinytuya.DecodeError) as e:
self._reset_bulb(dev_id)
raise e
def _bulb_call_safe(self, dev_id: str, method, *args, **kwargs):
"""
Like :meth:`._bulb_call`, but logs failures instead of raising, so one
unreachable device doesn't abort an action on the other devices.
Returns the call result or None on failure.
"""
try:
return self._bulb_call(dev_id, method, *args, **kwargs)
except Exception as e:
self.logger.warning(
'Could not call %s on %s: %s',
method,
self._dev_info.get(dev_id, {}).get('name', dev_id),
e,
)
self._cached_dps.pop(dev_id, None)
def _get_state(self, device) -> dict:
"""
Retrieve and normalize the state of a device.
"""
bulb = self._get_bulb(device)
info = self._dev_info[self._resolve_dev_id(device)]
st = self._bulb_call(device, 'status')
if not isinstance(st, dict) or st.get('Error'):
return {
'id': info['id'],
'name': info['name'],
'ip': info['ip'],
'reachable': False,
'on': False,
}
# Tuya devices sometimes answer a status query with a partial DPS
# update (especially right after a CONTROL command) - merge it with
# the last known state to get a consistent view.
dev_id = info['id']
cached_dps = self._cached_dps.setdefault(dev_id, {})
cached_dps.update(st.get('dps') or {})
dps = cached_dps
dpset = bulb.dpset or {}
v_min = dpset.get('value_min', 10)
v_max = dpset.get('value_max', 1000)
def dps_val(key, default=None):
idx = dpset.get(key)
return dps.get(str(idx), default) if idx else default
mode = dps_val('mode')
hsv = None
hexval = dps_val('colour')
if hexval:
try:
hsv = bulb.hexvalue_to_hsv(hexval, dpset.get('value_hexformat'))
except Exception as e:
self.logger.debug(
'Could not parse colour value %r for %s: %s',
hexval,
info['name'],
e,
)
hue = sat = red = green = blue = x = y = None
if hsv:
h, s, v = hsv
hue = round(h * self.MAX_HUE, 2)
sat = round(s * self.MAX_SAT, 2)
red, green, blue = [round(c * 255, 2) for c in colorsys.hsv_to_rgb(h, s, v)]
x, y = self._rgb_to_xy(red, green, blue)
bri_raw = dps_val('brightness')
bri = self._tuya_to_bri(bri_raw, v_min, v_max) if bri_raw is not None else None
# In colour mode the effective brightness is carried by the V
# component of the colour DP
if mode == bulb.DPS_MODE_COLOUR and hsv:
bri = round(hsv[2] * self.MAX_BRI)
ct_raw = dps_val('colourtemp')
state = {
'id': info['id'],
'name': info['name'],
'ip': info['ip'],
'reachable': True,
'on': bool(dps_val('switch')),
'mode': mode,
'brightness': bri,
'temperature': (
self._tuya_to_temp(ct_raw, v_max) if ct_raw is not None else None
),
'hue': hue,
'saturation': sat,
'x': x,
'y': y,
'red': red,
'green': green,
'blue': blue,
'dps': dps,
}
return state
def _get_states(self, devices: Optional[Iterable] = None) -> Dict[str, dict]:
states = {}
dev_ids = (
{self._resolve_dev_id(d) for d in devices}
if devices
else set(self._dev_info.keys())
)
for dev_id in dev_ids:
try:
states[dev_id] = self._get_state(dev_id)
except Exception as e:
self.logger.warning('Could not get the status of %s: %s', dev_id, e)
states[dev_id] = {
'id': dev_id,
'name': self._dev_info[dev_id]['name'],
'ip': self._dev_info[dev_id]['ip'],
'reachable': False,
'on': False,
}
return states
def _expand_lights(self, lights=None, groups=None) -> Set[str]:
lights = lights or []
if isinstance(lights, (str, Entity, Mapping)):
lights = [lights]
if isinstance(groups, str):
groups = [groups]
dev_ids = {self._resolve_dev_id(light) for light in lights}
for group in groups or []:
grp = self.groups.get(group)
assert grp is not None, f'No such group: {group}'
dev_ids.update(self._resolve_dev_id(light) for light in grp)
if not dev_ids:
dev_ids = set(self._dev_info.keys())
return dev_ids
def _switch_expectation(self, dev_id: str, on: bool) -> dict:
bulb = self._get_bulb(dev_id)
idx = (bulb.dpset or {}).get('switch')
return {str(idx): on} if idx else {}
def _refresh_states(
self, expectations: Optional[Dict[str, dict]] = None
) -> Dict[str, dict]:
"""
Poll the devices until the DPS keys we set reflect the expected
values (they may keep serving a stale snapshot for a while after a
control command), then publish and return the new states.
:param expectations: Optional ``device ID -> DPS payload`` mapping of
the values that were just set.
"""
expectations = expectations or {}
states: Dict[str, dict] = {}
for _ in range(self._state_refresh_retries):
self.wait_stop(self._state_refresh_delay)
states = self._get_states(expectations.keys())
if all(
all(
(states.get(dev_id) or {}).get('dps', {}).get(k) == v
for k, v in expected.items()
)
for dev_id, expected in expectations.items()
):
break
self.publish_entities(list(states.values()))
return states
@staticmethod
def _rgb_to_xy(red: float, green: float, blue: float) -> tuple:
"""
Convert an 0-255 RGB triplet to CIE 1931 x,y coordinates.
"""
r, g, b = (c / 255.0 for c in (red, green, blue))
r, g, b = (
c / 12.92 if c <= 0.04045 else ((c + 0.055) / 1.055) ** 2.4
for c in (r, g, b)
)
xx = r * 0.664511 + g * 0.154324 + b * 0.162028
yy = r * 0.283881 + g * 0.668433 + b * 0.047685
zz = r * 0.000088 + g * 0.072310 + b * 0.986039
norm = xx + yy + zz
if norm == 0:
return 0.0, 0.0
return round(xx / norm, 4), round(yy / norm, 4)
@staticmethod
def _xy_to_rgb(x: float, y: float, bri: float = MAX_BRI) -> tuple:
"""
Convert CIE 1931 x,y coordinates to an 0-255 RGB triplet.
"""
if y == 0:
return 0, 0, 0
yy = bri / 255.0
xx = (yy / y) * x
zz = (yy / y) * (1 - x - y)
# Wide-gamut D65 conversion (sRGB)
r = xx * 1.656492 - yy * 0.354851 - zz * 0.255038
g = -xx * 0.707196 + yy * 1.655397 + zz * 0.036152
b = xx * 0.051713 - yy * 0.121364 + zz * 1.011530
def gamma(c):
return 12.92 * c if c <= 0.0031308 else 1.055 * (c ** (1 / 2.4)) - 0.055
r, g, b = max(0.0, r), max(0.0, g), max(0.0, b)
m = max(r, g, b)
if m > 1:
r, g, b = r / m, g / m, b / m
return tuple(round(gamma(c) * 255) for c in (r, g, b))
@staticmethod
def _bri_to_tuya(bri: float, v_min: int = 10, v_max: int = 1000) -> int:
return round(v_min + (bri / LightTuyaPlugin.MAX_BRI) * (v_max - v_min))
@staticmethod
def _tuya_to_bri(val: float, v_min: int = 10, v_max: int = 1000) -> int:
return round((val - v_min) / (v_max - v_min) * LightTuyaPlugin.MAX_BRI)
@staticmethod
def _temp_to_tuya(temp: float, v_max: int = 1000) -> int:
temp = min(max(temp, LightTuyaPlugin.MIN_TEMP), LightTuyaPlugin.MAX_TEMP)
return round(
(temp - LightTuyaPlugin.MIN_TEMP)
/ (LightTuyaPlugin.MAX_TEMP - LightTuyaPlugin.MIN_TEMP)
* v_max
)
@staticmethod
def _tuya_to_temp(val: float, v_max: int = 1000) -> int:
return round(
LightTuyaPlugin.MIN_TEMP
+ val / v_max * (LightTuyaPlugin.MAX_TEMP - LightTuyaPlugin.MIN_TEMP)
)
def _build_dps(self, bulb, state: dict, **kwargs) -> dict:
"""
Convert entity-level attributes to a Tuya DPS payload.
"""
dpset = bulb.dpset or {}
v_min, v_max = dpset.get('value_min', 10), dpset.get('value_max', 1000)
payload = {}
def set_dp(key, value):
idx = dpset.get(key)
if idx:
payload[str(idx)] = value
on = kwargs.get('on')
if on is not None:
set_dp('switch', bool(on))
# Colour takes precedence over temperature if both are passed
has_colour = any(
kwargs.get(k) is not None
for k in ('hue', 'saturation', 'xy', 'x', 'y', 'rgb', 'red')
)
has_temp = kwargs.get('temperature', kwargs.get('ct')) is not None
bri = kwargs.get('brightness', kwargs.get('bri'))
mode = kwargs.get('mode')
cur_hsv = None
if has_colour or bri is not None:
hexval = (state.get('dps') or {}).get(str(dpset.get('colour')))
if hexval:
try:
cur_hsv = bulb.hexvalue_to_hsv(hexval, dpset.get('value_hexformat'))
except Exception:
pass
if has_colour:
if kwargs.get('xy') or (
kwargs.get('x') is not None and kwargs.get('y') is not None
):
xy = kwargs.get('xy') or [kwargs['x'], kwargs['y']]
r, g, b = self._xy_to_rgb(xy[0], xy[1], self.MAX_BRI)
h, s, v = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255)
elif kwargs.get('rgb') or kwargs.get('red') is not None:
rgb = kwargs.get('rgb') or [
kwargs.get('red', 0),
kwargs.get('green', 0),
kwargs.get('blue', 0),
]
h, s, v = colorsys.rgb_to_hsv(rgb[0] / 255, rgb[1] / 255, rgb[2] / 255)
else:
h, s, v = cur_hsv or (0.0, 1.0, 1.0)
if kwargs.get('hue') is not None:
h = kwargs['hue'] / self.MAX_HUE
if kwargs.get('saturation', kwargs.get('sat')) is not None:
s = kwargs.get('saturation', kwargs.get('sat'))
if s is not None:
s /= self.MAX_SAT
# In colour mode the brightness is carried by the V component
if bri is not None:
v = bri / self.MAX_BRI
set_dp('mode', bulb.DPS_MODE_COLOUR)
set_dp(
'colour',
bulb.hsv_to_hexvalue(h, s, v, dpset.get('value_hexformat')),
)
elif has_temp:
set_dp('mode', bulb.DPS_MODE_WHITE)
temp = kwargs.get('temperature', kwargs.get('ct'))
if temp is not None:
set_dp('colourtemp', self._temp_to_tuya(temp, v_max))
if bri is not None:
set_dp('brightness', self._bri_to_tuya(bri, v_min, v_max))
elif mode:
set_dp('mode', mode)
if bri is not None and mode == bulb.DPS_MODE_WHITE:
set_dp('brightness', self._bri_to_tuya(bri, v_min, v_max))
elif bri is not None:
if state.get('mode') == bulb.DPS_MODE_COLOUR and cur_hsv:
h, s, _ = cur_hsv
set_dp(
'colour',
bulb.hsv_to_hexvalue(
h, s, bri / self.MAX_BRI, dpset.get('value_hexformat')
),
)
else:
set_dp('brightness', self._bri_to_tuya(bri, v_min, v_max))
if bri is not None and bri > 0 and on is None and str(dpset.get('switch')):
set_dp('switch', True)
scene = kwargs.get('scene')
if scene is not None:
set_dp('mode', f'{bulb.DPS_MODE_SCENE}_{scene}')
return payload
[docs]
@action
def set_lights(self, lights=None, groups=None, *_, **kwargs):
"""
Set a set of properties on a set of lights.
:param lights: List of lights to set. Each item can represent a light
name, ID, IP or entity object. If neither ``lights`` nor
``groups`` are specified, the action is applied to all the
configured lights.
:param groups: List of configured groups to set.
:param kwargs: key-value list of the parameters to set. Supported
keys:
- ``on`` (bool): Turn the lights on/off.
- ``brightness`` (0-255): Set the brightness.
- ``temperature``/``ct`` (153-500 mireds): Set the white colour
temperature.
- ``hue`` (0-360): Set the colour hue.
- ``saturation``/``sat`` (0-100): Set the colour saturation.
- ``xy`` ([x, y]): Set the colour by CIE 1931 coordinates.
- ``rgb`` ([r, g, b]) or ``red``/``green``/``blue`` (0-255):
Set the colour by RGB components.
- ``mode`` (str): Set the working mode (``white``, ``colour``,
``scene_1``-``scene_4``, ``music``).
- ``scene`` (int): Set a scene by number.
Example call::
{
"type": "request",
"action": "light.tuya.set_lights",
"args": {
"lights": ["Bathroom Left", "Bathroom Right"],
"xy": [0.6243, 0.3447],
"brightness": 197
}
}
"""
dev_ids = self._expand_lights(lights=lights, groups=groups)
states = self._get_states(dev_ids)
expectations = {}
for dev_id in dev_ids:
state = states.get(dev_id) or {}
if not state.get('reachable'):
self.logger.warning(
'Skipping unreachable device %s', state.get('name', dev_id)
)
continue
bulb = self._get_bulb(dev_id)
payload = self._build_dps(bulb, state, **kwargs)
if not payload:
self.logger.warning(
'No supported attributes to set on %s', state.get('name')
)
continue
if self._bulb_call_safe(dev_id, 'set_multiple_values', payload):
expectations[dev_id] = payload
return self._refresh_states(expectations)
[docs]
@action
def on(self, lights=None, groups=None, **_):
"""
Turn lights/groups on.
:param lights: Lights to turn on (names, IDs or IPs). Default: all
the configured lights.
:param groups: Groups to turn on. Default: all the configured groups.
"""
dev_ids = self._expand_lights(lights=lights, groups=groups)
expectations = {}
for dev_id in dev_ids:
if self._bulb_call_safe(dev_id, 'turn_on'):
expectations[dev_id] = self._switch_expectation(dev_id, True)
return self._refresh_states(expectations)
[docs]
@action
def off(self, lights=None, groups=None, **_):
"""
Turn lights/groups off.
:param lights: Lights to turn off (names, IDs or IPs). Default: all
the configured lights.
:param groups: Groups to turn off. Default: all the configured groups.
"""
dev_ids = self._expand_lights(lights=lights, groups=groups)
expectations = {}
for dev_id in dev_ids:
if self._bulb_call_safe(dev_id, 'turn_off'):
expectations[dev_id] = self._switch_expectation(dev_id, False)
return self._refresh_states(expectations)
[docs]
@action
def toggle(self, lights=None, groups=None, **_):
"""
Toggle lights/groups.
:param lights: Lights to toggle (names, IDs or IPs). Default: all
the configured lights.
:param groups: Groups to toggle. Default: all the configured groups.
"""
dev_ids = self._expand_lights(lights=lights, groups=groups)
states = self._get_states(dev_ids)
expectations = {}
for dev_id in dev_ids:
new_state = not (states.get(dev_id) or {}).get('on')
if self._bulb_call_safe(dev_id, 'turn_off' if not new_state else 'turn_on'):
expectations[dev_id] = self._switch_expectation(dev_id, new_state)
return self._refresh_states(expectations)
[docs]
@action
def get_lights(self, *_, **__) -> dict:
"""
Get the configured lights in the format
``{id: {"name": ..., "state": {"on": ..., "bri": ..., "sat": ...,
"hue": ..., "ct": ..., "xy": ...}}}`` (compatible with the light
panel frontend components).
"""
states = self._get_states()
return {
dev_id: {
'id': dev_id,
'name': state['name'],
'state': {
'on': state.get('on'),
'reachable': state.get('reachable'),
'bri': state.get('brightness'),
'sat': state.get('saturation'),
'hue': state.get('hue'),
'ct': state.get('temperature'),
'xy': (
[state['x'], state['y']] if state.get('x') is not None else None
),
'red': state.get('red'),
'green': state.get('green'),
'blue': state.get('blue'),
'colormode': state.get('mode'),
},
}
for dev_id, state in states.items()
}
[docs]
@action
def get_groups(self, *_, **__) -> dict:
"""
Get the configured groups in the format
``{name: {"name": ..., "lights": [...], "state": {"any_on": ...}}}``
(compatible with the light panel frontend components).
"""
states = self._get_states()
return {
name: {
'id': name,
'name': name,
'lights': [self._resolve_dev_id(light) for light in lights],
'state': {
'any_on': any(
(states.get(dev_id) or {}).get('on')
for dev_id in (self._resolve_dev_id(light) for light in lights)
),
},
}
for name, lights in self.groups.items()
}
[docs]
@action
def status(self, *_, **__) -> Iterable[LightEntity]:
"""
Get the current state of all the configured lights.
"""
states = self._get_states()
entities = self.transform_entities(list(states.values()))
self.publish_entities(list(states.values()))
for light in entities:
light.id = light.external_id # type: ignore
for attr, value in (light.data or {}).items():
setattr(light, attr, value)
if light.external_id is not None:
del light.external_id
if light.data is not None:
del light.data
return entities
def main(self):
if not self.poll_interval:
return
states_prev = self._get_states()
self.publish_entities(list(states_prev.values()))
while not self.should_stop():
try:
states_new = self._get_states()
for dev_id, state in states_new.items():
prev = states_prev.get(dev_id) or {}
if state == prev:
continue
event_args = {
'plugin_name': 'light.tuya',
'light_id': dev_id,
'light_name': state.get('name'),
}
if state.get('on') != prev.get('on'):
event_args['on'] = state.get('on')
if state.get('brightness') != prev.get('brightness'):
event_args['bri'] = state.get('brightness')
if state.get('saturation') != prev.get('saturation'):
event_args['sat'] = state.get('saturation')
if state.get('hue') != prev.get('hue'):
event_args['hue'] = state.get('hue')
if state.get('temperature') != prev.get('temperature'):
event_args['ct'] = state.get('temperature')
if (state.get('x'), state.get('y')) != (
prev.get('x'),
prev.get('y'),
):
event_args['xy'] = [state.get('x'), state.get('y')]
if len(event_args) > 3:
get_bus().post(LightStatusChangeEvent(**event_args))
self.publish_entities([state])
states_prev = states_new
except Exception as e:
self.logger.warning('Error while polling Tuya lights: %s', e)
finally:
self.wait_stop(self.poll_interval)
# vim:sw=4:ts=4:et:
