Skip to content
Snippets Groups Projects
Commit 0c2dd719 authored by jkerdreu's avatar jkerdreu
Browse files

Added support for smart plug

git-svn-id: https://redmine.imt-atlantique.fr/svn/xaal/code/Python/trunk@2303 b32b6428-25c9-4566-ad07-03861ab6144f
parent 58cfdfd7
No related branches found
No related tags found
No related merge requests found
......@@ -24,18 +24,16 @@ class TuyaDev:
self.tuya_id = tuya_id
self.tuya_key = tuya_key
self.tuya_ip = tuya_ip
self.tuya_dev = None
self.base_addr = base_addr
self.last_update = 0
self.devices = []
self.setup()
self.init_properties()
def setup(self):
logger.warning('Please overide setup()')
logger.warning('Please override setup()')
def update_status(self):
pass
def init_properties(self):
for dev in self.devices:
dev.vendor_id = 'IHSEV / Tuya'
......@@ -43,52 +41,98 @@ class TuyaDev:
dev.info = 'Tuya device: %s @ %s' % (self.tuya_id,self.tuya_ip)
if len(self.devices) > 1:
dev.group_id = self.base_addr + 'ff'
def update_status(self):
try:
self._update_status()
except RetryError as e:
logger.warn(e)
class PowerRelay(TuyaDev):
def setup(self):
dev = devices.powerrelay_toggle(self.base_addr + '00')
dev.methods['on'] = self.on
dev.methods['off'] = self.off
dev.methods['toggle'] = self.toggle
dev.product_id = 'Generic Tuya Outlet'
self.devices.append(dev)
self.relay = pytuya.OutletDevice(self.tuya_id,self.tuya_ip,self.tuya_key)
self.relay.set_version(3.3)
@retry(stop=stop_after_attempt(2))
def _update_status(self):
time.sleep(0.1)
logger.info("Updating %s" % self.tuya_id)
self.last_update = now()
status = self.tuya_dev.status()
logger.warning(status)
self.on_status(status)
def on_status(self,status):
logger.warning('Please override on_status')
class OnOffMixing(object):
def setup_onoff(self,tuya_dev,xaal_dev):
tuya_dev.set_version(3.3)
self.tuya_dev = tuya_dev
xaal_dev.methods['on'] = self.on
xaal_dev.methods['off'] = self.off
xaal_dev.methods['toggle'] = self.toggle
self.xaal_dev = xaal_dev
@retry(stop=stop_after_attempt(2))
def on(self):
self.relay.turn_on()
self.tuya_dev.turn_on()
self.update_status()
@retry(stop=stop_after_attempt(2))
@retry(stop=stop_after_attempt(2))
def off(self):
self.relay.turn_off()
self.tuya_dev.turn_off()
self.update_status()
def toggle(self):
power = self.devices[0].attributes['power']
power = self.xaal_dev.attributes['power']
if power == None:
self.update_status()
power = self.devices[0].attributes['power']
power = self.xaal_dev.attributes['power']
if power == True:
self.off()
elif power == False:
self.on()
@retry(stop=stop_after_attempt(2))
def _update_status(self):
logger.info("Updating %s" % self.tuya_id)
time.sleep(0.1)
self.last_update = now()
power = self.relay.status()['dps']['1']
self.devices[0].attributes['power'] = power
def update_status(self):
try:
self._update_status()
except RetryError as e:
logger.warn(e)
class PowerRelay(TuyaDev,OnOffMixing):
def setup(self):
relay = devices.powerrelay_toggle(self.base_addr + '00')
relay.product_id = 'Generic Tuya Outlet'
outlet = pytuya.OutletDevice(self.tuya_id,self.tuya_ip,self.tuya_key)
outlet.set_version(3.3)
self.setup_onoff(outlet,relay)
self.devices.append(relay)
def on_status(self,status):
self.devices[0].attributes['power'] = status['dps'][1]
class SmartPlug(TuyaDev,OnOffMixing):
def setup(self):
# TBD switch to the right schema, but right now nothing exists
relay = devices.powerrelay_toggle(self.base_addr + '00')
relay.product_id = 'Tuya Smart Plug Outlet'
plug = pytuya.OutletDevice(self.tuya_id,self.tuya_ip,self.tuya_key)
self.setup_onoff(plug,relay)
pmeter = devices.powermeter(self.base_addr + '01')
pmeter.product_id = 'Tuya Smart Plug Power Meter'
pmeter.new_attribute('voltage')
pmeter.new_attribute('current')
pmeter.del_attribute(pmeter.get_attribute('energy'))
pmeter.del_attribute(pmeter.get_attribute('devices'))
pmeter.unsupported_attributes = ['energy','devices']
self.devices.append(relay)
self.devices.append(pmeter)
def on_status(self,status):
dps = status['dps']
print(dps)
self.devices[0].attributes['power'] = dps.get('1',None)
self.devices[1].attributes['current'] = int(dps.get('4',0)) / 1000
self.devices[1].attributes['power'] = int(dps.get('5',0)) / 10
self.devices[1].attributes['voltage'] = int(dps.get('6',0)) / 10
# TODO I don't have this kind of devices right now
# TODO I don't have this kind of devices right now
class RGBLamp(TuyaDev):pass
class PowerRelayMetering(TuyaDev):pass
......@@ -12,6 +12,9 @@ logger = logging.getLogger(__name__)
# poll devices only every n seconds
REFRESH_RATE=30
CFG_MAP = { 'powerrelay': devices.PowerRelay,
'smartplug' : devices.SmartPlug }
class GW:
def __init__(self,engine):
self.engine = engine
......@@ -44,7 +47,7 @@ class GW:
base_addr = cfg.get('base_addr',None)
ip = cfg.get('ip',None)
key = cfg.get('key',None)
type_ = cfg.get('type','powerrelay')
type_ = cfg.get('type','PowerRelay')
if base_addr == None:
cfg['base_addr'] = base_addr = tools.get_random_uuid()[:-2]
......@@ -57,13 +60,16 @@ class GW:
self.engine.add_device(gw)
def add_device(self,tuya_id,tuya_key,ip,base_addr,type_):
if type_ == 'powerrelay':
dev = devices.PowerRelay(tuya_id,tuya_key,ip,base_addr)
dev_type = CFG_MAP.get(type_,None)
if dev_type:
dev = dev_type(tuya_id,tuya_key,ip,base_addr)
self.devices[tuya_id] = dev
for d in dev.devices:
self.engine.add_device(d)
self.gw.attributes['embedded'].append(d.address)
dev.update_status()
else:
logger.warn(f"Unsupported device type {type_} {ip}")
def update(self):
# Tuya protocol doesn't support update broadcasting
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment