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

Merge branch 'pytuya_refactoring' into physical_device

Browse Source
This commit is contained in:
rospogrigio
2020-09-15 00:15:46 +02:00
parent c008339f43 ce2461df2b
commit 6c68cdcbcc
9 changed files with 515 additions and 464 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
custom_components/localtuya/__init__.py
View File

@@ -1,13 +1,71 @@
"""The LocalTuya integration integration.""" """The LocalTuya integration integration."""
import asyncio
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.const import CONF_PLATFORM from homeassistant.const import (
CONF_DEVICE_ID,
CONF_ID,
CONF_ICON,
CONF_NAME,
CONF_FRIENDLY_NAME,
CONF_HOST,
CONF_PLATFORM,
CONF_ENTITIES,
)
import homeassistant.helpers.config_validation as cv
from .const import DOMAIN, PLATFORMS from . import pytuya
from .const import CONF_LOCAL_KEY, CONF_PROTOCOL_VERSION, DOMAIN, PLATFORMS
DEFAULT_ID = "1"
DEFAULT_PROTOCOL_VERSION = 3.3
BASE_PLATFORM_SCHEMA = {
vol.Optional(CONF_ICON): cv.icon, # Deprecated: not used
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(CONF_LOCAL_KEY): cv.string,
vol.Optional(CONF_NAME): cv.string, # Deprecated: not used
vol.Required(CONF_FRIENDLY_NAME): cv.string,
vol.Required(CONF_PROTOCOL_VERSION, default=DEFAULT_PROTOCOL_VERSION): vol.Coerce(
float
),
vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string,
}
def prepare_setup_entities(config_entry, platform):
"""Prepare ro setup entities for a platform."""
entities_to_setup = [
entity
for entity in config_entry.data[CONF_ENTITIES]
if entity[CONF_PLATFORM] == platform
]
if not entities_to_setup:
return None, None
device = pytuya.TuyaDevice(
config_entry.data[CONF_DEVICE_ID],
config_entry.data[CONF_HOST],
config_entry.data[CONF_LOCAL_KEY],
)
device.set_version(float(config_entry.data[CONF_PROTOCOL_VERSION]))
device.set_dpsUsed({})
return device, entities_to_setup
def import_from_yaml(hass, config, platform):
"""Import configuration from YAML."""
config[CONF_PLATFORM] = platform
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
return True
async def async_setup(hass: HomeAssistant, config: dict): async def async_setup(hass: HomeAssistant, config: dict):

50
custom_components/localtuya/config_flow.py
View File

@@ -45,28 +45,36 @@ PICK_ENTITY_SCHEMA = vol.Schema(
) )
def platform_schema(dps, additional_fields): def dps_string_list(dps_data):
"""Return list of friendly DPS values."""
return [f"{id} (value: {value})" for id, value in dps_data.items()]
def platform_schema(dps_strings, schema):
"""Generate input validation schema for a platform.""" """Generate input validation schema for a platform."""
dps_list = vol.In([f"{id} (value: {value})" for id, value in dps.items()])
return vol.Schema( return vol.Schema(
{ {
vol.Required(CONF_ID): dps_list, vol.Required(CONF_ID): vol.In(dps_strings),
vol.Required(CONF_FRIENDLY_NAME): str, vol.Required(CONF_FRIENDLY_NAME): str,
} }
).extend({conf: dps_list for conf in additional_fields}) ).extend(schema)
def strip_dps_values(user_input, fields): def strip_dps_values(user_input, dps_strings):
"""Remove values and keep only index for DPS config items.""" """Remove values and keep only index for DPS config items."""
for field in [CONF_ID] + fields: stripped = {}
user_input[field] = user_input[field].split(" ")[0] for field, value in user_input.items():
return user_input if value in dps_strings:
stripped[field] = user_input[field].split(" ")[0]
else:
stripped[field] = user_input[field]
return stripped
async def validate_input(hass: core.HomeAssistant, data): async def validate_input(hass: core.HomeAssistant, data):
"""Validate the user input allows us to connect.""" """Validate the user input allows us to connect."""
pytuyadevice = pytuya.PytuyaDevice( pytuyadevice = pytuya.TuyaDevice(
data[CONF_DEVICE_ID], data[CONF_HOST], data[CONF_LOCAL_KEY], data[CONF_NAME] data[CONF_DEVICE_ID], data[CONF_HOST], data[CONF_LOCAL_KEY]
) )
pytuyadevice.set_version(float(data[CONF_PROTOCOL_VERSION])) pytuyadevice.set_version(float(data[CONF_PROTOCOL_VERSION]))
pytuyadevice.set_dpsUsed({}) pytuyadevice.set_dpsUsed({})
@@ -76,7 +84,7 @@ async def validate_input(hass: core.HomeAssistant, data):
raise CannotConnect raise CannotConnect
except ValueError: except ValueError:
raise InvalidAuth raise InvalidAuth
return data["dps"] return dps_string_list(data["dps"])
class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@@ -88,9 +96,9 @@ class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self): def __init__(self):
"""Initialize a new LocaltuyaConfigFlow.""" """Initialize a new LocaltuyaConfigFlow."""
self.basic_info = None self.basic_info = None
self.dps_data = None self.dps_strings = []
self.platform = None self.platform = None
self.platform_dps_fields = None self.platform_schema = None
self.entities = [] self.entities = []
async def async_step_user(self, user_input=None): async def async_step_user(self, user_input=None):
@@ -102,7 +110,7 @@ class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
try: try:
self.basic_info = user_input self.basic_info = user_input
self.dps_data = await validate_input(self.hass, user_input) self.dps_strings = await validate_input(self.hass, user_input)
return await self.async_step_pick_entity_type() return await self.async_step_pick_entity_type()
except CannotConnect: except CannotConnect:
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
@@ -145,16 +153,14 @@ class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
) )
if not already_configured: if not already_configured:
user_input[CONF_PLATFORM] = self.platform user_input[CONF_PLATFORM] = self.platform
self.entities.append( self.entities.append(strip_dps_values(user_input, self.dps_strings))
strip_dps_values(user_input, self.platform_dps_fields)
)
return await self.async_step_pick_entity_type() return await self.async_step_pick_entity_type()
errors["base"] = "entity_already_configured" errors["base"] = "entity_already_configured"
return self.async_show_form( return self.async_show_form(
step_id="add_entity", step_id="add_entity",
data_schema=platform_schema(self.dps_data, self.platform_dps_fields), data_schema=platform_schema(self.dps_strings, self.platform_schema),
errors=errors, errors=errors,
description_placeholders={"platform": self.platform}, description_placeholders={"platform": self.platform},
) )
@@ -168,14 +174,14 @@ class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
CONF_FRIENDLY_NAME: conf[CONF_FRIENDLY_NAME], CONF_FRIENDLY_NAME: conf[CONF_FRIENDLY_NAME],
CONF_PLATFORM: self.platform, CONF_PLATFORM: self.platform,
} }
for field in self.platform_dps_fields: for field in self.platform_schema.keys():
converted[str(field)] = conf[field] converted[str(field)] = conf[field]
return converted return converted
await self.async_set_unique_id(user_input[CONF_DEVICE_ID]) await self.async_set_unique_id(user_input[CONF_DEVICE_ID])
self._set_platform(user_input[CONF_PLATFORM]) self._set_platform(user_input[CONF_PLATFORM])
if len(user_input[CONF_SWITCHES]) > 0: if len(user_input.get(CONF_SWITCHES, [])) > 0:
for switch_conf in user_input[CONF_SWITCHES].values(): for switch_conf in user_input[CONF_SWITCHES].values():
self.entities.append(_convert_entity(switch_conf)) self.entities.append(_convert_entity(switch_conf))
else: else:
@@ -197,7 +203,9 @@ class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def _set_platform(self, platform): def _set_platform(self, platform):
integration_module = ".".join(__name__.split(".")[:-1]) integration_module = ".".join(__name__.split(".")[:-1])
self.platform = platform self.platform = platform
self.platform_dps_fields = import_module("." + platform, integration_module).DPS_FIELDS self.platform_schema = import_module(
"." + platform, integration_module
).flow_schema(self.dps_strings)
class CannotConnect(exceptions.HomeAssistantError): class CannotConnect(exceptions.HomeAssistantError):

9
custom_components/localtuya/const.py
View File

@@ -6,11 +6,18 @@ ATTR_VOLTAGE = 'voltage'
CONF_LOCAL_KEY = "local_key" CONF_LOCAL_KEY = "local_key"
CONF_PROTOCOL_VERSION = "protocol_version" CONF_PROTOCOL_VERSION = "protocol_version"
# switch
CONF_CURRENT = "current" CONF_CURRENT = "current"
CONF_CURRENT_CONSUMPTION = "current_consumption" CONF_CURRENT_CONSUMPTION = "current_consumption"
CONF_VOLTAGE = "voltage" CONF_VOLTAGE = "voltage"
# cover
CONF_OPEN_CMD = 'open_cmd'
CONF_CLOSE_CMD = 'close_cmd'
CONF_STOP_CMD = 'stop_cmd'
DOMAIN = "localtuya" DOMAIN = "localtuya"
# Platforms in this list must support config flows # Platforms in this list must support config flows
PLATFORMS = ["switch"] PLATFORMS = ["cover", "fan", "light", "switch"]

279
custom_components/localtuya/cover.py
View File

@@ -19,7 +19,8 @@ cover:
""" """
import logging import logging
import requests from time import time, sleep
from threading import Lock
import voluptuous as vol import voluptuous as vol
@@ -31,97 +32,85 @@ from homeassistant.components.cover import (
SUPPORT_STOP, SUPPORT_STOP,
SUPPORT_SET_POSITION, SUPPORT_SET_POSITION,
) )
from homeassistant.const import (
"""from . import DATA_TUYA, TuyaDevice""" CONF_ID,
from homeassistant.components.cover import CoverEntity, PLATFORM_SCHEMA CONF_FRIENDLY_NAME,
from homeassistant.const import (CONF_HOST, CONF_ID, CONF_FRIENDLY_NAME, CONF_ICON, CONF_NAME) )
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from time import time, sleep
from threading import Lock from . import BASE_PLATFORM_SCHEMA, prepare_setup_entities, import_from_yaml
from .const import CONF_OPEN_CMD, CONF_CLOSE_CMD, CONF_STOP_CMD
from .pytuya import TuyaDevice
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = 'localtuyacover' PLATFORM = "cover"
REQUIREMENTS = ['pytuya>=8.0.0'] DEFAULT_OPEN_CMD = "on"
DEFAULT_CLOSE_CMD = "off"
CONF_DEVICE_ID = 'device_id' DEFAULT_STOP_CMD = "stop"
CONF_LOCAL_KEY = 'local_key'
CONF_PROTOCOL_VERSION = 'protocol_version'
CONF_OPEN_CMD = 'open_cmd'
CONF_CLOSE_CMD = 'close_cmd'
CONF_STOP_CMD = 'stop_cmd'
DEFAULT_ID = '1'
DEFAULT_PROTOCOL_VERSION = 3.3
DEFAULT_OPEN_CMD = 'on'
DEFAULT_CLOSE_CMD = 'off'
DEFAULT_STOP_CMD = 'stop'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_ICON): cv.icon,
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(CONF_LOCAL_KEY): cv.string,
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_FRIENDLY_NAME): cv.string,
vol.Required(CONF_PROTOCOL_VERSION, default=DEFAULT_PROTOCOL_VERSION): vol.Coerce(float),
vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string,
vol.Optional(CONF_OPEN_CMD, default=DEFAULT_OPEN_CMD): cv.string,
vol.Optional(CONF_CLOSE_CMD, default=DEFAULT_CLOSE_CMD): cv.string,
vol.Optional(CONF_STOP_CMD, default=DEFAULT_STOP_CMD): cv.string,
})
def setup_platform(hass, config, add_entities, discovery_info=None): PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(BASE_PLATFORM_SCHEMA).extend(
"""Set up Tuya cover devices.""" {
from . import pytuya vol.Optional(CONF_OPEN_CMD, default=DEFAULT_OPEN_CMD): cv.string,
vol.Optional(CONF_CLOSE_CMD, default=DEFAULT_CLOSE_CMD): cv.string,
vol.Optional(CONF_STOP_CMD, default=DEFAULT_STOP_CMD): cv.string,
}
)
#_LOGGER.info("running def setup_platform from cover.py")
#_LOGGER.info("conf_open_cmd is %s", config.get(CONF_OPEN_CMD)) def flow_schema(dps):
#_LOGGER.info("conf_close_cmd is %s", config.get(CONF_CLOSE_CMD)) """Return schema used in config flow."""
#_LOGGER.info("conf_STOP_cmd is %s", config.get(CONF_STOP_CMD)) return {
vol.Optional(CONF_OPEN_CMD, default=DEFAULT_OPEN_CMD): str,
vol.Optional(CONF_CLOSE_CMD, default=DEFAULT_CLOSE_CMD): str,
vol.Optional(CONF_STOP_CMD, default=DEFAULT_STOP_CMD): str,
}
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Setup a Tuya cover based on a config entry."""
device, entities_to_setup = prepare_setup_entities(
config_entry, PLATFORM
)
if not entities_to_setup:
return
# TODO: keeping for now but should be removed
dps = {}
covers = [] covers = []
pytuyadevice = pytuya.PytuyaDevice( for device_config in entities_to_setup:
config_entry.data[CONF_DEVICE_ID], dps[device_config[CONF_ID]] = None
config_entry.data[CONF_HOST], covers.append(
config_entry.data[CONF_LOCAL_KEY],
config_entry.data[CONF_FRIENDLY_NAME],
config_entry.data[CONF_NAME],
)
pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION)))
dps = {}
dps[config.get(CONF_ID)]=None
pytuyadevice.set_dpsUsed(dps)
cover_device = TuyaCache(pytuyadevice)
covers.append(
LocaltuyaCover( LocaltuyaCover(
cover_device, TuyaCache(device),
config.get(CONF_NAME), device_config[CONF_FRIENDLY_NAME],
config.get(CONF_FRIENDLY_NAME), device_config[CONF_ID],
config.get(CONF_ICON), device_config.get(CONF_OPEN_CMD),
config.get(CONF_ID), device_config.get(CONF_CLOSE_CMD),
config.get(CONF_OPEN_CMD), device_config.get(CONF_STOP_CMD),
config.get(CONF_CLOSE_CMD),
config.get(CONF_STOP_CMD),
) )
) # print('Setup localtuya cover [{}] with device ID [{}] '.format(config.get(CONF_FRIENDLY_NAME), config.get(CONF_ID)))
print('Setup localtuya cover [{}] with device ID [{}] '.format(config.get(CONF_FRIENDLY_NAME), config.get(CONF_ID))) # _LOGGER.info("Setup localtuya cover %s with device ID=%s", config.get(CONF_FRIENDLY_NAME), config.get(CONF_ID) )
_LOGGER.info("Setup localtuya cover %s with device ID=%s", config.get(CONF_FRIENDLY_NAME), config.get(CONF_ID) ) # _LOGGER.debug("Cover %s uses open_cmd=%s close_cmd=%s stop_cmd=%s", config.get(CONF_FRIENDLY_NAME), config.get(CONF_OPEN_CMD), config.get(CONF_CLOSE_CMD), config.get(CONF_STOP_CMD) )
_LOGGER.debug("Cover %s uses open_cmd=%s close_cmd=%s stop_cmd=%s", config.get(CONF_FRIENDLY_NAME), config.get(CONF_OPEN_CMD), config.get(CONF_CLOSE_CMD), config.get(CONF_STOP_CMD) ) )
add_entities(covers, True) device.set_dpsUsed(dps)
async_add_entities(covers, True)
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up of the Tuya cover."""
return import_from_yaml(hass, config, PLATFORM)
class TuyaCache: class TuyaCache:
"""Cache wrapper for pytuya.PytuyaDevice""" """Cache wrapper for pytuya.TuyaDevice"""
def __init__(self, device): def __init__(self, device):
"""Initialize the cache.""" """Initialize the cache."""
self._cached_status = '' self._cached_status = ""
self._cached_status_time = 0 self._cached_status_time = 0
self._device = device self._device = device
self._lock = Lock() self._lock = Lock()
@@ -132,38 +121,49 @@ class TuyaCache:
return self._device.id return self._device.id
def __get_status(self): def __get_status(self):
#_LOGGER.info("running def __get_status from cover") # _LOGGER.info("running def __get_status from cover")
for i in range(5): for i in range(5):
try: try:
status = self._device.status() status = self._device.status()
return status return status
except Exception: except Exception:
print('Failed to update status of device [{}]'.format(self._device.address)) print(
"Failed to update status of device [{}]".format(
self._device.address
)
)
sleep(1.0) sleep(1.0)
if i+1 == 3: if i + 1 == 3:
_LOGGER.error("Failed to update status of device %s", self._device.address ) _LOGGER.error(
# return None "Failed to update status of device %s", self._device.address
)
# return None
raise ConnectionError("Failed to update status .") raise ConnectionError("Failed to update status .")
def set_dps(self, state, switchid): def set_dps(self, state, dps_index):
#_LOGGER.info("running def set_dps 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_dps 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 dps_index=%s", state, dps_index)
return self._device.set_dps(state, switchid) return self._device.set_dps(state, dps_index)
except Exception: except Exception:
print('Failed to set status of device [{}]'.format(self._device.address)) print(
if i+1 == 3: "Failed to set status of device [{}]".format(self._device.address)
_LOGGER.error("Failed to set status of device %s", self._device.address ) )
if i + 1 == 3:
_LOGGER.error(
"Failed to set status of device %s", self._device.address
)
return return
# raise ConnectionError("Failed to set status.")
# raise ConnectionError("Failed to set status.")
def status(self): def status(self):
"""Get state of Tuya switch and cache the results.""" """Get state of Tuya switch and cache the results."""
#_LOGGER.info("running def status(self) from cover") # _LOGGER.info("running def status(self) from cover")
self._lock.acquire() self._lock.acquire()
try: try:
now = time() now = time()
@@ -178,21 +178,41 @@ class TuyaCache:
class LocaltuyaCover(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, friendly_name, switchid, open_cmd, close_cmd, stop_cmd):
self._device = device self._device = device
self._available = False self._available = False
self._name = friendly_name self._name = friendly_name
self._friendly_name = friendly_name
self._icon = icon
self._switch_id = switchid self._switch_id = switchid
self._status = self._device.status() self._status = None
self._state = self._status['dps'][self._switch_id] self._state = None
self._position = 50 self._position = 50
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(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) #_LOGGER.info("running def __init__ of LocaltuyaCover(CoverEntity) 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
def device_info(self):
return {
"identifiers": {
# Serial numbers are unique identifiers within a specific domain
("LocalTuya", f"local_{self._device.unique_id}")
},
"name": self._device._device.friendly_name,
"manufacturer": "Tuya generic",
"model": "SmartCover",
"sw_version": "3.3",
}
@property
def unique_id(self):
"""Return unique device identifier."""
return f"local_{self._device.unique_id}_{self._switch_id}"
@property @property
def name(self): def name(self):
@@ -214,11 +234,6 @@ class LocaltuyaCover(CoverEntity):
"""Get name of stop command.""" """Get name of stop command."""
return self._stop_cmd return self._stop_cmd
@property
def unique_id(self):
"""Return unique device identifier."""
return f"local_{self._device.unique_id}"
@property @property
def available(self): def available(self):
"""Return if device is available or not.""" """Return if device is available or not."""
@@ -227,75 +242,73 @@ class LocaltuyaCover(CoverEntity):
@property @property
def supported_features(self): def supported_features(self):
"""Flag supported features.""" """Flag supported features."""
supported_features = SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP | SUPPORT_SET_POSITION supported_features = (
SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP | SUPPORT_SET_POSITION
)
return supported_features return supported_features
@property
def icon(self):
"""Return the icon."""
return self._icon
@property @property
def current_cover_position(self): def current_cover_position(self):
#self.update() # self.update()
#state = self._state # state = self._state
# _LOGGER.info("curr_pos() : %i", self._position) # _LOGGER.info("curr_pos() : %i", self._position)
#print('curr_pos() : state [{}]'.format(state)) # print('curr_pos() : state [{}]'.format(state))
return self._position return self._position
@property @property
def is_opening(self): def is_opening(self):
#self.update() # self.update()
state = self._state state = self._state
#print('is_opening() : state [{}]'.format(state)) # print('is_opening() : state [{}]'.format(state))
if state == 'on': if state == "on":
return True return True
return False return False
@property @property
def is_closing(self): def is_closing(self):
#self.update() # self.update()
state = self._state state = self._state
#print('is_closing() : state [{}]'.format(state)) # print('is_closing() : state [{}]'.format(state))
if state == 'off': if state == "off":
return True return True
return False return False
@property @property
def is_closed(self): def is_closed(self):
"""Return if the cover is closed or not.""" """Return if the cover is closed or not."""
#_LOGGER.info("running is_closed from cover") # _LOGGER.info("running is_closed from cover")
#self.update() # self.update()
state = self._state state = self._state
#print('is_closed() : state [{}]'.format(state)) # print('is_closed() : state [{}]'.format(state))
if state == 'off': if state == "off":
return False return False
if state == 'on': if state == "on":
return True return True
return None return None
def set_cover_position(self, **kwargs): def set_cover_position(self, **kwargs):
#_LOGGER.info("running set_cover_position from cover") # _LOGGER.info("running set_cover_position from cover")
"""Move the cover to a specific position.""" """Move the cover to a specific position."""
newpos = float(kwargs["position"]) newpos = float(kwargs["position"])
# _LOGGER.info("Set new pos: %f", newpos) # _LOGGER.info("Set new pos: %f", newpos)
currpos = self.current_cover_position currpos = self.current_cover_position
posdiff = abs(newpos - currpos) posdiff = abs(newpos - currpos)
# 25 sec corrisponde alla chiusura/apertura completa # 25 sec corrisponde alla chiusura/apertura completa
mydelay = posdiff / 2.0 mydelay = posdiff / 2.0
if newpos > currpos: if newpos > currpos:
# _LOGGER.info("Opening to %f: delay %f", newpos, mydelay ) # _LOGGER.info("Opening to %f: delay %f", newpos, mydelay )
self.open_cover() self.open_cover()
else: else:
# _LOGGER.info("Closing to %f: delay %f", newpos, mydelay ) # _LOGGER.info("Closing to %f: delay %f", newpos, mydelay )
self.close_cover() self.close_cover()
sleep( mydelay ) sleep(mydelay)
self.stop_cover() self.stop_cover()
self._position = 50 # newpos self._position = 50 # newpos
# self._state = 'on'
# self._device._device.open_cover() # self._state = 'on'
# self._device._device.open_cover()
def open_cover(self, **kwargs): def open_cover(self, **kwargs):
"""Open the cover.""" """Open the cover."""
@@ -305,7 +318,7 @@ class LocaltuyaCover(CoverEntity):
# 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_dps 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_dps(self._close_cmd, self._switch_id) self._device.set_dps(self._close_cmd, self._switch_id)
@@ -313,7 +326,7 @@ class LocaltuyaCover(CoverEntity):
# 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_dps(self._stop_cmd, self._switch_id) self._device.set_dps(self._stop_cmd, self._switch_id)
# self._state = 'stop' # self._state = 'stop'
@@ -321,11 +334,11 @@ class LocaltuyaCover(CoverEntity):
def update(self): def update(self):
"""Get state of Tuya switch.""" """Get state of Tuya switch."""
#_LOGGER.info("running update(self) from cover") # _LOGGER.info("running update(self) from cover")
try: try:
self._status = self._device.status() self._status = self._device.status()
self._state = self._status['dps'][self._switch_id] self._state = self._status["dps"][self._switch_id]
#print('update() : state [{}]'.format(self._state)) # print('update() : state [{}]'.format(self._state))
except Exception: except Exception:
self._available = False self._available = False
else: else:

114
custom_components/localtuya/fan.py
View File

@@ -15,84 +15,73 @@ fan:
""" """
import logging import logging
import requests
import voluptuous as vol import voluptuous as vol
from homeassistant.components.fan import (SPEED_LOW, SPEED_MEDIUM, SPEED_HIGH, from homeassistant.components.fan import (
FanEntity, SUPPORT_SET_SPEED, FanEntity,
SUPPORT_OSCILLATE, SUPPORT_DIRECTION, PLATFORM_SCHEMA) PLATFORM_SCHEMA,
SPEED_LOW,
from homeassistant.const import STATE_OFF SPEED_MEDIUM,
"""from . import DATA_TUYA, TuyaDevice""" SPEED_HIGH,
from homeassistant.const import (CONF_HOST, CONF_ID, CONF_FRIENDLY_NAME, CONF_ICON, CONF_NAME) SUPPORT_SET_SPEED,
SUPPORT_OSCILLATE,
SUPPORT_DIRECTION,
)
from homeassistant.const import CONF_ID, CONF_FRIENDLY_NAME, STATE_OFF
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from . import BASE_PLATFORM_SCHEMA, prepare_setup_entities, import_from_yaml
from .pytuya import TuyaDevice
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = 'localtuyafan' PLATFORM = "fan"
REQUIREMENTS = ['pytuya>=8.0.0'] PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(BASE_PLATFORM_SCHEMA)
CONF_DEVICE_ID = 'device_id'
CONF_LOCAL_KEY = 'local_key'
CONF_PROTOCOL_VERSION = 'protocol_version'
DEFAULT_ID = '1'
DEFAULT_PROTOCOL_VERSION = 3.3
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ def flow_schema(dps):
vol.Optional(CONF_ICON): cv.icon, """Return schema used in config flow."""
vol.Required(CONF_HOST): cv.string, return {}
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(CONF_LOCAL_KEY): cv.string,
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_FRIENDLY_NAME): cv.string,
vol.Required(CONF_PROTOCOL_VERSION, default=DEFAULT_PROTOCOL_VERSION): vol.Coerce(float),
vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string,
})
def setup_platform(hass, config, add_entities, discovery_info=None): async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up of the Tuya switch.""" """Setup a Tuya fan based on a config entry."""
from . import pytuya device, entities_to_setup = prepare_setup_entities(
fans = [] config_entry, PLATFORM
pytuyadevice = pytuya.PytuyaDevice(
config_entry.data[CONF_DEVICE_ID],
config_entry.data[CONF_HOST],
config_entry.data[CONF_LOCAL_KEY],
config_entry.data[CONF_FRIENDLY_NAME],
config_entry.data[CONF_NAME],
) )
pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION))) if not entities_to_setup:
_LOGGER.debug("localtuya fan: setup_platform: %s", pytuyadevice) return
fan_device = pytuyadevice fans = []
fans.append(
for device_config in entities_to_setup:
fans.append(
LocaltuyaFan( LocaltuyaFan(
fan_device, device,
config.get(CONF_NAME), device_config[CONF_FRIENDLY_NAME],
config.get(CONF_FRIENDLY_NAME), device_config[CONF_ID],
config.get(CONF_ICON),
config.get(CONF_ID),
) )
) )
_LOGGER.info("Setup localtuya fan %s with device ID %s ", config.get(CONF_FRIENDLY_NAME), config.get(CONF_DEVICE_ID) )
async_add_entities(fans, True)
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up of the Tuya fan."""
return import_from_yaml(hass, config, PLATFORM)
add_entities(fans, True)
class LocaltuyaFan(FanEntity): class LocaltuyaFan(FanEntity):
"""Representation of a Tuya fan.""" """Representation of a Tuya fan."""
# def __init__(self, hass, name: str, supported_features: int) -> None: def __init__(self, device, friendly_name, switchid):
def __init__(self, device, name, friendly_name, icon, switchid):
"""Initialize the entity.""" """Initialize the entity."""
self._device = device self._device = device
self._name = friendly_name self._name = friendly_name
self._available = False self._available = False
self._friendly_name = friendly_name self._friendly_name = friendly_name
self._icon = icon
self._switch_id = switchid self._switch_id = switchid
self._status = self._device.status() self._status = self._device.status()
self._state = False self._state = False
@@ -103,7 +92,7 @@ class LocaltuyaFan(FanEntity):
@property @property
def oscillating(self): def oscillating(self):
"""Return current oscillating status.""" """Return current oscillating status."""
#if self._speed == STATE_OFF: # if self._speed == STATE_OFF:
# return False # return False
_LOGGER.debug("localtuya fan: oscillating = %s", self._oscillating) _LOGGER.debug("localtuya fan: oscillating = %s", self._oscillating)
return self._oscillating return self._oscillating
@@ -171,7 +160,7 @@ class LocaltuyaFan(FanEntity):
def oscillate(self, oscillating: bool) -> None: def oscillate(self, oscillating: bool) -> None:
"""Set oscillation.""" """Set oscillation."""
self._oscillating = oscillating self._oscillating = oscillating
self._device.set_value('8', oscillating) self._device.set_value("8", oscillating)
self.schedule_update_ha_state() self.schedule_update_ha_state()
# @property # @property
@@ -182,8 +171,7 @@ class LocaltuyaFan(FanEntity):
@property @property
def unique_id(self): def unique_id(self):
"""Return unique device identifier.""" """Return unique device identifier."""
_LOGGER.debug("localtuya fan unique_id = %s", self._device) return f"local_{self._device.id}_{self._switch_id}"
return self._device.id
@property @property
def available(self): def available(self):
@@ -203,20 +191,20 @@ class LocaltuyaFan(FanEntity):
try: try:
status = self._device.status() status = self._device.status()
_LOGGER.debug("localtuya fan: status = %s", status) _LOGGER.debug("localtuya fan: status = %s", status)
self._state = status['dps']['1'] self._state = status["dps"]["1"]
if status['dps']['1'] == False: if status["dps"]["1"] == False:
self._speed = STATE_OFF self._speed = STATE_OFF
elif int(status['dps']['2']) == 1: elif int(status["dps"]["2"]) == 1:
self._speed = SPEED_LOW self._speed = SPEED_LOW
elif int(status['dps']['2']) == 2: elif int(status["dps"]["2"]) == 2:
self._speed = SPEED_MEDIUM self._speed = SPEED_MEDIUM
elif int(status['dps']['2']) == 3: elif int(status["dps"]["2"]) == 3:
self._speed = SPEED_HIGH self._speed = SPEED_HIGH
# self._speed = status['dps']['2'] # self._speed = status['dps']['2']
self._oscillating = status['dps']['8'] self._oscillating = status["dps"]["8"]
success = True success = True
except ConnectionError: except ConnectionError:
if i+1 == 3: if i + 1 == 3:
success = False success = False
raise ConnectionError("Failed to update status.") raise ConnectionError("Failed to update status.")
self._available = success self._available = success

319
custom_components/localtuya/light.py
View File

@@ -12,84 +12,78 @@ light:
friendly_name: This Light friendly_name: This Light
protocol_version: 3.3 protocol_version: 3.3
""" """
import voluptuous as vol import socket
from homeassistant.const import (CONF_HOST, CONF_ID, CONF_SWITCHES, CONF_FRIENDLY_NAME, CONF_ICON, CONF_NAME) import logging
import homeassistant.helpers.config_validation as cv
from time import time, sleep from time import time, sleep
from threading import Lock from threading import Lock
import logging
from homeassistant.const import (
CONF_ID,
CONF_FRIENDLY_NAME,
)
from homeassistant.components.light import ( from homeassistant.components.light import (
LightEntity,
PLATFORM_SCHEMA,
ATTR_BRIGHTNESS, ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP, ATTR_COLOR_TEMP,
ATTR_HS_COLOR, ATTR_HS_COLOR,
SUPPORT_BRIGHTNESS, SUPPORT_BRIGHTNESS,
SUPPORT_COLOR, SUPPORT_COLOR,
SUPPORT_COLOR_TEMP, SUPPORT_COLOR_TEMP,
LightEntity,
PLATFORM_SCHEMA
) )
from homeassistant.util import color as colorutil from homeassistant.util import color as colorutil
import socket
REQUIREMENTS = ['pytuya>=8.0.0'] from . import BASE_PLATFORM_SCHEMA, import_from_yaml, prepare_setup_entities
from .pytuya import TuyaDevice
_LOGGER = logging.getLogger(__name__)
PLATFORM = "light"
CONF_DEVICE_ID = 'device_id'
CONF_LOCAL_KEY = 'local_key'
CONF_PROTOCOL_VERSION = 'protocol_version'
# IMPORTANT, id is used as key for state and turning on and off, 1 was fine switched apparently but my bulbs need 20, other feature attributes count up from this, e.g. 21 mode, 22 brightnes etc, see my pytuya modification.
DEFAULT_ID = '1'
DEFAULT_PROTOCOL_VERSION = 3.3
MIN_MIRED = 153 MIN_MIRED = 153
MAX_MIRED = 370 MAX_MIRED = 370
UPDATE_RETRY_LIMIT = 3 UPDATE_RETRY_LIMIT = 3
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(BASE_PLATFORM_SCHEMA)
vol.Optional(CONF_ICON): cv.icon,
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_DEVICE_ID): cv.string, def flow_schema(dps):
vol.Required(CONF_LOCAL_KEY): cv.string, """Return schema used in config flow."""
vol.Required(CONF_NAME): cv.string, return {}
vol.Required(CONF_FRIENDLY_NAME): cv.string,
vol.Required(CONF_PROTOCOL_VERSION, default=DEFAULT_PROTOCOL_VERSION): vol.Coerce(float),
vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string, async def async_setup_entry(hass, config_entry, async_add_entities):
}) """Setup a Tuya switch based on a config entry."""
log = logging.getLogger(__name__) device, entities_to_setup = prepare_setup_entities(
log.setLevel(level=logging.DEBUG) # Debug hack! config_entry, PLATFORM
)
if not entities_to_setup:
return
lights = []
for device_config in entities_to_setup:
lights.append(
LocaltuyaLight(
TuyaCache(device),
device_config[CONF_FRIENDLY_NAME],
device_config[CONF_ID],
)
)
async_add_entities(lights, True)
def setup_platform(hass, config, add_devices, discovery_info=None): def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up of the Tuya switch.""" """Set up of the Tuya switch."""
from . import pytuya return import_from_yaml(hass, config, PLATFORM)
lights = []
pytuyadevice = pytuya.PytuyaDevice(
config_entry.data[CONF_DEVICE_ID],
config_entry.data[CONF_HOST],
config_entry.data[CONF_LOCAL_KEY],
config_entry.data[CONF_FRIENDLY_NAME],
config_entry.data[CONF_NAME],
)
pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION)))
bulb_device = TuyaCache(pytuyadevice)
lights.append(
LocaltuyaLight(
bulb_device,
config.get(CONF_NAME),
config.get(CONF_FRIENDLY_NAME),
config.get(CONF_ICON),
config.get(CONF_ID)
)
)
add_devices(lights)
class TuyaCache: class TuyaCache:
"""Cache wrapper for pytuya.PytuyaDevices""" """Cache wrapper for pytuya.TuyaDevices"""
def __init__(self, device): def __init__(self, device):
"""Initialize the cache.""" """Initialize the cache."""
self._cached_status = '' self._cached_status = ""
self._cached_status_time = 0 self._cached_status_time = 0
self._device = device self._device = device
self._lock = Lock() self._lock = Lock()
@@ -102,116 +96,49 @@ class TuyaCache:
def __get_status(self, switchid): def __get_status(self, switchid):
for _ in range(UPDATE_RETRY_LIMIT): for _ in range(UPDATE_RETRY_LIMIT):
try: try:
status = self._device.status()['dps'][switchid] return self._device.status()["dps"][switchid]
return status except (ConnectionError, socket.timeout):
except ConnectionError:
pass pass
except socket.timeout: _LOGGER.warning("Failed to get status after %d tries", UPDATE_RETRY_LIMIT)
pass
log.warn(
"Failed to get status after {} tries".format(UPDATE_RETRY_LIMIT))
def set_dps(self, state, switchid): def set_dps(self, state, dps_index):
"""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_dps(state, switchid) return self._device.set_dps(state, dps_index)
except ConnectionError: except ConnectionError:
pass pass
except socket.timeout: except socket.timeout:
pass pass
log.warn( _LOGGER.warning("Failed to set status after %d tries", UPDATE_RETRY_LIMIT)
"Failed to set status after {} tries".format(UPDATE_RETRY_LIMIT))
def status(self, switchid): def status(self, switchid):
"""Get state of Tuya switch and cache the results.""" """Get state of Tuya switch and cache the results."""
self._lock.acquire() with self._lock:
try:
now = time() now = time()
if not self._cached_status or now - self._cached_status_time > 15: if not self._cached_status or now - self._cached_status_time > 15:
sleep(0.5) sleep(0.5)
self._cached_status = self.__get_status(switchid) self._cached_status = self.__get_status(switchid)
self._cached_status_time = time() self._cached_status_time = time()
return self._cached_status return self._cached_status
finally:
self._lock.release()
def cached_status(self): def cached_status(self):
return self._cached_status return self._cached_status
def support_color(self): def state(self):
return self._device.support_color()
def support_color_temp(self):
return self._device.support_color_temp()
def brightness(self):
for _ in range(UPDATE_RETRY_LIMIT):
try:
return self._device.brightness()
except ConnectionError:
pass
except KeyError:
return "999"
except socket.timeout:
pass
log.warn(
"Failed to get brightness after {} tries".format(UPDATE_RETRY_LIMIT))
def color_temp(self):
for _ in range(UPDATE_RETRY_LIMIT):
try:
return self._device.colourtemp()
except ConnectionError:
pass
except KeyError:
return "999"
except socket.timeout:
pass
log.warn(
"Failed to get color temp after {} tries".format(UPDATE_RETRY_LIMIT))
def set_brightness(self, brightness):
for _ in range(UPDATE_RETRY_LIMIT):
try:
return self._device.set_brightness(brightness)
except ConnectionError:
pass
except KeyError:
pass
except socket.timeout:
pass
log.warn(
"Failed to set brightness after {} tries".format(UPDATE_RETRY_LIMIT))
def set_color_temp(self, color_temp):
for _ in range(UPDATE_RETRY_LIMIT):
try:
return self._device.set_colourtemp(color_temp)
except ConnectionError:
pass
except KeyError:
pass
except socket.timeout:
pass
log.warn(
"Failed to set color temp after {} tries".format(UPDATE_RETRY_LIMIT))
def state(self):
self._device.state(); self._device.state();
def turn_on(self):
self._device.turn_on();
def turn_off(self):
self._device.turn_off();
class LocaltuyaLight(LightEntity): class LocaltuyaLight(LightEntity):
"""Representation of a Tuya switch.""" """Representation of a Tuya switch."""
DPS_INDEX_ON = '1'
DPS_INDEX_MODE = '2'
DPS_INDEX_BRIGHTNESS = '3'
DPS_INDEX_COLOURTEMP = '4'
DPS_INDEX_COLOUR = '5'
def __init__(self, device, name, friendly_name, icon, bulbid): def __init__(self, device, friendly_name, bulbid):
"""Initialize the Tuya switch.""" """Initialize the Tuya switch."""
self._device = device self._device = device
self._available = False self._available = False
@@ -219,9 +146,11 @@ class LocaltuyaLight(LightEntity):
self._state = False self._state = False
self._brightness = 127 self._brightness = 127
self._color_temp = 127 self._color_temp = 127
self._icon = icon
self._bulb_id = bulbid self._bulb_id = bulbid
def state(self):
self._device.state()
@property @property
def name(self): def name(self):
"""Get name of Tuya switch.""" """Get name of Tuya switch."""
@@ -242,11 +171,6 @@ class LocaltuyaLight(LightEntity):
"""Check if Tuya switch is on.""" """Check if Tuya switch is on."""
return self._state return self._state
@property
def icon(self):
"""Return the icon."""
return self._icon
def update(self): def update(self):
"""Get state of Tuya switch.""" """Get state of Tuya switch."""
try: try:
@@ -260,12 +184,12 @@ class LocaltuyaLight(LightEntity):
status = self._device.status(self._bulb_id) status = self._device.status(self._bulb_id)
self._state = status self._state = status
try: try:
brightness = int(self._device.brightness()) brightness = int(self._device.brightness())
if brightness > 254: if brightness > 254:
brightness = 255 brightness = 255
if brightness < 25: if brightness < 25:
brightness = 25 brightness = 25
self._brightness = brightness self._brightness = brightness
except TypeError: except TypeError:
pass pass
self._color_temp = self._device.color_temp() self._color_temp = self._device.color_temp()
@@ -275,16 +199,16 @@ class LocaltuyaLight(LightEntity):
"""Return the brightness of the light.""" """Return the brightness of the light."""
return self._brightness return self._brightness
# @property # @property
# def hs_color(self): # def hs_color(self):
# """Return the hs_color of the light.""" # """Return the hs_color of the light."""
# return (self._device.color_hsv()[0],self._device.color_hsv()[1]) # return (self._device.color_hsv()[0],self._device.color_hsv()[1])
@property @property
def color_temp(self): def color_temp(self):
"""Return the color_temp of the light.""" """Return the color_temp of the light."""
try: try:
return int(MAX_MIRED - (((MAX_MIRED - MIN_MIRED) / 255) * self._color_temp)) return int(MAX_MIRED - (((MAX_MIRED - MIN_MIRED) / 255) * self._color_temp))
except TypeError: except TypeError:
pass pass
@@ -300,18 +224,22 @@ class LocaltuyaLight(LightEntity):
def turn_on(self, **kwargs): def turn_on(self, **kwargs):
"""Turn on or control the light.""" """Turn on or control the light."""
log.debug("Turning on, state: " + str(self._device.cached_status())) _LOGGER.debug("Turning on, state: %s", self._device.cached_status())
if not self._device.cached_status(): if not self._device.cached_status():
self._device.set_dps(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:
converted_brightness = 25 converted_brightness = 25
self._device.set_brightness(converted_brightness) self.set_brightness(converted_brightness)
if ATTR_HS_COLOR in kwargs: if ATTR_HS_COLOR in kwargs:
raise ValueError(" TODO implement RGB from HS") raise ValueError(" TODO implement RGB from HS")
if ATTR_COLOR_TEMP in kwargs: if ATTR_COLOR_TEMP in kwargs:
color_temp = int(255 - (255 / (MAX_MIRED - MIN_MIRED)) * (int(kwargs[ATTR_COLOR_TEMP]) - MIN_MIRED)) color_temp = int(
255
- (255 / (MAX_MIRED - MIN_MIRED))
* (int(kwargs[ATTR_COLOR_TEMP]) - MIN_MIRED)
)
self._device.set_color_temp(color_temp) self._device.set_color_temp(color_temp)
def turn_off(self, **kwargs): def turn_off(self, **kwargs):
@@ -324,5 +252,88 @@ class LocaltuyaLight(LightEntity):
supports = SUPPORT_BRIGHTNESS supports = SUPPORT_BRIGHTNESS
if self._device.color_temp() != "999": if self._device.color_temp() != "999":
supports = supports | SUPPORT_COLOR supports = supports | SUPPORT_COLOR
#supports = supports | SUPPORT_COLOR_TEMP # supports = supports | SUPPORT_COLOR_TEMP
return supports return supports
def support_color(self):
return supported_features() & SUPPORT_COLOR
def support_color_temp(self):
return supported_features() & SUPPORT_COLOR_TEMP
def color_temp(self):
for _ in range(UPDATE_RETRY_LIMIT):
try:
return self._state[self.DPS][self.DPS_INDEX_COLOURTEMP]
except ConnectionError:
pass
except KeyError:
return "999"
except socket.timeout:
pass
log.warn(
"Failed to get color temp after {} tries".format(UPDATE_RETRY_LIMIT))
def set_color_temp(self, color_temp):
for _ in range(UPDATE_RETRY_LIMIT):
try:
if not 0 <= colourtemp <= 255:
raise ValueError("The colour temperature needs to be between 0 and 255.")
self._device.set_dps(colourtemp, self.DPS_INDEX_COLOURTEMP)
except ConnectionError:
pass
except KeyError:
pass
except socket.timeout:
pass
log.warn(
"Failed to set color temp after {} tries".format(UPDATE_RETRY_LIMIT))
def set_brightness(self, brightness):
for _ in range(UPDATE_RETRY_LIMIT):
try:
if not 25 <= brightness <= 255:
raise ValueError("The brightness needs to be between 25 and 255.")
self._device.set_dps(brightness, self.DPS_INDEX_BRIGHTNESS)
except ConnectionError:
pass
except KeyError:
pass
except socket.timeout:
pass
log.warn(
"Failed to set brightness after {} tries".format(UPDATE_RETRY_LIMIT))
@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 TuyaDevice._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 TuyaDevice._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)

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

@@ -11,7 +11,7 @@
For more information see https://github.com/clach04/python-tuya For more information see https://github.com/clach04/python-tuya
Classes Classes
PytuyaDevice(dev_id, address, local_key=None) TuyaDevice(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.
@@ -20,8 +20,7 @@
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_dps(on, switch=1) # Set the of the device to 'on' or 'off' (bool) set_dps(on, dps_index) # Set int value of any dps index.
set_value(index, value) # Set int value of any index.
set_timer(num_secs): set_timer(num_secs):
@@ -173,8 +172,8 @@ payload_dict = {
} }
class PytuyaDevice(object): class TuyaDevice(object):
def __init__(self, dev_id, address, local_key, friendly_name, name = "", connection_timeout=10): def __init__(self, dev_id, address, local_key=None, connection_timeout=10):
""" """
Represents a Tuya device. Represents a Tuya device.
@@ -363,20 +362,20 @@ class PytuyaDevice(object):
return result return result
def set_dps(self, value, dpsIndex): def set_dps(self, value, dps_index):
""" """
Set int value of any index. Set value (may be any type: bool, int or string) of any dps index.
Args: Args:
dpsIndex(int): dps index to set dps_index(int): dps index to set
value: new value for the dps index value: new value for the dps index
""" """
# open device, send request, then close connection # open device, send request, then close connection
if isinstance(dpsIndex, int): if isinstance(dps_index, int):
dpsIndex = str(dpsIndex) # index and payload is a string dps_index = str(dps_index) # index and payload is a string
payload = self.generate_payload(SET, { payload = self.generate_payload(SET, {
dpsIndex: value}) dps_index: value})
data = self._send_receive(payload) data = self._send_receive(payload)
log.debug('set_dps received data=%r', data) log.debug('set_dps received data=%r', data)

114
custom_components/localtuya/switch.py
View File

@@ -22,51 +22,48 @@ switch:
sw02: sw02:
name: usb_plug name: usb_plug
friendly_name: USB Plug friendly_name: USB Plug
id: 7 id: 7
""" """
import logging import logging
import voluptuous as vol
from homeassistant.components.switch import SwitchEntity, PLATFORM_SCHEMA
from homeassistant.const import (
CONF_HOST,
CONF_ID,
CONF_ENTITIES,
CONF_SWITCHES,
CONF_DEVICE_ID,
CONF_FRIENDLY_NAME,
CONF_ICON,
CONF_NAME,
CONF_PLATFORM,
)
from homeassistant.config_entries import SOURCE_IMPORT
import homeassistant.helpers.config_validation as cv
from time import time, sleep from time import time, sleep
from threading import Lock from threading import Lock
from . import pytuya import voluptuous as vol
from homeassistant.components.switch import (
SwitchEntity,
PLATFORM_SCHEMA,
)
from homeassistant.const import (
CONF_ID,
CONF_SWITCHES,
CONF_FRIENDLY_NAME,
CONF_NAME,
)
import homeassistant.helpers.config_validation as cv
from . import BASE_PLATFORM_SCHEMA, prepare_setup_entities, import_from_yaml
from .const import ( from .const import (
ATTR_CURRENT, ATTR_CURRENT,
ATTR_CURRENT_CONSUMPTION, ATTR_CURRENT_CONSUMPTION,
ATTR_VOLTAGE, ATTR_VOLTAGE,
CONF_LOCAL_KEY,
CONF_PROTOCOL_VERSION,
CONF_CURRENT, CONF_CURRENT,
CONF_CURRENT_CONSUMPTION, CONF_CURRENT_CONSUMPTION,
CONF_VOLTAGE, CONF_VOLTAGE,
DOMAIN,
) )
from .pytuya import TuyaDevice
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ['pytuya>=8.0.0']
PLATFORM = "switch"
DEFAULT_ID = "1" DEFAULT_ID = "1"
DEFAULT_PROTOCOL_VERSION = 3.3
# TODO: This will eventully merge with flow_schema
SWITCH_SCHEMA = vol.Schema( SWITCH_SCHEMA = vol.Schema(
{ {
vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string, vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string,
vol.Required(CONF_NAME): cv.string, vol.Optional(CONF_NAME): cv.string, # Deprecated: not used
vol.Required(CONF_FRIENDLY_NAME): cv.string, vol.Required(CONF_FRIENDLY_NAME): cv.string,
vol.Optional(CONF_CURRENT, default="-1"): cv.string, vol.Optional(CONF_CURRENT, default="-1"): cv.string,
vol.Optional(CONF_CURRENT_CONSUMPTION, default="-1"): cv.string, vol.Optional(CONF_CURRENT_CONSUMPTION, default="-1"): cv.string,
@@ -74,18 +71,8 @@ SWITCH_SCHEMA = vol.Schema(
} }
) )
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(BASE_PLATFORM_SCHEMA).extend(
{ {
vol.Optional(CONF_ICON): cv.icon,
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(CONF_LOCAL_KEY): cv.string,
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_FRIENDLY_NAME): cv.string,
vol.Required(
CONF_PROTOCOL_VERSION, default=DEFAULT_PROTOCOL_VERSION
): vol.Coerce(float),
vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string,
vol.Optional(CONF_CURRENT, default="-1"): cv.string, vol.Optional(CONF_CURRENT, default="-1"): cv.string,
vol.Optional(CONF_CURRENT_CONSUMPTION, default="-1"): cv.string, vol.Optional(CONF_CURRENT_CONSUMPTION, default="-1"): cv.string,
vol.Optional(CONF_VOLTAGE, default="-1"): cv.string, vol.Optional(CONF_VOLTAGE, default="-1"): cv.string,
@@ -94,44 +81,28 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
) )
DPS_FIELDS = [ def flow_schema(dps):
vol.Optional(CONF_CURRENT), """Return schema used in config flow."""
vol.Optional(CONF_CURRENT_CONSUMPTION), return {
vol.Optional(CONF_VOLTAGE), vol.Optional(CONF_CURRENT): vol.In(dps),
] vol.Optional(CONF_CURRENT_CONSUMPTION): vol.In(dps),
vol.Optional(CONF_VOLTAGE): vol.In(dps),
}
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(hass, config_entry, async_add_entities):
"""Setup a Tuya switch based on a config entry.""" """Setup a Tuya switch based on a config entry."""
device, entities_to_setup = prepare_setup_entities(
print('config_entry: [{}] '.format(config_entry.data)) config_entry, PLATFORM
)
if not entities_to_setup:
switches_to_setup = [
entity
for entity in config_entry.data[CONF_ENTITIES]
if entity[CONF_PLATFORM] == "switch"
]
if not switches_to_setup:
return return
switches = [] switches = []
pytuyadevice = pytuya.PytuyaDevice( for device_config in entities_to_setup:
config_entry.data[CONF_DEVICE_ID],
config_entry.data[CONF_HOST],
config_entry.data[CONF_LOCAL_KEY],
config_entry.data[CONF_FRIENDLY_NAME],
config_entry.data[CONF_NAME],
)
pytuyadevice.set_version(float(config_entry.data[CONF_PROTOCOL_VERSION]))
pytuyadevice.set_dpsUsed({})
print('switches_to_setup: [{}] '.format(switches_to_setup))
for device_config in switches_to_setup:
switches.append( switches.append(
LocaltuyaSwitch( LocaltuyaSwitch(
TuyaCache(pytuyadevice), TuyaCache(device),
device_config[CONF_FRIENDLY_NAME], device_config[CONF_FRIENDLY_NAME],
device_config[CONF_ID], device_config[CONF_ID],
device_config.get(CONF_CURRENT), device_config.get(CONF_CURRENT),
@@ -145,18 +116,11 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
def setup_platform(hass, config, add_devices, discovery_info=None): def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up of the Tuya switch.""" """Set up of the Tuya switch."""
config[CONF_PLATFORM] = "switch" return import_from_yaml(hass, config, PLATFORM)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
return True
class TuyaCache: class TuyaCache:
"""Cache wrapper for pytuya.PytuyaDevice""" """Cache wrapper for pytuya.TuyaDevice"""
def __init__(self, device): def __init__(self, device):
"""Initialize the cache.""" """Initialize the cache."""
@@ -189,13 +153,13 @@ class TuyaCache:
# return None # return None
raise ConnectionError("Failed to update status .") raise ConnectionError("Failed to update status .")
def set_dps(self, state, switchid): def set_dps(self, state, dps_index):
"""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_dps(state, switchid) return self._device.set_dps(state, dps_index)
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)

5
custom_components/localtuya/translations/en.json
View File

@@ -38,7 +38,10 @@
"friendly_name": "Friendly name", "friendly_name": "Friendly name",
"current": "Current", "current": "Current",
"current_consumption": "Current Consumption", "current_consumption": "Current Consumption",
"voltage": "Voltage" "voltage": "Voltage",
"open_cmd": "Open Command",
"close_cmd": "Close Command",
"stop_cmd": "Stop Command"
} }
} }
} }