Merge pull request #537 from Cougar/fix-533

Rewrite Tornado AsyncModbusSerialClient - #533

Author: dhoomakethu

pymodbus/client/asynchronous/tornado/__init__.py

@@ -7,6 +7,7 @@
 
 import logging
 
+import time
 import socket
 from serial import Serial
 from tornado import gen
@@ -17,8 +18,15 @@
 
 from pymodbus.client.asynchronous.mixins import (AsyncModbusClientMixin,
                                                  AsyncModbusSerialClientMixin)
-from pymodbus.exceptions import ConnectionException
-from pymodbus.utilities import hexlify_packets
+
+from pymodbus.compat import byte2int
+from pymodbus.exceptions import (ConnectionException,
+                                 ModbusIOException,
+                                 TimeOutException)
+from pymodbus.utilities import (hexlify_packets,
+                                ModbusTransactionState)
+from pymodbus.constants import Defaults
+
 
 LOGGER = logging.getLogger(__name__)
 
@@ -291,6 +299,24 @@ class AsyncModbusSerialClient(BaseTornadoSerialClient):
     """
     Tornado based asynchronous serial client
     """
+    def __init__(self, *args, **kwargs):
+        """
+        Initializes AsyncModbusSerialClient.
+        :param args:
+        :param kwargs:
+        """
+        self.state = ModbusTransactionState.IDLE
+        self.timeout = kwargs.get('timeout', Defaults.Timeout)
+        self.baudrate = kwargs.get('baudrate', Defaults.Baudrate)
+        if self.baudrate > 19200:
+            self.silent_interval = 1.75 / 1000  # ms
+        else:
+            self._t0 = float((1 + 8 + 2)) / self.baudrate
+            self.silent_interval = 3.5 * self._t0
+        self.silent_interval = round(self.silent_interval, 6)
+        self.last_frame_end = 0.0
+        super().__init__(*args, **kwargs)
+
     def get_socket(self):
         """
         Creates Pyserial object
@@ -318,6 +344,134 @@ def connect(self):
 
         raise gen.Return(self)
 
+    def execute(self, request):
+        """
+        Executes a transaction
+        :param request: Request to be written on to the bus
+        :return:
+        """
+        request.transaction_id = self.transaction.getNextTID()
+
+        def _clear_timer():
+            """
+            Clear serial waiting timeout
+            """
+            if self.timeout_handle:
+                self.io_loop.remove_timeout(self.timeout_handle)
+                self.timeout_handle = None
+
+        def _on_timeout():
+            """
+            Got timeout while waiting data from serial port
+            """
+            LOGGER.warning("serial receive timeout")
+            _clear_timer()
+            if self.stream:
+                self.io_loop.remove_handler(self.stream.fileno())
+            self.framer.resetFrame()
+            transaction = self.transaction.getTransaction(request.transaction_id)
+            if transaction:
+                transaction.set_exception(TimeOutException())
+
+        def _on_write_done(*args):
+            """
+            Set up reader part after sucessful write to the serial
+            """
+            LOGGER.debug("frame sent, waiting for a reply")
+            self.last_frame_end = round(time.time(), 6)
+            self.state = ModbusTransactionState.WAITING_FOR_REPLY
+            self.io_loop.add_handler(self.stream.fileno(), _on_receive, IOLoop.READ)
+
+        def _on_fd_error(fd, *args):
+            _clear_timer()
+            self.io_loop.remove_handler(fd)
+            self.close()
+            self.transaction.getTransaction(request.transaction_id).set_exception(ModbusIOException(*args))
+
+        def _on_receive(fd, events):
+            """
+            New data in serial buffer to read or serial port closed
+            """
+            if events & IOLoop.ERROR:
+                _on_fd_error(fd)
+                return
+
+            try:
+                waiting = self.stream.connection.in_waiting
+                if waiting:
+                    data = self.stream.connection.read(waiting)
+                    LOGGER.debug(
+                        "recv: " + hexlify_packets(data))
+                    self.last_frame_end = round(time.time(), 6)
+            except OSError as ex:
+                _on_fd_error(fd, ex)
+                return
+
+            self.framer.addToFrame(data)
+
+            # check if we have regular frame or modbus exception
+            fcode = self.framer.decode_data(self.framer.getRawFrame()).get("fcode", 0)
+            if fcode and (
+                  (fcode > 0x80 and len(self.framer.getRawFrame()) == exception_response_length)
+                or
+                  (len(self.framer.getRawFrame()) == expected_response_length)
+            ):
+                _clear_timer()
+                self.io_loop.remove_handler(fd)
+                self.state = ModbusTransactionState.IDLE
+                self.framer.processIncomingPacket(
+                    b'',            # already sent via addToFrame()
+                    self._handle_response,
+                    0,              # don't care when `single=True`
+                    single=True,
+                    tid=request.transaction_id
+                )
+
+        packet = self.framer.buildPacket(request)
+        f = self._build_response(request.transaction_id)
+
+        response_pdu_size = request.get_response_pdu_size()
+        expected_response_length = self.transaction._calculate_response_length(response_pdu_size)
+        LOGGER.debug("expected_response_length = %d", expected_response_length)
+
+        exception_response_length = self.transaction._calculate_exception_length() # TODO: calculate once
+
+        if self.timeout:
+            self.timeout_handle = self.io_loop.add_timeout(time.time() + self.timeout, _on_timeout)
+        self._sendPacket(packet, callback=_on_write_done)
+
+        return f
+
+    def _sendPacket(self, message, callback):
+        """
+        Sends packets on the bus with 3.5char delay between frames
+        :param message: Message to be sent over the bus
+        :return:
+        """
+        @gen.coroutine
+        def sleep(timeout):
+            yield gen.sleep(timeout)
+
+        try:
+            waiting = self.stream.connection.in_waiting
+            if waiting:
+                result = self.stream.connection.read(waiting)
+                LOGGER.info(
+                    "Cleanup recv buffer before send: " + hexlify_packets(result))
+        except OSError as e:
+            self.transaction.getTransaction(request.transaction_id).set_exception(ModbusIOException(e))
+            return
+
+        start = time.time()
+        if self.last_frame_end:
+            waittime = self.last_frame_end + self.silent_interval - start
+            if waittime > 0:
+                LOGGER.debug("Waiting for 3.5 char before next send - %f ms", waittime)
+                sleep(waittime)
+
+        self.state = ModbusTransactionState.SENDING
+        LOGGER.debug("send: " + hexlify_packets(message))
+        self.stream.write(message, callback)
 
 class AsyncModbusTCPClient(BaseTornadoClient):
     """

pymodbus/exceptions.py

@@ -105,6 +105,17 @@ def __init__(self, string=""):
         message = '[Error registering message] %s' % string
         ModbusException.__init__(self, message)
 
+class TimeOutException(ModbusException):
+    """ Error resulting from modbus response timeout """
+
+    def __init__(self, string=""):
+        """ Initialize the exception
+
+        :param string: The message to append to the error
+        """
+        message = "[Timeout] %s" % string
+        ModbusException.__init__(self, message)
+
 
 # --------------------------------------------------------------------------- #
 # Exported symbols
@@ -114,5 +125,5 @@ def __init__(self, string=""):
     "ParameterException", "NotImplementedException",
     "ConnectionException", "NoSuchSlaveException",
     "InvalidMessageReceivedException",
-    "MessageRegisterException"
+    "MessageRegisterException", "TimeOutException"
 ]

test/test_client_async_tornado.py

@@ -228,10 +228,12 @@ def testSerialClientExecute(self, mock_serial, mock_seriostream, mock_ioloop):
         client = AsyncModbusSerialClient(ioloop=schedulers.IO_LOOP,
                                          framer=ModbusRtuFramer(
                                              ClientDecoder()),
-                                         port=SERIAL_PORT)
+                                         port=SERIAL_PORT,
+                                         timeout=0)
         client.connect()
         client.stream = Mock()
         client.stream.write = Mock()
+        client.stream.connection.read.return_value = b''
 
         request = ReadCoilsRequest(1, 1)
         d = client.execute(request)