diff --git a/custom_components/localtuya/switch.py b/custom_components/localtuya/switch.py new file mode 100644 index 0000000..427e500 --- /dev/null +++ b/custom_components/localtuya/switch.py @@ -0,0 +1,168 @@ +""" +Simple platform to control **SOME** Tuya switch devices. + +For more details about this platform, please refer to the documentation at +https://home-assistant.io/components/switch.tuya/ +""" +import voluptuous as vol +from homeassistant.components.switch import SwitchDevice, PLATFORM_SCHEMA +from homeassistant.const import (CONF_NAME, CONF_HOST, CONF_ID, CONF_SWITCHES, CONF_FRIENDLY_NAME, CONF_ICON) +import homeassistant.helpers.config_validation as cv +from time import time +from threading import Lock + +REQUIREMENTS = ['pytuya==7.0.2'] + +CONF_DEVICE_ID = 'device_id' +CONF_LOCAL_KEY = 'local_key' + +DEFAULT_ID = '1' + +ATTR_CURRENT = 'current' +ATTR_CURRENT_CONSUMPTION = 'current_consumption' +ATTR_VOLTAGE = 'voltage' + +SWITCH_SCHEMA = vol.Schema({ + vol.Optional(CONF_ID, default=DEFAULT_ID): cv.string, + vol.Optional(CONF_FRIENDLY_NAME): cv.string, +}) + +PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ + vol.Optional(CONF_NAME): cv.string, + 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.Optional(CONF_ID, default=DEFAULT_ID): cv.string, + vol.Optional(CONF_SWITCHES, default={}): + vol.Schema({cv.slug: SWITCH_SCHEMA}), +}) + + +def setup_platform(hass, config, add_devices, discovery_info=None): + """Set up of the Tuya switch.""" + import pytuya + + devices = config.get(CONF_SWITCHES) + switches = [] + + outlet_device = TuyaCache( + pytuya.OutletDevice( + config.get(CONF_DEVICE_ID), + config.get(CONF_HOST), + config.get(CONF_LOCAL_KEY) + ) + ) + + for object_id, device_config in devices.items(): + switches.append( + TuyaDevice( + outlet_device, + device_config.get(CONF_FRIENDLY_NAME, object_id), + device_config.get(CONF_ICON), + device_config.get(CONF_ID) + ) + ) + + name = config.get(CONF_NAME) + if name: + switches.append( + TuyaDevice( + outlet_device, + name, + config.get(CONF_ICON), + config.get(CONF_ID) + ) + ) + + add_devices(switches) + +class TuyaCache: + """Cache wrapper for pytuya.OutletDevice""" + + def __init__(self, device): + """Initialize the cache.""" + self._cached_status = '' + self._cached_status_time = 0 + self._device = device + self._lock = Lock() + + def __get_status(self): + for i in range(3): + try: + status = self._device.status() + return status + except ConnectionError: + if i+1 == 3: + raise ConnectionError("Failed to update status.") + + def set_status(self, state, switchid): + """Change the Tuya switch status and clear the cache.""" + self._cached_status = '' + self._cached_status_time = 0 + return self._device.set_status(state, switchid) + + def status(self): + """Get state of Tuya switch and cache the results.""" + self._lock.acquire() + try: + now = time() + if not self._cached_status or now - self._cached_status_time > 20: + self._cached_status = self.__get_status() + self._cached_status_time = time() + return self._cached_status + finally: + self._lock.release() + +class TuyaDevice(SwitchDevice): + """Representation of a Tuya switch.""" + + def __init__(self, device, name, icon, switchid): + """Initialize the Tuya switch.""" + self._device = device + self._name = name + self._state = False + self._icon = icon + self._switchid = switchid + self._status = self._device.status() + + @property + def name(self): + """Get name of Tuya switch.""" + return self._name + + @property + def is_on(self): + """Check if Tuya switch is on.""" + return self._state + + @property + def device_state_attributes(self): + attrs = {} + try: + attrs[ATTR_CURRENT] = "{}".format(self._status['dps']['104']) + attrs[ATTR_CURRENT_CONSUMPTION] = "{}".format(self._status['dps']['105']/10) + attrs[ATTR_VOLTAGE] = "{}".format(self._status['dps']['106']/10) + except KeyError: + pass + return attrs + + @property + def icon(self): + """Return the icon.""" + return self._icon + + def turn_on(self, **kwargs): + """Turn Tuya switch on.""" + self._device.set_status(True, self._switchid) + + def turn_off(self, **kwargs): + """Turn Tuya switch off.""" + self._device.set_status(False, self._switchid) + + def update(self): + """Get state of Tuya switch.""" + status = self._device.status() + self._status= status + self._state = status['dps'][self._switchid] +