Retry updates on empty or error response for sync clients

Author: dhoomakethu

pymodbus/framer/rtu_framer.py

@@ -282,6 +282,11 @@ def sendPacket(self, message):
                     # Recovering from last error ??
                     time.sleep(self.client.silent_interval)
                 self.client.state = ModbusTransactionState.IDLE
+            elif self.client.state == ModbusTransactionState.RETRYING:
+                # Simple lets settle down!!!
+                # To check for higher baudrates
+                time.sleep(self.client.timeout)
+                break
             else:
                 if time.time() > timeout:
                     _logger.debug("Spent more time than the read time out, "

pymodbus/transaction.py

@@ -64,8 +64,10 @@ def __init__(self, client, **kwargs):
         self.tid = Defaults.TransactionId
         self.client = client
         self.backoff = kwargs.get('backoff', Defaults.Backoff) or 0.3
-        self.retry_on_empty = kwargs.get('retry_on_empty', Defaults.RetryOnEmpty)
-        self.retry_on_invalid = kwargs.get('retry_on_invalid', Defaults.RetryOnInvalid)
+        self.retry_on_empty = kwargs.get('retry_on_empty',
+                                         Defaults.RetryOnEmpty)
+        self.retry_on_invalid = kwargs.get('retry_on_invalid',
+                                           Defaults.RetryOnInvalid)
         self.retries = kwargs.get('retries', Defaults.Retries) or 1
         self._transaction_lock = RLock()
         self._no_response_devices = []
@@ -108,6 +110,25 @@ def _calculate_exception_length(self):
 
         return None
 
+    def _validate_response(self, request, response, exp_resp_len):
+        """
+        Validate Incoming response against request
+        :param request: Request sent
+        :param response: Response received
+        :param exp_resp_len: Expected response length
+        :return: New transactions state
+        """
+        if not response:
+            return False
+
+        mbap = self.client.framer.decode_data(response)
+        if mbap.get('unit') != request.unit_id or mbap.get('fcode') != request.function_code:
+            return False
+
+        if 'length' in mbap and exp_resp_len:
+            return mbap.get('length') == exp_resp_len
+        return True
+
     def execute(self, request):
         """ Starts the producer to send the next request to
         consumer.write(Frame(request))
@@ -132,6 +153,7 @@ def execute(self, request):
                     self._transact(request, None, broadcast=True)
                     response = b'Broadcast write sent - no response expected'
                 else:
+                    invalid_response = False
                     expected_response_length = None
                     if not isinstance(self.client.framer, ModbusSocketFramer):
                         if hasattr(request, "get_response_pdu_size"):
@@ -149,35 +171,47 @@ def execute(self, request):
                         full = True
                         if not expected_response_length:
                             expected_response_length = Defaults.ReadSize
-                    retries += 1
+                    # retries += 1
                     while retries > 0:
                         response, last_exception = self._transact(
                             request,
                             expected_response_length,
                             full=full,
                             broadcast=broadcast
                         )
-                        if not response and (
-                                request.unit_id not in self._no_response_devices):
+                        valid_response = self._validate_response(
+                            request, response, expected_response_length
+                        )
+                        if not response and request.unit_id \
+                                not in self._no_response_devices:
                             self._no_response_devices.append(request.unit_id)
                         elif request.unit_id in self._no_response_devices and response:
                             self._no_response_devices.remove(request.unit_id)
                         if not response and self.retry_on_empty:
                             _logger.debug("Retry on empty - {}".format(retries))
+                            retries -= 1
+                            _logger.debug("Changing transaction state from "
+                                          "'WAITING_FOR_REPLY' to 'RETRYING'")
+                            self.client.state = ModbusTransactionState.RETRYING
+                            continue
                         elif not response:
                             break
-                        if not self.retry_on_invalid:
-                            break
                         mbap = self.client.framer.decode_data(response)
-                        if (mbap.get('unit') == request.unit_id):
+                        if mbap.get('unit') == request.unit_id:
                             break
                         if ('length' in mbap and expected_response_length and
-                            mbap.get('length') == expected_response_length):
+                            mbap.get('length') == expected_response_length and
+                            mbap.get('fcode') == request.function_code):
+                            break
+                        else:
+                            invalid_response = True
+                        if invalid_response and not self.retry_on_invalid:
                             break
                         _logger.debug("Retry on invalid - {}".format(retries))
                         if hasattr(self.client, "state"):
-                            _logger.debug("RESETTING Transaction state to 'IDLE' for retry")
-                            self.client.state = ModbusTransactionState.IDLE
+                            _logger.debug("RESETTING Transaction "
+                                          "state to 'RETRY' for retry")
+                            self.client.state = ModbusTransactionState.RETRYING
                         if self.backoff:
                             delay = 2 ** (self.retries - retries) * self.backoff
                             time.sleep(delay)
@@ -206,14 +240,26 @@ def execute(self, request):
                                       "'TRANSACTION_COMPLETE'")
                         self.client.state = (
                             ModbusTransactionState.TRANSACTION_COMPLETE)
+                self.client.close()
                 return response
             except ModbusIOException as ex:
                 # Handle decode errors in processIncomingPacket method
                 _logger.exception(ex)
+                self.client.close()
                 self.client.state = ModbusTransactionState.TRANSACTION_COMPLETE
                 return ex
 
-    def _transact(self, packet, response_length, full=False, broadcast=False):
+    def _retry(self, packet, response_length, full=False):
+        self.client.connect()
+        in_waiting = self.client._in_waiting()
+        if in_waiting:
+            if response_length == in_waiting:
+                result = self._recv(response_length, full)
+                return result, None
+        return self._transact(packet, response_length, full=full)
+
+    def _transact(self, packet, response_length,
+                  full=False, broadcast=False):
         """
         Does a Write and Read transaction
         :param packet: packet to be sent
@@ -229,6 +275,11 @@ def _transact(self, packet, response_length, full=False, broadcast=False):
             if _logger.isEnabledFor(logging.DEBUG):
                 _logger.debug("SEND: " + hexlify_packets(packet))
             size = self._send(packet)
+            if isinstance(size, bytes) and self.client.state == ModbusTransactionState.RETRYING:
+                _logger.debug("Changing transaction state from "
+                              "'RETRYING' to 'PROCESSING REPLY'")
+                self.client.state = ModbusTransactionState.PROCESSING_REPLY
+                return size, None
             if broadcast:
                 if size:
                     _logger.debug("Changing transaction state from 'SENDING' "
@@ -244,6 +295,7 @@ def _transact(self, packet, response_length, full=False, broadcast=False):
                 if local_echo_packet != packet:
                     return b'', "Wrong local echo"
             result = self._recv(response_length, full)
+            # result2 = self._recv(response_length, full)
             if _logger.isEnabledFor(logging.DEBUG):
                     _logger.debug("RECV: " + hexlify_packets(result))
 
@@ -255,7 +307,7 @@ def _transact(self, packet, response_length, full=False, broadcast=False):
             result = b''
         return result, last_exception
 
-    def _send(self, packet):
+    def _send(self, packet, retrying=False):
         return self.client.framer.sendPacket(packet)
 
     def _recv(self, expected_response_length, full):

pymodbus/utilities.py

@@ -20,6 +20,8 @@ class ModbusTransactionState(object):
     PROCESSING_REPLY = 4
     PROCESSING_ERROR = 5
     TRANSACTION_COMPLETE = 6
+    RETRYING = 7
+    NO_RESPONSE_STATE = 8
 
     @classmethod
     def to_string(cls, state):
@@ -30,7 +32,8 @@ def to_string(cls, state):
             ModbusTransactionState.WAITING_TURNAROUND_DELAY: "WAITING_TURNAROUND_DELAY",
             ModbusTransactionState.PROCESSING_REPLY: "PROCESSING_REPLY",
             ModbusTransactionState.PROCESSING_ERROR: "PROCESSING_ERROR",
-            ModbusTransactionState.TRANSACTION_COMPLETE: "TRANSACTION_COMPLETE"
+            ModbusTransactionState.TRANSACTION_COMPLETE: "TRANSACTION_COMPLETE",
+            ModbusTransactionState.RETRYING: "RETRYING TRANSACTION",
         }
         return states.get(state, None)