williamp/localtuya-modded
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/pytuya_refactoring' into physical_device

Browse Source
This commit is contained in:
rospogrigio
2020-09-11 11:54:04 +02:00
parent 61fd48cbda 184199ef87
commit 9c3786b7a6
5 changed files with 64 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
custom_components/localtuya/cover.py
View File

@@ -43,7 +43,7 @@ _LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = 'localtuyacover' DEFAULT_NAME = 'localtuyacover'
REQUIREMENTS = ['pytuya>=7.1.0'] REQUIREMENTS = ['pytuya>=8.0.0']
CONF_DEVICE_ID = 'device_id' CONF_DEVICE_ID = 'device_id'
CONF_LOCAL_KEY = 'local_key' CONF_LOCAL_KEY = 'local_key'
@@ -84,15 +84,15 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
#_LOGGER.info("conf_STOP_cmd is %s", config.get(CONF_STOP_CMD)) #_LOGGER.info("conf_STOP_cmd is %s", config.get(CONF_STOP_CMD))
covers = [] covers = []
pytuyadevice = pytuya.CoverEntity(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY)) pytuyadevice = pytuya.PytuyaDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY))
pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION))) pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION)))
dps = {} dps = {}
dps[config.get(CONF_ID)]=None dps[config.get(CONF_ID)]=None
pytuyadevice.set_dpsUsed(dps) pytuyadevice.set_dpsUsed(dps)
cover_device = TuyaCoverCache(pytuyadevice) cover_device = TuyaCache(pytuyadevice)
covers.append( covers.append(
TuyaDevice( LocaltuyaCover(
cover_device, cover_device,
config.get(CONF_NAME), config.get(CONF_NAME),
config.get(CONF_FRIENDLY_NAME), config.get(CONF_FRIENDLY_NAME),
@@ -110,8 +110,8 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
add_entities(covers, True) add_entities(covers, True)
class TuyaCoverCache: class TuyaCache:
"""Cache wrapper for pytuya.CoverEntity""" """Cache wrapper for pytuya.PytuyaDevice"""
def __init__(self, device): def __init__(self, device):
"""Initialize the cache.""" """Initialize the cache."""
@@ -139,15 +139,15 @@ class TuyaCoverCache:
# return None # return None
raise ConnectionError("Failed to update status .") raise ConnectionError("Failed to update status .")
def set_status(self, state, switchid): def set_dps(self, state, switchid):
#_LOGGER.info("running def set_status from cover") #_LOGGER.info("running def set_dps from cover")
"""Change the Tuya switch status and clear the cache.""" """Change the Tuya switch status and clear the cache."""
self._cached_status = '' self._cached_status = ''
self._cached_status_time = 0 self._cached_status_time = 0
for i in range(5): for i in range(5):
try: try:
#_LOGGER.info("Running a try from def set_status from cover where state=%s and switchid=%s", state, switchid) #_LOGGER.info("Running a try from def set_dps from cover where state=%s and switchid=%s", state, switchid)
return self._device.set_status(state, switchid) return self._device.set_dps(state, switchid)
except Exception: except Exception:
print('Failed to set status of device [{}]'.format(self._device.address)) print('Failed to set status of device [{}]'.format(self._device.address))
if i+1 == 3: if i+1 == 3:
@@ -169,7 +169,7 @@ class TuyaCoverCache:
finally: finally:
self._lock.release() self._lock.release()
class TuyaDevice(CoverEntity): class LocaltuyaCover(CoverEntity):
"""Tuya cover devices.""" """Tuya cover devices."""
def __init__(self, device, name, friendly_name, icon, switchid, open_cmd, close_cmd, stop_cmd): def __init__(self, device, name, friendly_name, icon, switchid, open_cmd, close_cmd, stop_cmd):
@@ -185,7 +185,7 @@ class TuyaDevice(CoverEntity):
self._open_cmd = open_cmd self._open_cmd = open_cmd
self._close_cmd = close_cmd self._close_cmd = close_cmd
self._stop_cmd = stop_cmd self._stop_cmd = stop_cmd
#_LOGGER.info("running def __init__ of TuyaDevice(CoverEntity) from cover.py with self=%s device=%s name=%s friendly_name=%s icon=%s switchid=%s open_cmd=%s close_cmd=%s stop_cmd=%s", self, device, name, friendly_name, icon, switchid, open_cmd, close_cmd, stop_cmd) #_LOGGER.info("running def __init__ of TuyaDevice(PytuyaDevice) from cover.py with self=%s device=%s name=%s friendly_name=%s icon=%s switchid=%s open_cmd=%s close_cmd=%s stop_cmd=%s", self, device, name, friendly_name, icon, switchid, open_cmd, close_cmd, stop_cmd)
print('Initialized tuya cover [{}] with switch status [{}] and state [{}]'.format(self._name, self._status, self._state)) print('Initialized tuya cover [{}] with switch status [{}] and state [{}]'.format(self._name, self._status, self._state))
@property @property
@@ -294,22 +294,22 @@ class TuyaDevice(CoverEntity):
def open_cover(self, **kwargs): def open_cover(self, **kwargs):
"""Open the cover.""" """Open the cover."""
#_LOGGER.info("running open_cover from cover") #_LOGGER.info("running open_cover from cover")
self._device.set_status(self._open_cmd, self._switch_id) self._device.set_dps(self._open_cmd, self._switch_id)
# self._state = 'on' # self._state = 'on'
# self._device._device.open_cover() # self._device._device.open_cover()
def close_cover(self, **kwargs): def close_cover(self, **kwargs):
#_LOGGER.info("running close_cover from cover") #_LOGGER.info("running close_cover from cover")
"""Close cover.""" """Close cover."""
#_LOGGER.info('about to set_status from cover of off, %s', self._switch_id) #_LOGGER.info('about to set_dps from cover of off, %s', self._switch_id)
self._device.set_status(self._close_cmd, self._switch_id) self._device.set_dps(self._close_cmd, self._switch_id)
# self._state = 'off' # self._state = 'off'
# self._device._device.close_cover() # self._device._device.close_cover()
def stop_cover(self, **kwargs): def stop_cover(self, **kwargs):
#_LOGGER.info("running stop_cover from cover") #_LOGGER.info("running stop_cover from cover")
"""Stop the cover.""" """Stop the cover."""
self._device.set_status(self._stop_cmd, self._switch_id) self._device.set_dps(self._stop_cmd, self._switch_id)
# self._state = 'stop' # self._state = 'stop'
# self._device._device.stop_cover() # self._device._device.stop_cover()

26
custom_components/localtuya/fan.py
View File

@@ -32,7 +32,7 @@ _LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = 'localtuyafan' DEFAULT_NAME = 'localtuyafan'
REQUIREMENTS = ['pytuya>=7.1.0'] REQUIREMENTS = ['pytuya>=8.0.0']
CONF_DEVICE_ID = 'device_id' CONF_DEVICE_ID = 'device_id'
CONF_LOCAL_KEY = 'local_key' CONF_LOCAL_KEY = 'local_key'
@@ -58,13 +58,13 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up of the Tuya switch.""" """Set up of the Tuya switch."""
from . import pytuya from . import pytuya
fans = [] fans = []
localtuyadevice = pytuya.FanDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY)) pytuyadevice = pytuya.PytuyaDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY))
localtuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION))) pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION)))
_LOGGER.debug("localtuya fan: setup_platform: %s", localtuyadevice) _LOGGER.debug("localtuya fan: setup_platform: %s", pytuyadevice)
fan_device = localtuyadevice fan_device = pytuyadevice
fans.append( fans.append(
TuyaDevice( LocaltuyaFan(
fan_device, fan_device,
config.get(CONF_NAME), config.get(CONF_NAME),
config.get(CONF_FRIENDLY_NAME), config.get(CONF_FRIENDLY_NAME),
@@ -76,8 +76,8 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
add_entities(fans, True) add_entities(fans, True)
class TuyaDevice(FanEntity): class LocaltuyaFan(FanEntity):
"""A demonstration fan component.""" """Representation of a Tuya fan."""
# def __init__(self, hass, name: str, supported_features: int) -> None: # def __init__(self, hass, name: str, supported_features: int) -> None:
def __init__(self, device, name, friendly_name, icon, switchid): def __init__(self, device, name, friendly_name, icon, switchid):
@@ -137,7 +137,7 @@ class TuyaDevice(FanEntity):
def turn_off(self, **kwargs) -> None: def turn_off(self, **kwargs) -> None:
"""Turn off the entity.""" """Turn off the entity."""
_LOGGER.debug("localtuya fan: turn_off") _LOGGER.debug("localtuya fan: turn_off")
self._device.set_status(False, '1') self._device.set_dps(False, '1')
self._state = False self._state = False
self.schedule_update_ha_state() self.schedule_update_ha_state()
@@ -147,14 +147,14 @@ class TuyaDevice(FanEntity):
self._speed = speed self._speed = speed
# self.schedule_update_ha_state() # self.schedule_update_ha_state()
if speed == STATE_OFF: if speed == STATE_OFF:
self._device.set_status(False, '1') self._device.set_dps(False, '1')
self._state = False self._state = False
elif speed == SPEED_LOW: elif speed == SPEED_LOW:
self._device.set_value('2', '1') self._device.set_dps('1', '2')
elif speed == SPEED_MEDIUM: elif speed == SPEED_MEDIUM:
self._device.set_value('2', '2') self._device.set_dps('2', '2')
elif speed == SPEED_HIGH: elif speed == SPEED_HIGH:
self._device.set_value('2', '3') self._device.set_dps('3', '2')
self.schedule_update_ha_state() self.schedule_update_ha_state()
# def set_direction(self, direction: str) -> None: # def set_direction(self, direction: str) -> None:

18
custom_components/localtuya/light.py
View File

@@ -31,7 +31,7 @@ from homeassistant.components.light import (
from homeassistant.util import color as colorutil from homeassistant.util import color as colorutil
import socket import socket
REQUIREMENTS = ['pytuya>=7.1.0'] REQUIREMENTS = ['pytuya>=8.0.0']
CONF_DEVICE_ID = 'device_id' CONF_DEVICE_ID = 'device_id'
CONF_LOCAL_KEY = 'local_key' CONF_LOCAL_KEY = 'local_key'
@@ -62,12 +62,12 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
from . import pytuya from . import pytuya
lights = [] lights = []
pytuyadevice = pytuya.BulbDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY)) pytuyadevice = pytuya.PytuyaDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY))
pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION))) pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION)))
bulb_device = TuyaCache(pytuyadevice) bulb_device = TuyaCache(pytuyadevice)
lights.append( lights.append(
TuyaDevice( LocaltuyaLight(
bulb_device, bulb_device,
config.get(CONF_NAME), config.get(CONF_NAME),
config.get(CONF_FRIENDLY_NAME), config.get(CONF_FRIENDLY_NAME),
@@ -79,7 +79,7 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
add_devices(lights) add_devices(lights)
class TuyaCache: class TuyaCache:
"""Cache wrapper for pytuya.BulbDevice""" """Cache wrapper for pytuya.PytuyaDevices"""
def __init__(self, device): def __init__(self, device):
"""Initialize the cache.""" """Initialize the cache."""
@@ -105,13 +105,13 @@ class TuyaCache:
log.warn( log.warn(
"Failed to get status after {} tries".format(UPDATE_RETRY_LIMIT)) "Failed to get status after {} tries".format(UPDATE_RETRY_LIMIT))
def set_status(self, state, switchid): def set_dps(self, state, switchid):
"""Change the Tuya switch status and clear the cache.""" """Change the Tuya switch status and clear the cache."""
self._cached_status = '' self._cached_status = ''
self._cached_status_time = 0 self._cached_status_time = 0
for _ in range(UPDATE_RETRY_LIMIT): for _ in range(UPDATE_RETRY_LIMIT):
try: try:
return self._device.set_status(state, switchid) return self._device.set_dps(state, switchid)
except ConnectionError: except ConnectionError:
pass pass
except socket.timeout: except socket.timeout:
@@ -202,7 +202,7 @@ class TuyaCache:
def turn_off(self): def turn_off(self):
self._device.turn_off(); self._device.turn_off();
class TuyaDevice(LightEntity): class LocaltuyaLight(LightEntity):
"""Representation of a Tuya switch.""" """Representation of a Tuya switch."""
def __init__(self, device, name, friendly_name, icon, bulbid): def __init__(self, device, name, friendly_name, icon, bulbid):
@@ -296,7 +296,7 @@ class TuyaDevice(LightEntity):
"""Turn on or control the light.""" """Turn on or control the light."""
log.debug("Turning on, state: " + str(self._device.cached_status())) log.debug("Turning on, state: " + str(self._device.cached_status()))
if not self._device.cached_status(): if not self._device.cached_status():
self._device.set_status(True, self._bulb_id) self._device.set_dps(True, self._bulb_id)
if ATTR_BRIGHTNESS in kwargs: if ATTR_BRIGHTNESS in kwargs:
converted_brightness = int(kwargs[ATTR_BRIGHTNESS]) converted_brightness = int(kwargs[ATTR_BRIGHTNESS])
if converted_brightness <= 25: if converted_brightness <= 25:
@@ -310,7 +310,7 @@ class TuyaDevice(LightEntity):
def turn_off(self, **kwargs): def turn_off(self, **kwargs):
"""Turn Tuya switch off.""" """Turn Tuya switch off."""
self._device.set_status(False, self._bulb_id) self._device.set_dps(False, self._bulb_id)
@property @property
def supported_features(self): def supported_features(self):

319
custom_components/localtuya/pytuya/__init__.py
View File

@@ -1,4 +1,4 @@
# TinyTuya Module # PyTuya Module
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
Python module to interface with Tuya WiFi smart devices Python module to interface with Tuya WiFi smart devices
@@ -11,10 +11,7 @@
For more information see https://github.com/clach04/python-tuya For more information see https://github.com/clach04/python-tuya
Classes Classes
OutletDevice(dev_id, address, local_key=None) PytuyaDevice(dev_id, address, local_key=None)
CoverDevice(dev_id, address, local_key=None)
BulbDevice(dev_id, address, local_key=None)
dev_id (str): Device ID e.g. 01234567891234567890 dev_id (str): Device ID e.g. 01234567891234567890
address (str): Device Network IP Address e.g. 10.0.1.99 address (str): Device Network IP Address e.g. 10.0.1.99
local_key (str, optional): The encryption key. Defaults to None. local_key (str, optional): The encryption key. Defaults to None.
@@ -23,27 +20,10 @@
json = status() # returns json payload json = status() # returns json payload
set_version(version) # 3.1 [default] or 3.3 set_version(version) # 3.1 [default] or 3.3
set_dpsUsed(dpsUsed) set_dpsUsed(dpsUsed)
set_status(on, switch=1) # Set status of the device to 'on' or 'off' (bool) set_dps(on, switch=1) # Set the of the device to 'on' or 'off' (bool)
set_value(index, value) # Set int value of any index. set_value(index, value) # Set int value of any index.
turn_on(switch=1):
turn_off(switch=1):
set_timer(num_secs): set_timer(num_secs):
CoverDevice:
open_cover(switch=1):
close_cover(switch=1):
stop_cover(switch=1):
BulbDevice
set_colour(r, g, b):
set_white(brightness, colourtemp):
set_brightness(brightness):
set_colourtemp(colourtemp):
result = brightness():
result = colourtemp():
(r, g, b) = colour_rgb():
(h,s,v) = colour_hsv()
result = state():
Credits Credits
* TuyaAPI https://github.com/codetheweb/tuyapi by codetheweb and blackrozes * TuyaAPI https://github.com/codetheweb/tuyapi by codetheweb and blackrozes
@@ -73,7 +53,7 @@ except ImportError:
import pyaes # https://github.com/ricmoo/pyaes import pyaes # https://github.com/ricmoo/pyaes
version_tuple = (7, 1, 0) version_tuple = (8, 0, 0)
version = version_string = __version__ = '%d.%d.%d' % version_tuple version = version_string = __version__ = '%d.%d.%d' % version_tuple
__author__ = 'rospogrigio' __author__ = 'rospogrigio'
@@ -192,7 +172,12 @@ payload_dict = {
} }
} }
class XenonDevice(object): #class PytuyaDevice(XenonDevice):
# def __init__(self, dev_id, address, local_key=None, dev_type=None):
# super(PytuyaDevice, self).__init__(dev_id, address, local_key, dev_type)
#class XenonDevice(object):
class PytuyaDevice(object):
def __init__(self, dev_id, address, local_key=None, connection_timeout=10): def __init__(self, dev_id, address, local_key=None, connection_timeout=10):
""" """
Represents a Tuya device. Represents a Tuya device.
@@ -208,7 +193,6 @@ class XenonDevice(object):
self.id = dev_id self.id = dev_id
self.address = address self.address = address
self.local_key = local_key
self.local_key = local_key.encode('latin1') self.local_key = local_key.encode('latin1')
self.connection_timeout = connection_timeout self.connection_timeout = connection_timeout
self.version = 3.1 self.version = 3.1
@@ -332,11 +316,7 @@ class XenonDevice(object):
#print('full buffer(%d) %r' % (len(buffer), bin2hex(buffer, pretty=True) )) #print('full buffer(%d) %r' % (len(buffer), bin2hex(buffer, pretty=True) ))
#print('full buffer(%d) %r' % (len(buffer), " ".join("{:02x}".format(ord(c)) for c in buffer))) #print('full buffer(%d) %r' % (len(buffer), " ".join("{:02x}".format(ord(c)) for c in buffer)))
return buffer return buffer
class Device(XenonDevice):
def __init__(self, dev_id, address, local_key=None, dev_type=None):
super(Device, self).__init__(dev_id, address, local_key, dev_type)
def status(self): def status(self):
log.debug('status() entry (dev_type is %s)', self.dev_type) log.debug('status() entry (dev_type is %s)', self.dev_type)
# open device, send request, then close connection # open device, send request, then close connection
@@ -384,52 +364,27 @@ class Device(XenonDevice):
log.error('Unexpected status() payload=%r', result) log.error('Unexpected status() payload=%r', result)
return result return result
def set_status(self, on, switch=1):
"""
Set status of the device to 'on' or 'off'.
Args:
on(bool): True for 'on', False for 'off'.
switch(int): The switch to set
"""
# open device, send request, then close connection
if isinstance(switch, int):
switch = str(switch) # index and payload is a string
payload = self.generate_payload(SET, {switch:on})
#print('payload %r' % payload)
data = self._send_receive(payload)
log.debug('set_status received data=%r', data)
return data
def set_value(self, index, value): def set_dps(self, value, dpsIndex):
""" """
Set int value of any index. Set int value of any index.
Args: Args:
index(int): index to set dpsIndex(int): dps index to set
value(int): new value for the index value: new value for the dps index
""" """
# open device, send request, then close connection # open device, send request, then close connection
if isinstance(index, int): if isinstance(dpsIndex, int):
index = str(index) # index and payload is a string dpsIndex = str(dpsIndex) # index and payload is a string
payload = self.generate_payload(SET, { payload = self.generate_payload(SET, {
index: value}) dpsIndex: value})
data = self._send_receive(payload) data = self._send_receive(payload)
log.debug('set_dps received data=%r', data)
return data return data
def turn_on(self, switch=1):
"""Turn the device on"""
self.set_status(True, switch)
def turn_off(self, switch=1):
"""Turn the device off"""
self.set_status(False, switch)
def set_timer(self, num_secs): def set_timer(self, num_secs):
""" """
@@ -452,241 +407,3 @@ class Device(XenonDevice):
data = self._send_receive(payload) data = self._send_receive(payload)
log.debug('set_timer received data=%r', data) log.debug('set_timer received data=%r', data)
return data return data
class OutletDevice(Device):
def __init__(self, dev_id, address, local_key=None):
super(OutletDevice, self).__init__(dev_id, address, local_key)
class FanDevice(Device):
DPS_INDEX_SPEED = '2'
def __init__(self, dev_id, address, local_key=None):
super(FanDevice, self).__init__(dev_id, address, local_key)
class CoverEntity(Device):
DPS_INDEX_MOVE = '1'
DPS_INDEX_BL = '101'
DPS_2_STATE = {
'1':'movement',
'101':'backlight',
}
def __init__(self, dev_id, address, local_key=None):
print('%s version %s' % ( __name__, version))
print('Python %s on %s' % (sys.version, sys.platform))
if Crypto is None:
print('Using pyaes version ', pyaes.VERSION)
print('Using pyaes from ', pyaes.__file__)
else:
print('Using PyCrypto ', Crypto.version_info)
print('Using PyCrypto from ', Crypto.__file__)
super(CoverEntity, self).__init__(dev_id, address, local_key)
def open_cover(self, switch=1):
"""Turn the device on"""
self.set_status('on', switch)
def close_cover(self, switch=1):
"""Turn the device off"""
self.set_status('off', switch)
def stop_cover(self, switch=1):
"""Turn the device off"""
self.set_status('stop', switch)
class BulbDevice(Device):
DPS_INDEX_ON = '1'
DPS_INDEX_MODE = '2'
DPS_INDEX_BRIGHTNESS = '3'
DPS_INDEX_COLOURTEMP = '4'
DPS_INDEX_COLOUR = '5'
DPS = 'dps'
DPS_MODE_COLOUR = 'colour'
DPS_MODE_WHITE = 'white'
DPS_2_STATE = {
'1':'is_on',
'2':'mode',
'3':'brightness',
'4':'colourtemp',
'5':'colour',
}
def __init__(self, dev_id, address, local_key=None):
super(BulbDevice, self).__init__(dev_id, address, local_key)
@staticmethod
def _rgb_to_hexvalue(r, g, b):
"""
Convert an RGB value to the hex representation expected by tuya.
Index '5' (DPS_INDEX_COLOUR) is assumed to be in the format:
rrggbb0hhhssvv
While r, g and b are just hexadecimal values of the corresponding
Red, Green and Blue values, the h, s and v values (which are values
between 0 and 1) are scaled to 360 (h) and 255 (s and v) respectively.
Args:
r(int): Value for the colour red as int from 0-255.
g(int): Value for the colour green as int from 0-255.
b(int): Value for the colour blue as int from 0-255.
"""
rgb = [r,g,b]
hsv = colorsys.rgb_to_hsv(rgb[0]/255, rgb[1]/255, rgb[2]/255)
hexvalue = ""
for value in rgb:
temp = str(hex(int(value))).replace("0x","")
if len(temp) == 1:
temp = "0" + temp
hexvalue = hexvalue + temp
hsvarray = [int(hsv[0] * 360), int(hsv[1] * 255), int(hsv[2] * 255)]
hexvalue_hsv = ""
for value in hsvarray:
temp = str(hex(int(value))).replace("0x","")
if len(temp) == 1:
temp = "0" + temp
hexvalue_hsv = hexvalue_hsv + temp
if len(hexvalue_hsv) == 7:
hexvalue = hexvalue + "0" + hexvalue_hsv
else:
hexvalue = hexvalue + "00" + hexvalue_hsv
return hexvalue
@staticmethod
def _hexvalue_to_rgb(hexvalue):
"""
Converts the hexvalue used by tuya for colour representation into
an RGB value.
Args:
hexvalue(string): The hex representation generated by BulbDevice._rgb_to_hexvalue()
"""
r = int(hexvalue[0:2], 16)
g = int(hexvalue[2:4], 16)
b = int(hexvalue[4:6], 16)
return (r, g, b)
@staticmethod
def _hexvalue_to_hsv(hexvalue):
"""
Converts the hexvalue used by tuya for colour representation into
an HSV value.
Args:
hexvalue(string): The hex representation generated by BulbDevice._rgb_to_hexvalue()
"""
h = int(hexvalue[7:10], 16) / 360
s = int(hexvalue[10:12], 16) / 255
v = int(hexvalue[12:14], 16) / 255
return (h, s, v)
def set_colour(self, r, g, b):
"""
Set colour of an rgb bulb.
Args:
r(int): Value for the colour red as int from 0-255.
g(int): Value for the colour green as int from 0-255.
b(int): Value for the colour blue as int from 0-255.
"""
if not 0 <= r <= 255:
raise ValueError("The value for red needs to be between 0 and 255.")
if not 0 <= g <= 255:
raise ValueError("The value for green needs to be between 0 and 255.")
if not 0 <= b <= 255:
raise ValueError("The value for blue needs to be between 0 and 255.")
#print(BulbDevice)
hexvalue = BulbDevice._rgb_to_hexvalue(r, g, b)
payload = self.generate_payload(SET, {
self.DPS_INDEX_MODE: self.DPS_MODE_COLOUR,
self.DPS_INDEX_COLOUR: hexvalue})
data = self._send_receive(payload)
return data
def set_white(self, brightness, colourtemp):
"""
Set white coloured theme of an rgb bulb.
Args:
brightness(int): Value for the brightness (25-255).
colourtemp(int): Value for the colour temperature (0-255).
"""
if not 25 <= brightness <= 255:
raise ValueError("The brightness needs to be between 25 and 255.")
if not 0 <= colourtemp <= 255:
raise ValueError("The colour temperature needs to be between 0 and 255.")
payload = self.generate_payload(SET, {
self.DPS_INDEX_MODE: self.DPS_MODE_WHITE,
self.DPS_INDEX_BRIGHTNESS: brightness,
self.DPS_INDEX_COLOURTEMP: colourtemp})
data = self._send_receive(payload)
return data
def set_brightness(self, brightness):
"""
Set the brightness value of an rgb bulb.
Args:
brightness(int): Value for the brightness (25-255).
"""
if not 25 <= brightness <= 255:
raise ValueError("The brightness needs to be between 25 and 255.")
payload = self.generate_payload(SET, {self.DPS_INDEX_BRIGHTNESS: brightness})
data = self._send_receive(payload)
return data
def set_colourtemp(self, colourtemp):
"""
Set the colour temperature of an rgb bulb.
Args:
colourtemp(int): Value for the colour temperature (0-255).
"""
if not 0 <= colourtemp <= 255:
raise ValueError("The colour temperature needs to be between 0 and 255.")
payload = self.generate_payload(SET, {self.DPS_INDEX_COLOURTEMP: colourtemp})
data = self._send_receive(payload)
return data
def brightness(self):
"""Return brightness value"""
return self.status()[self.DPS][self.DPS_INDEX_BRIGHTNESS]
def colourtemp(self):
"""Return colour temperature"""
return self.status()[self.DPS][self.DPS_INDEX_COLOURTEMP]
def colour_rgb(self):
"""Return colour as RGB value"""
hexvalue = self.status()[self.DPS][self.DPS_INDEX_COLOUR]
return BulbDevice._hexvalue_to_rgb(hexvalue)
def colour_hsv(self):
"""Return colour as HSV value"""
hexvalue = self.status()[self.DPS][self.DPS_INDEX_COLOUR]
return BulbDevice._hexvalue_to_hsv(hexvalue)
def state(self):
status = self.status()
state = {}
for key in status[self.DPS].keys():
if(int(key)<=5):
state[self.DPS_2_STATE[key]]=status[self.DPS][key]
return state

18
custom_components/localtuya/switch.py
View File

@@ -58,8 +58,7 @@ from .const import (
) )
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ['pytuya>=8.0.0']
REQUIREMENTS = ["pytuya==7.0.9"]
DEFAULT_ID = "1" DEFAULT_ID = "1"
DEFAULT_PROTOCOL_VERSION = 3.3 DEFAULT_PROTOCOL_VERSION = 3.3
@@ -113,7 +112,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
return return
switches = [] switches = []
pytuyadevice = pytuya.OutletDevice( pytuyadevice = pytuya.PytuyaDevice(
config_entry.data[CONF_DEVICE_ID], config_entry.data[CONF_DEVICE_ID],
config_entry.data[CONF_HOST], config_entry.data[CONF_HOST],
config_entry.data[CONF_LOCAL_KEY], config_entry.data[CONF_LOCAL_KEY],
@@ -149,7 +148,7 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
class TuyaCache: class TuyaCache:
"""Cache wrapper for pytuya.OutletDevice""" """Cache wrapper for pytuya.PytuyaDevice"""
def __init__(self, device): def __init__(self, device):
"""Initialize the cache.""" """Initialize the cache."""
@@ -182,13 +181,13 @@ class TuyaCache:
# return None # return None
raise ConnectionError("Failed to update status .") raise ConnectionError("Failed to update status .")
def set_status(self, state, switchid): def set_dps(self, state, switchid):
"""Change the Tuya switch status and clear the cache.""" """Change the Tuya switch status and clear the cache."""
self._cached_status = "" self._cached_status = ""
self._cached_status_time = 0 self._cached_status_time = 0
for i in range(5): for i in range(5):
try: try:
return self._device.set_status(state, switchid) return self._device.set_dps(state, switchid)
except Exception: except Exception:
print( print(
"Failed to set status of device [{}]".format(self._device.address) "Failed to set status of device [{}]".format(self._device.address)
@@ -214,8 +213,7 @@ class TuyaCache:
finally: finally:
self._lock.release() self._lock.release()
class LocaltuyaSwitch(SwitchEntity):
class TuyaDevice(SwitchEntity):
"""Representation of a Tuya switch.""" """Representation of a Tuya switch."""
def __init__( def __init__(
@@ -278,11 +276,11 @@ class TuyaDevice(SwitchEntity):
def turn_on(self, **kwargs): def turn_on(self, **kwargs):
"""Turn Tuya switch on.""" """Turn Tuya switch on."""
self._device.set_status(True, self._switch_id) self._device.set_dps(True, self._switch_id)
def turn_off(self, **kwargs): def turn_off(self, **kwargs):
"""Turn Tuya switch off.""" """Turn Tuya switch off."""
self._device.set_status(False, self._switch_id) self._device.set_dps(False, self._switch_id)
def update(self): def update(self):
"""Get state of Tuya switch.""" """Get state of Tuya switch."""