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

Merge branch 'master' into pytuya_refactoring

Browse Source
This commit is contained in:
rospogrigio
2020-09-14 23:32:55 +02:00
parent 47b92b0178 e824f285fa
commit ce2461df2b
13 changed files with 811 additions and 429 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
custom_components/localtuya/__init__.py
View File

@@ -1 +1,88 @@
"""The Tuya local integration."""
"""The LocalTuya integration integration."""
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.core import HomeAssistant
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 . 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):
"""Set up the LocalTuya integration component."""
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up LocalTuya integration from a config entry."""
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
# Nothing is stored and no persistent connections exist, so nothing to do
return True

213
custom_components/localtuya/config_flow.py Normal file
View File

@@ -0,0 +1,213 @@
"""Config flow for LocalTuya integration integration."""
import logging
from importlib import import_module
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant.const import (
CONF_ENTITIES,
CONF_ID,
CONF_HOST,
CONF_DEVICE_ID,
CONF_NAME,
CONF_FRIENDLY_NAME,
CONF_PLATFORM,
CONF_SWITCHES,
)
from . import pytuya
from .const import ( # pylint: disable=unused-import
CONF_LOCAL_KEY,
CONF_PROTOCOL_VERSION,
DOMAIN,
PLATFORMS,
)
_LOGGER = logging.getLogger(__name__)
PLATFORM_TO_ADD = "platform_to_add"
NO_ADDITIONAL_PLATFORMS = "no_additional_platforms"
USER_SCHEMA = vol.Schema(
{
vol.Required(CONF_NAME): str,
vol.Required(CONF_HOST): str,
vol.Required(CONF_DEVICE_ID): str,
vol.Required(CONF_LOCAL_KEY): str,
vol.Required(CONF_PROTOCOL_VERSION, default="3.3"): vol.In(["3.1", "3.3"]),
}
)
PICK_ENTITY_SCHEMA = vol.Schema(
{vol.Required(PLATFORM_TO_ADD, default=PLATFORMS[0]): vol.In(PLATFORMS)}
)
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."""
return vol.Schema(
{
vol.Required(CONF_ID): vol.In(dps_strings),
vol.Required(CONF_FRIENDLY_NAME): str,
}
).extend(schema)
def strip_dps_values(user_input, dps_strings):
"""Remove values and keep only index for DPS config items."""
stripped = {}
for field, value in user_input.items():
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):
"""Validate the user input allows us to connect."""
pytuyadevice = pytuya.Device(
data[CONF_DEVICE_ID], data[CONF_HOST], data[CONF_LOCAL_KEY]
)
pytuyadevice.set_version(float(data[CONF_PROTOCOL_VERSION]))
pytuyadevice.set_dpsUsed({})
try:
data = await hass.async_add_executor_job(pytuyadevice.status)
except (ConnectionRefusedError, ConnectionResetError):
raise CannotConnect
except ValueError:
raise InvalidAuth
return dps_string_list(data["dps"])
class LocaltuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for LocalTuya integration."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Initialize a new LocaltuyaConfigFlow."""
self.basic_info = None
self.dps_strings = []
self.platform = None
self.platform_schema = None
self.entities = []
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
await self.async_set_unique_id(user_input[CONF_DEVICE_ID])
self._abort_if_unique_id_configured()
try:
self.basic_info = user_input
self.dps_strings = await validate_input(self.hass, user_input)
return await self.async_step_pick_entity_type()
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
return self.async_show_form(
step_id="user", data_schema=USER_SCHEMA, errors=errors
)
async def async_step_pick_entity_type(self, user_input=None):
"""Handle asking if user wants to add another entity."""
if user_input is not None:
if user_input.get(NO_ADDITIONAL_PLATFORMS):
config = {**self.basic_info, CONF_ENTITIES: self.entities}
return self.async_create_entry(title=config[CONF_NAME], data=config)
self._set_platform(user_input[PLATFORM_TO_ADD])
return await self.async_step_add_entity()
# Add a checkbox that allows bailing out from config flow iff at least one
# entity has been added
schema = PICK_ENTITY_SCHEMA
if self.platform is not None:
schema = schema.extend(
{vol.Required(NO_ADDITIONAL_PLATFORMS, default=True): bool}
)
return self.async_show_form(step_id="pick_entity_type", data_schema=schema)
async def async_step_add_entity(self, user_input=None):
"""Handle adding a new entity."""
errors = {}
if user_input is not None:
already_configured = any(
switch[CONF_ID] == user_input[CONF_ID] for switch in self.entities
)
if not already_configured:
user_input[CONF_PLATFORM] = self.platform
self.entities.append(strip_dps_values(user_input, self.dps_strings))
return await self.async_step_pick_entity_type()
errors["base"] = "entity_already_configured"
return self.async_show_form(
step_id="add_entity",
data_schema=platform_schema(self.dps_strings, self.platform_schema),
errors=errors,
description_placeholders={"platform": self.platform},
)
async def async_step_import(self, user_input):
"""Handle import from YAML."""
def _convert_entity(conf):
converted = {
CONF_ID: conf[CONF_ID],
CONF_FRIENDLY_NAME: conf[CONF_FRIENDLY_NAME],
CONF_PLATFORM: self.platform,
}
for field in self.platform_schema.keys():
converted[str(field)] = conf[field]
return converted
await self.async_set_unique_id(user_input[CONF_DEVICE_ID])
self._set_platform(user_input[CONF_PLATFORM])
if len(user_input.get(CONF_SWITCHES, [])) > 0:
for switch_conf in user_input[CONF_SWITCHES].values():
self.entities.append(_convert_entity(switch_conf))
else:
self.entities.append(_convert_entity(user_input))
config = {
CONF_NAME: f"{user_input[CONF_DEVICE_ID]} (import from configuration.yaml)",
CONF_HOST: user_input[CONF_HOST],
CONF_DEVICE_ID: user_input[CONF_DEVICE_ID],
CONF_LOCAL_KEY: user_input[CONF_LOCAL_KEY],
CONF_PROTOCOL_VERSION: user_input[CONF_PROTOCOL_VERSION],
CONF_ENTITIES: self.entities,
}
self._abort_if_unique_id_configured(updates=config)
return self.async_create_entry(title=config[CONF_NAME], data=config)
def _set_platform(self, platform):
integration_module = ".".join(__name__.split(".")[:-1])
self.platform = platform
self.platform_schema = import_module(
"." + platform, integration_module
).flow_schema(self.dps_strings)
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""
class InvalidAuth(exceptions.HomeAssistantError):
"""Error to indicate there is invalid auth."""

23
custom_components/localtuya/const.py Normal file
View File

@@ -0,0 +1,23 @@
"""Constants for localtuya integration."""
ATTR_CURRENT = 'current'
ATTR_CURRENT_CONSUMPTION = 'current_consumption'
ATTR_VOLTAGE = 'voltage'
CONF_LOCAL_KEY = "local_key"
CONF_PROTOCOL_VERSION = "protocol_version"
# switch
CONF_CURRENT = "current"
CONF_CURRENT_CONSUMPTION = "current_consumption"
CONF_VOLTAGE = "voltage"
# cover
CONF_OPEN_CMD = 'open_cmd'
CONF_CLOSE_CMD = 'close_cmd'
CONF_STOP_CMD = 'stop_cmd'
DOMAIN = "localtuya"
# Platforms in this list must support config flows
PLATFORMS = ["cover", "fan", "light", "switch"]

238
custom_components/localtuya/cover.py
View File

@@ -19,7 +19,8 @@ cover:
"""
import logging
import requests
from time import time, sleep
from threading import Lock
import voluptuous as vol
@@ -31,81 +32,77 @@ from homeassistant.components.cover import (
SUPPORT_STOP,
SUPPORT_SET_POSITION,
)
"""from . import DATA_TUYA, TuyaDevice"""
from homeassistant.components.cover import CoverEntity, PLATFORM_SCHEMA
from homeassistant.const import (CONF_HOST, CONF_ID, CONF_FRIENDLY_NAME, CONF_ICON, CONF_NAME)
from homeassistant.const import (
CONF_ID,
CONF_FRIENDLY_NAME,
)
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__)
DEFAULT_NAME = 'localtuyacover'
PLATFORM = "cover"
CONF_DEVICE_ID = 'device_id'
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,
})
DEFAULT_OPEN_CMD = "on"
DEFAULT_CLOSE_CMD = "off"
DEFAULT_STOP_CMD = "stop"
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up Tuya cover devices."""
from . import pytuya
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(BASE_PLATFORM_SCHEMA).extend(
{
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))
#_LOGGER.info("conf_close_cmd is %s", config.get(CONF_CLOSE_CMD))
#_LOGGER.info("conf_STOP_cmd is %s", config.get(CONF_STOP_CMD))
def flow_schema(dps):
"""Return schema used in config flow."""
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 = []
pytuyadevice = pytuya.TuyaDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY))
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(
for device_config in entities_to_setup:
dps[device_config[CONF_ID]] = None
covers.append(
LocaltuyaCover(
cover_device,
config.get(CONF_NAME),
config.get(CONF_FRIENDLY_NAME),
config.get(CONF_ICON),
config.get(CONF_ID),
config.get(CONF_OPEN_CMD),
config.get(CONF_CLOSE_CMD),
config.get(CONF_STOP_CMD),
TuyaCache(device),
device_config[CONF_FRIENDLY_NAME],
device_config[CONF_ID],
device_config.get(CONF_OPEN_CMD),
device_config.get(CONF_CLOSE_CMD),
device_config.get(CONF_STOP_CMD),
)
)
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.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) )
# 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.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:
@@ -113,7 +110,7 @@ class TuyaCache:
def __init__(self, device):
"""Initialize the cache."""
self._cached_status = ''
self._cached_status = ""
self._cached_status_time = 0
self._device = device
self._lock = Lock()
@@ -124,38 +121,49 @@ class TuyaCache:
return self._device.id
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):
try:
status = self._device.status()
return status
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)
if i+1 == 3:
_LOGGER.error("Failed to update status of device %s", self._device.address )
# return None
if i + 1 == 3:
_LOGGER.error(
"Failed to update status of device %s", self._device.address
)
# return None
raise ConnectionError("Failed to update status .")
def set_dps(self, state, dps_index):
#_LOGGER.info("running def set_dps from cover")
"""Change the Tuya switch status and clear the cache."""
self._cached_status = ''
self._cached_status = ""
self._cached_status_time = 0
for i in range(5):
try:
#_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, dps_index)
except Exception:
print('Failed to set status of device [{}]'.format(self._device.address))
if i+1 == 3:
_LOGGER.error("Failed to set status of device %s", self._device.address )
print(
"Failed to set status of device [{}]".format(self._device.address)
)
if i + 1 == 3:
_LOGGER.error(
"Failed to set status of device %s", self._device.address
)
return
# raise ConnectionError("Failed to set status.")
# raise ConnectionError("Failed to set status.")
def status(self):
"""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()
try:
now = time()
@@ -170,21 +178,23 @@ class TuyaCache:
class LocaltuyaCover(CoverEntity):
"""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._available = False
self._name = friendly_name
self._friendly_name = friendly_name
self._icon = icon
self._switch_id = switchid
self._status = self._device.status()
self._state = self._status['dps'][self._switch_id]
self._status = None
self._state = None
self._position = 50
self._open_cmd = open_cmd
self._close_cmd = close_cmd
self._stop_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 name(self):
@@ -219,75 +229,73 @@ class LocaltuyaCover(CoverEntity):
@property
def supported_features(self):
"""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
@property
def icon(self):
"""Return the icon."""
return self._icon
@property
def current_cover_position(self):
#self.update()
#state = self._state
# _LOGGER.info("curr_pos() : %i", self._position)
#print('curr_pos() : state [{}]'.format(state))
# self.update()
# state = self._state
# _LOGGER.info("curr_pos() : %i", self._position)
# print('curr_pos() : state [{}]'.format(state))
return self._position
@property
def is_opening(self):
#self.update()
# self.update()
state = self._state
#print('is_opening() : state [{}]'.format(state))
if state == 'on':
# print('is_opening() : state [{}]'.format(state))
if state == "on":
return True
return False
@property
def is_closing(self):
#self.update()
# self.update()
state = self._state
#print('is_closing() : state [{}]'.format(state))
if state == 'off':
# print('is_closing() : state [{}]'.format(state))
if state == "off":
return True
return False
@property
def is_closed(self):
"""Return if the cover is closed or not."""
#_LOGGER.info("running is_closed from cover")
#self.update()
# _LOGGER.info("running is_closed from cover")
# self.update()
state = self._state
#print('is_closed() : state [{}]'.format(state))
if state == 'off':
# print('is_closed() : state [{}]'.format(state))
if state == "off":
return False
if state == 'on':
if state == "on":
return True
return None
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."""
newpos = float(kwargs["position"])
# _LOGGER.info("Set new pos: %f", newpos)
# _LOGGER.info("Set new pos: %f", newpos)
currpos = self.current_cover_position
posdiff = abs(newpos - currpos)
# 25 sec corrisponde alla chiusura/apertura completa
# 25 sec corrisponde alla chiusura/apertura completa
mydelay = posdiff / 2.0
if newpos > currpos:
# _LOGGER.info("Opening to %f: delay %f", newpos, mydelay )
# _LOGGER.info("Opening to %f: delay %f", newpos, mydelay )
self.open_cover()
else:
# _LOGGER.info("Closing to %f: delay %f", newpos, mydelay )
# _LOGGER.info("Closing to %f: delay %f", newpos, mydelay )
self.close_cover()
sleep( mydelay )
sleep(mydelay)
self.stop_cover()
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):
"""Open the cover."""
@@ -297,7 +305,7 @@ class LocaltuyaCover(CoverEntity):
# self._device._device.open_cover()
def close_cover(self, **kwargs):
#_LOGGER.info("running close_cover from cover")
# _LOGGER.info("running close_cover from cover")
"""Close cover."""
#_LOGGER.info('about to set_dps from cover of off, %s', self._switch_id)
self._device.set_dps(self._close_cmd, self._switch_id)
@@ -305,7 +313,7 @@ class LocaltuyaCover(CoverEntity):
# self._device._device.close_cover()
def stop_cover(self, **kwargs):
#_LOGGER.info("running stop_cover from cover")
# _LOGGER.info("running stop_cover from cover")
"""Stop the cover."""
self._device.set_dps(self._stop_cmd, self._switch_id)
# self._state = 'stop'
@@ -313,11 +321,11 @@ class LocaltuyaCover(CoverEntity):
def update(self):
"""Get state of Tuya switch."""
#_LOGGER.info("running update(self) from cover")
# _LOGGER.info("running update(self) from cover")
try:
self._status = self._device.status()
self._state = self._status['dps'][self._switch_id]
#print('update() : state [{}]'.format(self._state))
self._state = self._status["dps"][self._switch_id]
# print('update() : state [{}]'.format(self._state))
except Exception:
self._available = False
else:

107
custom_components/localtuya/fan.py
View File

@@ -15,76 +15,72 @@ fan:
"""
import logging
import requests
import voluptuous as vol
from homeassistant.components.fan import (SPEED_LOW, SPEED_MEDIUM, SPEED_HIGH,
FanEntity, SUPPORT_SET_SPEED,
SUPPORT_OSCILLATE, SUPPORT_DIRECTION, PLATFORM_SCHEMA)
from homeassistant.const import STATE_OFF
"""from . import DATA_TUYA, TuyaDevice"""
from homeassistant.const import (CONF_HOST, CONF_ID, CONF_FRIENDLY_NAME, CONF_ICON, CONF_NAME)
from homeassistant.components.fan import (
FanEntity,
PLATFORM_SCHEMA,
SPEED_LOW,
SPEED_MEDIUM,
SPEED_HIGH,
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
from . import BASE_PLATFORM_SCHEMA, prepare_setup_entities, import_from_yaml
from .pytuya import TuyaDevice
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = 'localtuyafan'
PLATFORM = "fan"
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(BASE_PLATFORM_SCHEMA)
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,
})
def flow_schema(dps):
"""Return schema used in config flow."""
return {}
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up of the Tuya switch."""
from . import pytuya
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Setup a Tuya fan based on a config entry."""
device, entities_to_setup = prepare_setup_entities(
config_entry, PLATFORM
)
if not entities_to_setup:
return
fans = []
pytuyadevice = pytuya.TuyaDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY))
pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION)))
_LOGGER.debug("localtuya fan: setup_platform: %s", pytuyadevice)
fan_device = pytuyadevice
fans.append(
for device_config in entities_to_setup:
fans.append(
LocaltuyaFan(
fan_device,
config.get(CONF_NAME),
config.get(CONF_FRIENDLY_NAME),
config.get(CONF_ICON),
config.get(CONF_ID),
device,
device_config[CONF_FRIENDLY_NAME],
device_config[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):
"""Representation of a Tuya fan."""
# def __init__(self, hass, name: str, supported_features: int) -> None:
def __init__(self, device, name, friendly_name, icon, switchid):
def __init__(self, device, friendly_name, switchid):
"""Initialize the entity."""
self._device = device
self._name = friendly_name
self._available = False
self._friendly_name = friendly_name
self._icon = icon
self._switch_id = switchid
self._status = self._device.status()
self._state = False
@@ -95,7 +91,7 @@ class LocaltuyaFan(FanEntity):
@property
def oscillating(self):
"""Return current oscillating status."""
#if self._speed == STATE_OFF:
# if self._speed == STATE_OFF:
# return False
_LOGGER.debug("localtuya fan: oscillating = %s", self._oscillating)
return self._oscillating
@@ -163,7 +159,7 @@ class LocaltuyaFan(FanEntity):
def oscillate(self, oscillating: bool) -> None:
"""Set oscillation."""
self._oscillating = oscillating
self._device.set_value('8', oscillating)
self._device.set_value("8", oscillating)
self.schedule_update_ha_state()
# @property
@@ -174,8 +170,7 @@ class LocaltuyaFan(FanEntity):
@property
def unique_id(self):
"""Return unique device identifier."""
_LOGGER.debug("localtuya fan unique_id = %s", self._device)
return self._device.id
return f"local_{self._device.id}_{self._switch_id}"
@property
def available(self):
@@ -195,20 +190,20 @@ class LocaltuyaFan(FanEntity):
try:
status = self._device.status()
_LOGGER.debug("localtuya fan: status = %s", status)
self._state = status['dps']['1']
if status['dps']['1'] == False:
self._state = status["dps"]["1"]
if status["dps"]["1"] == False:
self._speed = STATE_OFF
elif int(status['dps']['2']) == 1:
elif int(status["dps"]["2"]) == 1:
self._speed = SPEED_LOW
elif int(status['dps']['2']) == 2:
elif int(status["dps"]["2"]) == 2:
self._speed = SPEED_MEDIUM
elif int(status['dps']['2']) == 3:
elif int(status["dps"]["2"]) == 3:
self._speed = SPEED_HIGH
# self._speed = status['dps']['2']
self._oscillating = status['dps']['8']
self._oscillating = status["dps"]["8"]
success = True
except ConnectionError:
if i+1 == 3:
if i + 1 == 3:
success = False
raise ConnectionError("Failed to update status.")
self._available = success

155
custom_components/localtuya/light.py
View File

@@ -12,76 +12,78 @@ light:
friendly_name: This Light
protocol_version: 3.3
"""
import voluptuous as vol
from homeassistant.const import (CONF_HOST, CONF_ID, CONF_SWITCHES, CONF_FRIENDLY_NAME, CONF_ICON, CONF_NAME)
import homeassistant.helpers.config_validation as cv
import socket
import logging
from time import time, sleep
from threading import Lock
import logging
from homeassistant.const import (
CONF_ID,
CONF_FRIENDLY_NAME,
)
from homeassistant.components.light import (
LightEntity,
PLATFORM_SCHEMA,
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP,
ATTR_HS_COLOR,
SUPPORT_BRIGHTNESS,
SUPPORT_COLOR,
SUPPORT_COLOR_TEMP,
LightEntity,
PLATFORM_SCHEMA
)
from homeassistant.util import color as colorutil
import socket
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
from . import BASE_PLATFORM_SCHEMA, import_from_yaml, prepare_setup_entities
from .pytuya import TuyaDevice
_LOGGER = logging.getLogger(__name__)
PLATFORM = "light"
MIN_MIRED = 153
MAX_MIRED = 370
UPDATE_RETRY_LIMIT = 3
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,
})
log = logging.getLogger(__name__)
log.setLevel(level=logging.DEBUG) # Debug hack!
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(BASE_PLATFORM_SCHEMA)
def flow_schema(dps):
"""Return schema used in config flow."""
return {}
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Setup a Tuya switch based on a config entry."""
device, entities_to_setup = prepare_setup_entities(
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):
"""Set up of the Tuya switch."""
from . import pytuya
return import_from_yaml(hass, config, PLATFORM)
lights = []
pytuyadevice = pytuya.TuyaDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY))
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:
"""Cache wrapper for pytuya.TuyaDevices"""
def __init__(self, device):
"""Initialize the cache."""
self._cached_status = ''
self._cached_status = ""
self._cached_status_time = 0
self._device = device
self._lock = Lock()
@@ -94,18 +96,14 @@ class TuyaCache:
def __get_status(self, switchid):
for _ in range(UPDATE_RETRY_LIMIT):
try:
status = self._device.status()['dps'][switchid]
return status
except ConnectionError:
return self._device.status()["dps"][switchid]
except (ConnectionError, socket.timeout):
pass
except socket.timeout:
pass
log.warn(
"Failed to get status after {} tries".format(UPDATE_RETRY_LIMIT))
_LOGGER.warning("Failed to get status after %d tries", UPDATE_RETRY_LIMIT)
def set_dps(self, state, dps_index):
"""Change the Tuya switch status and clear the cache."""
self._cached_status = ''
self._cached_status = ""
self._cached_status_time = 0
for _ in range(UPDATE_RETRY_LIMIT):
try:
@@ -114,21 +112,17 @@ class TuyaCache:
pass
except socket.timeout:
pass
log.warn(
"Failed to set status after {} tries".format(UPDATE_RETRY_LIMIT))
_LOGGER.warning("Failed to set status after %d tries", UPDATE_RETRY_LIMIT)
def status(self, switchid):
"""Get state of Tuya switch and cache the results."""
self._lock.acquire()
try:
with self._lock:
now = time()
if not self._cached_status or now - self._cached_status_time > 15:
sleep(0.5)
self._cached_status = self.__get_status(switchid)
self._cached_status_time = time()
return self._cached_status
finally:
self._lock.release()
def cached_status(self):
return self._cached_status
@@ -144,7 +138,7 @@ class LocaltuyaLight(LightEntity):
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."""
self._device = device
self._available = False
@@ -152,9 +146,11 @@ class LocaltuyaLight(LightEntity):
self._state = False
self._brightness = 127
self._color_temp = 127
self._icon = icon
self._bulb_id = bulbid
def state(self):
self._device.state()
@property
def name(self):
"""Get name of Tuya switch."""
@@ -175,11 +171,6 @@ class LocaltuyaLight(LightEntity):
"""Check if Tuya switch is on."""
return self._state
@property
def icon(self):
"""Return the icon."""
return self._icon
def update(self):
"""Get state of Tuya switch."""
try:
@@ -193,12 +184,12 @@ class LocaltuyaLight(LightEntity):
status = self._device.status(self._bulb_id)
self._state = status
try:
brightness = int(self._device.brightness())
if brightness > 254:
brightness = 255
if brightness < 25:
brightness = 25
self._brightness = brightness
brightness = int(self._device.brightness())
if brightness > 254:
brightness = 255
if brightness < 25:
brightness = 25
self._brightness = brightness
except TypeError:
pass
self._color_temp = self._device.color_temp()
@@ -208,16 +199,16 @@ class LocaltuyaLight(LightEntity):
"""Return the brightness of the light."""
return self._brightness
# @property
# def hs_color(self):
# """Return the hs_color of the light."""
# return (self._device.color_hsv()[0],self._device.color_hsv()[1])
# @property
# def hs_color(self):
# """Return the hs_color of the light."""
# return (self._device.color_hsv()[0],self._device.color_hsv()[1])
@property
def color_temp(self):
"""Return the color_temp of the light."""
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:
pass
@@ -233,7 +224,7 @@ class LocaltuyaLight(LightEntity):
def turn_on(self, **kwargs):
"""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():
self._device.set_dps(True, self._bulb_id)
if ATTR_BRIGHTNESS in kwargs:
@@ -244,7 +235,11 @@ class LocaltuyaLight(LightEntity):
if ATTR_HS_COLOR in kwargs:
raise ValueError(" TODO implement RGB from HS")
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)
def turn_off(self, **kwargs):
@@ -257,7 +252,7 @@ class LocaltuyaLight(LightEntity):
supports = SUPPORT_BRIGHTNESS
if self._device.color_temp() != "999":
supports = supports | SUPPORT_COLOR
#supports = supports | SUPPORT_COLOR_TEMP
# supports = supports | SUPPORT_COLOR_TEMP
return supports
def support_color(self):
@@ -319,7 +314,7 @@ class LocaltuyaLight(LightEntity):
an RGB value.
Args:
hexvalue(string): The hex representation generated by BulbDevice._rgb_to_hexvalue()
hexvalue(string): The hex representation generated by TuyaDevice._rgb_to_hexvalue()
"""
r = int(hexvalue[0:2], 16)
g = int(hexvalue[2:4], 16)
@@ -334,7 +329,7 @@ class LocaltuyaLight(LightEntity):
an HSV value.
Args:
hexvalue(string): The hex representation generated by BulbDevice._rgb_to_hexvalue()
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

17
custom_components/localtuya/manifest.json
View File

@@ -1,8 +1,11 @@
{
"domain": "localtuya",
"name": "LocalTuya integration",
"documentation": "https://github.com/rospogrigio/localtuya-homeassistant/",
"dependencies": [],
"codeowners": ["@rospogrigio"],
"requirements": ["pycryptodome==3.9.8"]
{
"domain": "localtuya",
"name": "LocalTuya integration",
"documentation": "https://github.com/rospogrigio/localtuya-homeassistant/",
"dependencies": [],
"codeowners": [
"@rospogrigio"
],
"requirements": ["pycryptodome==3.9.8"],
"config_flow": true
}

42
custom_components/localtuya/strings.json Normal file
View File

@@ -0,0 +1,42 @@
{
"config": {
"abort": {
"already_configured": "Device has already been configured.",
"unsupported_device_type": "Unsupported device type!"
},
"error": {
"cannot_connect": "Cannot connect to device. Verify that address is correct.",
"invalid_auth": "Failed to authenticate with device. Verify that device id and local key are correct.",
"unknown": "An unknown error occurred. See log for details.",
"switch_already_configured": "Switch with this ID has already been configured."
},
"step": {
"user": {
"title": "Add Tuya device",
"description": "Fill in the basic details and pick device type. The name entered here will be used to identify the integration itself (as seen in the `Integrations` page). You will name each sub-device in the following steps.",
"data": {
"name": "Name",
"host": "Host",
"device_id": "Device ID",
"local_key": "Local key",
"protocol_version": "Protocol Version",
"device_type": "Device type"
}
},
"power_outlet": {
"title": "Add subswitch",
"description": "You are about to add subswitch number `{number}`. If you want to add another, tick `Add another switch` before continuing.",
"data": {
"id": "ID",
"name": "Name",
"friendly_name": "Friendly name",
"current": "Current",
"current_consumption": "Current Consumption",
"voltage": "Voltage",
"add_another_switch": "Add another switch"
}
}
}
},
"title": "LocalTuya"
}

260
custom_components/localtuya/switch.py
View File

@@ -22,132 +22,109 @@ switch:
sw02:
name: usb_plug
friendly_name: USB Plug
id: 7
id: 7
"""
import logging
import voluptuous as vol
from homeassistant.components.switch import SwitchEntity, PLATFORM_SCHEMA
from homeassistant.const import (CONF_HOST, CONF_ID, CONF_SWITCHES, CONF_FRIENDLY_NAME, CONF_ICON, CONF_NAME)
import homeassistant.helpers.config_validation as cv
from time import time, sleep
from threading import Lock
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 (
ATTR_CURRENT,
ATTR_CURRENT_CONSUMPTION,
ATTR_VOLTAGE,
CONF_CURRENT,
CONF_CURRENT_CONSUMPTION,
CONF_VOLTAGE,
)
from .pytuya import TuyaDevice
_LOGGER = logging.getLogger(__name__)
CONF_DEVICE_ID = 'device_id'
CONF_LOCAL_KEY = 'local_key'
CONF_PROTOCOL_VERSION = 'protocol_version'
CONF_CURRENT = 'current'
CONF_CURRENT_CONSUMPTION = 'current_consumption'
CONF_VOLTAGE = 'voltage'
PLATFORM = "switch"
DEFAULT_ID = '1'
DEFAULT_PROTOCOL_VERSION = 3.3
DEFAULT_ID = "1"
ATTR_CURRENT = 'current'
ATTR_CURRENT_CONSUMPTION = 'current_consumption'
ATTR_VOLTAGE = 'voltage'
# TODO: This will eventully merge with flow_schema
SWITCH_SCHEMA = vol.Schema(
{
vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string,
vol.Optional(CONF_NAME): cv.string, # Deprecated: not used
vol.Required(CONF_FRIENDLY_NAME): cv.string,
vol.Optional(CONF_CURRENT, default="-1"): cv.string,
vol.Optional(CONF_CURRENT_CONSUMPTION, default="-1"): cv.string,
vol.Optional(CONF_VOLTAGE, default="-1"): cv.string,
}
)
SWITCH_SCHEMA = vol.Schema({
vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string,
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_FRIENDLY_NAME): cv.string,
vol.Optional(CONF_CURRENT, default='-1'): cv.string,
vol.Optional(CONF_CURRENT_CONSUMPTION, default='-1'): cv.string,
vol.Optional(CONF_VOLTAGE, default='-1'): cv.string,
})
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(BASE_PLATFORM_SCHEMA).extend(
{
vol.Optional(CONF_CURRENT, default="-1"): cv.string,
vol.Optional(CONF_CURRENT_CONSUMPTION, default="-1"): cv.string,
vol.Optional(CONF_VOLTAGE, default="-1"): cv.string,
vol.Optional(CONF_SWITCHES, default={}): vol.Schema({cv.slug: SWITCH_SCHEMA}),
}
)
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_CURRENT, default='-1'): cv.string,
vol.Optional(CONF_CURRENT_CONSUMPTION, default='-1'): cv.string,
vol.Optional(CONF_VOLTAGE, default='-1'): cv.string,
vol.Optional(CONF_SWITCHES, default={}):
vol.Schema({cv.slug: SWITCH_SCHEMA}),
})
def flow_schema(dps):
"""Return schema used in config flow."""
return {
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):
"""Setup a Tuya switch based on a config entry."""
device, entities_to_setup = prepare_setup_entities(
config_entry, PLATFORM
)
if not entities_to_setup:
return
switches = []
for device_config in entities_to_setup:
switches.append(
LocaltuyaSwitch(
TuyaCache(device),
device_config[CONF_FRIENDLY_NAME],
device_config[CONF_ID],
device_config.get(CONF_CURRENT),
device_config.get(CONF_CURRENT_CONSUMPTION),
device_config.get(CONF_VOLTAGE),
)
)
async_add_entities(switches, True)
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up of the Tuya switch."""
from . import pytuya
return import_from_yaml(hass, config, PLATFORM)
devices = config.get(CONF_SWITCHES)
switches = []
pytuyadevice = pytuya.TuyaDevice(config.get(CONF_DEVICE_ID), config.get(CONF_HOST), config.get(CONF_LOCAL_KEY))
pytuyadevice.set_version(float(config.get(CONF_PROTOCOL_VERSION)))
if len(devices) > 0:
for object_id, device_config in devices.items():
dps = {}
dps[device_config.get(CONF_ID)]=None
if device_config.get(CONF_CURRENT) != '-1':
dps[device_config.get(CONF_CURRENT)]=None
if device_config.get(CONF_CURRENT_CONSUMPTION) != '-1':
dps[device_config.get(CONF_CURRENT_CONSUMPTION)]=None
if device_config.get(CONF_VOLTAGE) != '-1':
dps[device_config.get(CONF_VOLTAGE)]=None
pytuyadevice.set_dpsUsed(dps)
outlet_device = TuyaCache(pytuyadevice)
switches.append(
LocaltuyaSwitch(
outlet_device,
device_config.get(CONF_NAME),
device_config.get(CONF_FRIENDLY_NAME, object_id),
device_config.get(CONF_ICON),
device_config.get(CONF_ID),
device_config.get(CONF_CURRENT),
device_config.get(CONF_CURRENT_CONSUMPTION),
device_config.get(CONF_VOLTAGE)
)
)
print('Setup localtuya subswitch [{}] with device ID [{}] '.format(device_config.get(CONF_FRIENDLY_NAME, object_id), device_config.get(CONF_ID)))
_LOGGER.info("Setup localtuya subswitch %s with device ID %s ", device_config.get(CONF_FRIENDLY_NAME, object_id), device_config.get(CONF_ID) )
else:
dps = {}
dps[config.get(CONF_ID)]=None
if config.get(CONF_CURRENT) != '-1':
dps[config.get(CONF_CURRENT)]=None
if config.get(CONF_CURRENT_CONSUMPTION) != '-1':
dps[config.get(CONF_CURRENT_CONSUMPTION)]=None
if config.get(CONF_VOLTAGE) != '-1':
dps[config.get(CONF_VOLTAGE)]=None
pytuyadevice.set_dpsUsed(dps)
outlet_device = TuyaCache(pytuyadevice)
switches.append(
LocaltuyaSwitch(
outlet_device,
config.get(CONF_NAME),
config.get(CONF_FRIENDLY_NAME),
config.get(CONF_ICON),
config.get(CONF_ID),
config.get(CONF_CURRENT),
config.get(CONF_CURRENT_CONSUMPTION),
config.get(CONF_VOLTAGE)
)
)
print('Setup localtuya switch [{}] with device ID [{}] '.format(config.get(CONF_FRIENDLY_NAME), config.get(CONF_ID)))
_LOGGER.info("Setup localtuya switch %s with device ID %s ", config.get(CONF_FRIENDLY_NAME), config.get(CONF_ID) )
add_devices(switches, True)
class TuyaCache:
"""Cache wrapper for pytuya.TuyaDevice"""
def __init__(self, device):
"""Initialize the cache."""
self._cached_status = ''
self._cached_status = ""
self._cached_status_time = 0
self._device = device
self._lock = Lock()
@@ -163,26 +140,37 @@ class TuyaCache:
status = self._device.status()
return status
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)
if i+1 == 3:
_LOGGER.error("Failed to update status of device %s", self._device.address )
# return None
if i + 1 == 3:
_LOGGER.error(
"Failed to update status of device %s", self._device.address
)
# return None
raise ConnectionError("Failed to update status .")
def set_dps(self, state, dps_index):
"""Change the Tuya switch status and clear the cache."""
self._cached_status = ''
self._cached_status = ""
self._cached_status_time = 0
for i in range(5):
try:
return self._device.set_dps(state, dps_index)
except Exception:
print('Failed to set status of device [{}]'.format(self._device.address))
if i+1 == 3:
_LOGGER.error("Failed to set status of device %s", self._device.address )
print(
"Failed to set status of device [{}]".format(self._device.address)
)
if i + 1 == 3:
_LOGGER.error(
"Failed to set status of device %s", self._device.address
)
return
# raise ConnectionError("Failed to set status.")
# raise ConnectionError("Failed to set status.")
def status(self):
"""Get state of Tuya switch and cache the results."""
@@ -200,19 +188,30 @@ class TuyaCache:
class LocaltuyaSwitch(SwitchEntity):
"""Representation of a Tuya switch."""
def __init__(self, device, name, friendly_name, icon, switchid, attr_current, attr_consumption, attr_voltage):
def __init__(
self,
device,
friendly_name,
switchid,
attr_current,
attr_consumption,
attr_voltage,
):
"""Initialize the Tuya switch."""
self._device = device
self._name = friendly_name
self._available = False
self._icon = icon
self._switch_id = switchid
self._attr_current = attr_current
self._attr_consumption = attr_consumption
self._attr_voltage = attr_voltage
self._status = self._device.status()
self._state = self._status['dps'][self._switch_id]
print('Initialized tuya switch [{}] with switch status [{}] and state [{}]'.format(self._name, self._status, self._state))
self._status = None
self._state = None
print(
"Initialized tuya switch [{}] with switch status [{}] and state [{}]".format(
self._name, self._status, self._state
)
)
@property
def name(self):
@@ -237,23 +236,16 @@ class LocaltuyaSwitch(SwitchEntity):
@property
def device_state_attributes(self):
attrs = {}
try:
attrs[ATTR_CURRENT] = "{}".format(self._status['dps'][self._attr_current])
attrs[ATTR_CURRENT_CONSUMPTION] = "{}".format(self._status['dps'][self._attr_consumption]/10)
attrs[ATTR_VOLTAGE] = "{}".format(self._status['dps'][self._attr_voltage]/10)
# print('attrs[ATTR_CURRENT]: [{}]'.format(attrs[ATTR_CURRENT]))
# print('attrs[ATTR_CURRENT_CONSUMPTION]: [{}]'.format(attrs[ATTR_CURRENT_CONSUMPTION]))
# print('attrs[ATTR_VOLTAGE]: [{}]'.format(attrs[ATTR_VOLTAGE]))
except KeyError:
pass
if self._attr_current:
attrs[ATTR_CURRENT] = self._status["dps"][self._attr_current]
if self._attr_consumption:
attrs[ATTR_CURRENT_CONSUMPTION] = (
self._status["dps"][self._attr_consumption] / 10
)
if self._attr_voltage:
attrs[ATTR_VOLTAGE] = self._status["dps"][self._attr_voltage] / 10
return attrs
@property
def icon(self):
"""Return the icon."""
return self._icon
def turn_on(self, **kwargs):
"""Turn Tuya switch on."""
self._device.set_dps(True, self._switch_id)
@@ -266,7 +258,7 @@ class LocaltuyaSwitch(SwitchEntity):
"""Get state of Tuya switch."""
try:
self._status = self._device.status()
self._state = self._status['dps'][self._switch_id]
self._state = self._status["dps"][self._switch_id]
except Exception:
self._available = False
else:

49
custom_components/localtuya/translations/en.json Normal file
View File

@@ -0,0 +1,49 @@
{
"config": {
"abort": {
"already_configured": "Device has already been configured."
},
"error": {
"cannot_connect": "Cannot connect to device. Verify that address is correct and try again.",
"invalid_auth": "Failed to authenticate with device. Verify that device id and local key are correct.",
"unknown": "An unknown error occurred. See log for details.",
"entity_already_configured": "Entity with this ID has already been configured."
},
"step": {
"user": {
"title": "Add Tuya device",
"description": "Fill in the basic details and pick device type. The name entered here will be used to identify the integration itself (as seen in the `Integrations` page). You will add entities and give them names in the following steps.",
"data": {
"name": "Name",
"host": "Host",
"device_id": "Device ID",
"local_key": "Local key",
"protocol_version": "Protocol Version"
}
},
"pick_entity_type": {
"title": "Entity type selection",
"description": "Please pick the type of entity you want to add.",
"data": {
"platform_to_add": "Platform",
"no_additional_platforms": "Do not add any more entities"
}
},
"add_entity": {
"title": "Add new entity",
"description": "Please fill out the details for an entity with type `{platform}`.",
"data": {
"id": "ID",
"friendly_name": "Friendly name",
"current": "Current",
"current_consumption": "Current Consumption",
"voltage": "Voltage",
"open_cmd": "Open Command",
"close_cmd": "Close Command",
"stop_cmd": "Stop Command"
}
}
}
},
"title": "LocalTuya"
}