Tuya 3.3 support (fixes #3)

* support tuya 3.3 protocol
* Switch argument order
* Debug logging for demo
* Home Assistant config changes
* Vacuum has a 'power' property
* Formatting changes
* Improve Tuya message repr
* Include socket error in error output.

README.md

@@ -15,7 +15,7 @@ bin/pip install -e .
 
 ## Demo usage
 ```
-bin/demo DEVICE_ID LOCAL_KEY IP
+bin/demo DEVICE_ID IP LOCAL_KEY
 ```
 
 The demo:

eufy_robovac/demo.py

@@ -15,12 +15,16 @@
 # limitations under the License.
 
 import asyncio
+import logging
 import pprint
 import sys
 
 from eufy_robovac.robovac import Robovac
 
 
+logging.basicConfig(level=logging.DEBUG)
+
+
 async def connected_callback(message, device):
     print("Connected. Current device state:")
     pprint.pprint(device.state)
@@ -30,8 +34,8 @@ async def cleaning_started_callback(message, device):
     print("Cleaning started.")
 
 
-async def async_main(device_id, local_key, ip, *args, **kwargs):
+async def async_main(device_id, ip, local_key=None, *args, **kwargs):
-    r = Robovac(device_id, local_key, ip, *args, **kwargs)
+    r = Robovac(device_id, ip, local_key, *args, **kwargs)
     await r.async_connect(connected_callback)
     await asyncio.sleep(1)
     print("Starting cleaning...")

eufy_robovac/platform.py

@@ -17,7 +17,7 @@ DOMAIN = 'eufy_vacuum'
 
 DEVICE_SCHEMA = vol.Schema({
     vol.Required(CONF_ADDRESS): cv.string,
-    vol.Required(CONF_ACCESS_TOKEN): cv.string,
+    vol.Optional(CONF_ACCESS_TOKEN): cv.string,
     vol.Required(CONF_ID): cv.string,
     vol.Required(CONF_TYPE): cv.string,
     vol.Optional(CONF_NAME): cv.string
@@ -36,9 +36,9 @@ def setup(hass, config):
     for device_info in config.get(DOMAIN, {}).get(CONF_DEVICES, []):
         device = {}
         device['address'] = device_info[CONF_ADDRESS]
-        device['local_key'] = device_info[CONF_ACCESS_TOKEN]
+        device['local_key'] = device_info.get(CONF_ACCESS_TOKEN)
         device['device_id'] = device_info[CONF_ID]
-        device['name'] = device_info[CONF_NAME]
+        device['name'] = device_info.get(CONF_NAME)
         device['model'] = device_info[CONF_TYPE]
         discovery.load_platform(hass, 'vacuum', DOMAIN, device, config)
 

eufy_robovac/robovac.py

@@ -75,6 +75,7 @@ class ErrorCode(StringEnum):
 class Robovac(TuyaDevice):
     """Represents a generic Eufy Robovac."""
 
+    POWER = '1'
     PLAY_PAUSE = '2'
     DIRECTION = '3'
     WORK_MODE = '5'
@@ -85,6 +86,7 @@ class Robovac(TuyaDevice):
     BATTERY_LEVEL = '104'
     ERROR_CODE = '106'
 
+    power = DeviceProperty(POWER)
     play_pause = DeviceProperty(PLAY_PAUSE)
     direction = DeviceProperty(DIRECTION)
     work_mode = DeviceProperty(WORK_MODE, WorkMode)

eufy_robovac/tuya.py

@@ -166,23 +166,34 @@ class TuyaCipher:
         self.cipher = Cipher(algorithms.AES(key.encode('ascii')), modes.ECB(),
                              backend=openssl_backend)
 
-    def has_prefix(self, encrypted_data):
+    def get_prefix_size_and_validate(self, command, encrypted_data):
-        version = tuple(map(int, encrypted_data[:3].decode('utf8').split('.')))
+        try:
+            version = tuple(map(int, encrypted_data[:3].decode('utf8').split('.')))
+        except UnicodeDecodeError:
+            version = (0, 0)
         if version != self.version:
-            return False
+            return 0
-        hash = encrypted_data[3:19].decode('ascii')
+        if version < (3, 3):
-        expected_hash = self.hash(encrypted_data[19:])
+            hash = encrypted_data[3:19].decode('ascii')
-        if hash != expected_hash:
+            expected_hash = self.hash(encrypted_data[19:])
-            return False
+            if hash != expected_hash:
+                return 0
+            return 19
+        else:
+            if command in (Message.SET_COMMAND, Message.GRATUITOUS_UPDATE):
+                _, sequence, __, ___ = struct.unpack_from(
+                    '>IIIH', encrypted_data, 3)
+                return 15
+        return 0
         
-        return True
 
-    def decrypt(self, data):
+    def decrypt(self, command, data):
-        # Strip version and MD5 hash
+        prefix_size = self.get_prefix_size_and_validate(command, data)
-        if self.has_prefix(data):
+        data = data[prefix_size:]
-            data = data[19:]
         decryptor = self.cipher.decryptor()
-        decrypted_data = decryptor.update(base64.b64decode(data))
+        if self.version < (3, 3):
+            data = base64.b64decode(data)
+        decrypted_data = decryptor.update(data)
         decrypted_data += decryptor.finalize()
         unpadder = PKCS7(128).unpadder()
         unpadded_data = unpadder.update(decrypted_data)
@@ -190,23 +201,29 @@ class TuyaCipher:
 
         return unpadded_data
 
-    def encrypt(self, data):
+    def encrypt(self, command, data):
-        padder = PKCS7(128).padder()
+        encrypted_data = b''
-        padded_data = padder.update(data)
+        if data:
-        padded_data += padder.finalize()
+            padder = PKCS7(128).padder()
-        encryptor = self.cipher.encryptor()
+            padded_data = padder.update(data)
-        encrypted_data = encryptor.update(padded_data)
+            padded_data += padder.finalize()
-        encrypted_data += encryptor.finalize()
+            encryptor = self.cipher.encryptor()
+            encrypted_data = encryptor.update(padded_data)
+            encrypted_data += encryptor.finalize()
 
-        payload = base64.b64encode(encrypted_data)
+        prefix = '.'.join(map(str, self.version)).encode('utf8')
+        if self.version < (3, 3):
+            payload = base64.b64encode(encrypted_data)
+            hash = self.hash(payload)
+            prefix += hash.encode('utf8')
+        else:
+            payload = encrypted_data
+            if command in (Message.SET_COMMAND, Message.GRATUITOUS_UPDATE):
+                prefix += b'\x00' * 12
+            else:
+                prefix = b''
 
-        hash = self.hash(payload)
+        return prefix + payload
-        result = "{}{}".format(
-            '.'.join(map(str, self.version)),
-            hash
-        ).encode('utf8') + payload
-
-        return result
 
     def hash(self, data):
         digest = Hash(MD5(), backend=openssl_backend)
@@ -253,11 +270,13 @@ class Message:
             self.encrypt = True
 
     def __repr__(self):
-        return "{}({}, {}, {})".format(
+        return "{}({}, {!r}, {!r}, {})".format(
             self.__class__.__name__,
             hex(self.command),
-            repr(self.payload),
+            self.payload,
-            self.sequence)
+            self.sequence,
+            "<Device {}>".format(self.device) if self.device else None
+            )
 
     def hex(self):
         return self.bytes().hex()
@@ -273,9 +292,9 @@ class Message:
             payload_data = payload_data.encode('utf8')
 
         if self.encrypt:
-            payload_data = self.device.cipher.encrypt(payload_data)
+            payload_data = self.device.cipher.encrypt(self.command, payload_data)
 
-        payload_size = len(payload_data) + 8
+        payload_size = len(payload_data) + struct.calcsize(MESSAGE_SUFFIX_FORMAT)
 
         header = struct.pack(
             MESSAGE_PREFIX_FORMAT,
@@ -284,9 +303,13 @@ class Message:
             self.command,
             payload_size,
         )
+        if self.device and self.device.version >= (3, 3):
+            checksum = crc(header + payload_data)
+        else:
+            checksum = crc(payload_data)
         footer = struct.pack(
             MESSAGE_SUFFIX_FORMAT,
-            crc(payload_data),
+            checksum,
             MAGIC_SUFFIX
         )
         return (
@@ -295,6 +318,8 @@ class Message:
             footer
         )
 
+    __bytes__ = bytes
+
     class AsyncWrappedCallback:
         def __init__(self, request, callback):
             self.request = request
@@ -345,54 +370,45 @@ class Message:
         except struct.error as e:
             raise InvalidMessage("Unable to unpack return code.") from e
         if return_code >> 8:
-            payload_data = data[header_size:header_size + payload_size - 8]
+            payload_data = data[header_size:header_size + payload_size - struct.calcsize(MESSAGE_SUFFIX_FORMAT)]
             return_code = None
         else:
-            payload_data = data[header_size + 4:header_size + payload_size - 8]
+            payload_data = data[header_size + struct.calcsize('>I'):header_size + payload_size - struct.calcsize(MESSAGE_SUFFIX_FORMAT)]
 
         try:
             expected_crc, suffix = struct.unpack_from(
                 MESSAGE_SUFFIX_FORMAT,
                 data,
-                header_size + payload_size - 8
+                header_size + payload_size - struct.calcsize(MESSAGE_SUFFIX_FORMAT)
             )
         except struct.error as e:
             raise InvalidMessage("Invalid message suffix format.") from e
         if suffix != MAGIC_SUFFIX:
             raise InvalidMessage("Magic suffix missing from message")
 
-        actual_crc = crc(data[:header_size + payload_size - 8])
+        actual_crc = crc(data[:header_size + payload_size - struct.calcsize(MESSAGE_SUFFIX_FORMAT)])
         if expected_crc != actual_crc:
             raise InvalidMessage("CRC check failed")
 
         payload = None
         if payload_data:
-            try_decrypt = False
+            try:
+                payload_data = cipher.decrypt(command, payload_data)
+            except ValueError as e:
+                pass
             try:
                 payload_text = payload_data.decode('utf8')
-            except UnicodeDecodeError:
+            except UnicodeDecodeError as e:
-                try_decrypt = True
+                _LOGGER.debug(payload_data.hex())
-            else:
+                _LOGGER.error(e)
-                try:
+                raise MessageDecodeFailed() from e
-                    payload = json.loads(payload_text)
-                except json.decoder.JSONDecodeError:
-                    # data may be encrypted
-                    try_decrypt = True
-
-            if try_decrypt:
-                try:
-                    payload_data = cipher.decrypt(payload_data)
-                except ValueError as e:
-                    _LOGGER.debug(payload_data.hex())
-                    _LOGGER.error(e)
-                    raise MessageDecodeFailed() from e
-
             try:
-                payload = payload_data.decode('utf8')
+                payload = json.loads(payload_text)
-                payload = json.loads(payload)
+            except json.decoder.JSONDecodeError as e:
-            except (UnicodeDecodeError, json.decoder.JSONDecodeError):
+                # data may be encrypted
-                # keep it as bytes
+                _LOGGER.debug(payload_data.hex())
-                pass
+                _LOGGER.error(e)
+                raise MessageDecodeFailed() from e
 
         return cls(command, payload, sequence)
 
@@ -418,8 +434,8 @@ class TuyaDevice:
 
     PING_INTERVAL = 10
 
-    def __init__(self, device_id, local_key, host, port=6668, gateway_id=None,
+    def __init__(self, device_id, host, local_key=None, port=6668,
-                 version=(3, 1), timeout=10):
+                 gateway_id=None, version=(3, 3), timeout=10):
         """Initialize the device."""
         self.device_id = device_id
         self.host = host
@@ -445,12 +461,12 @@ class TuyaDevice:
         self._connected = False
 
     def __repr__(self):
-        return "{}({}, {}, {}, {})".format(
+        return "{}({!r}, {!r}, {!r}, {!r})".format(
             self.__class__.__name__,
             self.device_id,
             self.host,
             self.port,
-            self.local_key
+            self.cipher.key
         )
 
     def __str__(self):
@@ -485,13 +501,13 @@ class TuyaDevice:
             'gwId': self.gateway_id,
             'devId': self.device_id
         }
-        message = Message(Message.GET_COMMAND, payload)
+        maybe_self = None if self.version < (3, 3) else self
+        message = Message(Message.GET_COMMAND, payload, encrypt_for=maybe_self)
         return await message.async_send(self, callback)
 
     async def async_set(self, dps, callback=None):
         t = int(time.time())
         payload = {
-            'gwId': self.gateway_id,
             'devId': self.device_id,
             'uid': '',
             't': t,
@@ -505,7 +521,9 @@ class TuyaDevice:
 
     async def _async_ping(self):
         self.last_ping = time.time()
-        message = Message(Message.PING_COMMAND, sequence=0)
+        maybe_self = None if self.version < (3, 3) else self
+        message = Message(Message.PING_COMMAND, sequence=0,
+                          encrypt_for=maybe_self)
         await self._async_send(message)
         await asyncio.sleep(self.PING_INTERVAL)
         if self.last_pong < self.last_ping:
@@ -532,7 +550,7 @@ class TuyaDevice:
         try:
             response_data = await self.reader.readuntil(MAGIC_SUFFIX_BYTES)
         except socket.error as e:
-            _LOGGER.error("Connection to {} failed".format(self))
+            _LOGGER.error("Connection to {} failed: {}".format(self, e))
             asyncio.ensure_future(self.async_disconnect())
             return
 

eufy_robovac/vacuum.py

@@ -66,8 +66,8 @@ class EufyVacuum(VacuumDevice):
             v: k for k, v in self._config['fan_speeds'].items()}
         self._device_id = device_config['device_id']
         self.robovac = robovac.Robovac(
-            device_config['device_id'], device_config['local_key'],
+            device_config['device_id'], device_config['address'],
-            device_config['address'])
+            device_config['local_key'])
         self._name = device_config['name']
 
     async def async_update(self):