Add retry logic to recieving messages

This commit is contained in:
Luke Bonaccorsi 2023-09-28 12:49:14 +01:00
parent 40480cc319
commit bea3817c5e
1 changed files with 28 additions and 10 deletions

View File

@ -347,6 +347,10 @@ class RequestResponseCommandMismatch(TuyaException):
"""The command in the response didn't match the one from the request.""" """The command in the response didn't match the one from the request."""
class ResponseTimeoutException(TuyaException):
"""Did not recieve a response to the request within the timeout"""
class TuyaCipher: class TuyaCipher:
"""Tuya cryptographic helpers.""" """Tuya cryptographic helpers."""
@ -445,11 +449,11 @@ class Message:
payload = b"" payload = b""
self.payload = payload self.payload = payload
self.command = command self.command = command
self.original_sequence = sequence
if sequence is None: if sequence is None:
# Use millisecond process time as the sequence number. Not ideal, self.set_sequence()
# but good for one month's continuous connection time though. else:
sequence = int(time.perf_counter() * 1000) & 0xFFFFFFFF self.sequence = sequence
self.sequence = sequence
self.encrypt = False self.encrypt = False
self.device = None self.device = None
if encrypt_for is not None: if encrypt_for is not None:
@ -465,6 +469,9 @@ class Message:
"<Device {}>".format(self.device) if self.device else None, "<Device {}>".format(self.device) if self.device else None,
) )
def set_sequence(self):
self.sequence = int(time.perf_counter() * 1000) & 0xFFFFFFFF
def hex(self): def hex(self):
return self.bytes().hex() return self.bytes().hex()
@ -496,7 +503,7 @@ class Message:
__bytes__ = bytes __bytes__ = bytes
async def async_send(self, device): async def async_send(self, device, retries=4):
device._listeners[self.sequence] = asyncio.Semaphore(0) device._listeners[self.sequence] = asyncio.Semaphore(0)
await device._async_send(self) await device._async_send(self)
try: try:
@ -504,13 +511,24 @@ class Message:
device._listeners[self.sequence].acquire(), timeout=device.timeout device._listeners[self.sequence].acquire(), timeout=device.timeout
) )
except: except:
del device._listeners[self.sequence]
if retries == 0:
raise ResponseTimeoutException(
"Timed out waiting for response to sequence number {}".format(
self.sequence
)
)
_LOGGER.debug( _LOGGER.debug(
"Timed out waiting for response to sequence number {}".format( "Timed out waiting for response to sequence number {}. Retrying".format(
self.sequence self.sequence
) )
) )
del device._listeners[self.sequence]
raise if self.original_sequence is None:
self.set_sequence()
return self.async_send(device, retries - 1)
return device._listeners.pop(self.sequence) return device._listeners.pop(self.sequence)
@ -734,9 +752,9 @@ class TuyaDevice:
response_data = await self.reader.readuntil(MAGIC_SUFFIX_BYTES) response_data = await self.reader.readuntil(MAGIC_SUFFIX_BYTES)
message = Message.from_bytes(response_data, self.cipher) message = Message.from_bytes(response_data, self.cipher)
except InvalidMessage as e: except InvalidMessage as e:
_LOGGER.error("Invalid message from {}: {}".format(self, e)) _LOGGER.debug("Invalid message from {}: {}".format(self, e))
except MessageDecodeFailed as e: except MessageDecodeFailed as e:
_LOGGER.error("Failed to decrypt message from {}".format(self)) _LOGGER.debug("Failed to decrypt message from {}".format(self))
else: else:
_LOGGER.debug("Received message from {}: {}".format(self, message)) _LOGGER.debug("Received message from {}: {}".format(self, message))
if message.sequence in self._listeners: if message.sequence in self._listeners: