Add AsyncModbusUnixClient.

Author: janiversen

doc/source/library/pymodbus.client.rst

@@ -72,6 +72,11 @@ TCP transport.
 
 .. autoclass:: pymodbus.client.tcp.ModbusTcpClient
 
+Unix domain socket transport.
+~~~~~~~~~~~~~~
+
+.. autoclass:: pymodbus.client.unix.AsyncModbusUnixClient
+
 TLS transport.
 ~~~~~~~~~~~~~~
 

pymodbus/client/__init__.py

@@ -7,6 +7,7 @@
 from pymodbus.client.tcp import AsyncModbusTcpClient, ModbusTcpClient
 from pymodbus.client.tls import AsyncModbusTlsClient, ModbusTlsClient
 from pymodbus.client.udp import AsyncModbusUdpClient, ModbusUdpClient
+from pymodbus.client.unix import AsyncModbusUnixClient
 
 
 # ---------------------------------------------------------------------------#
@@ -17,6 +18,7 @@
     "AsyncModbusTcpClient",
     "AsyncModbusTlsClient",
     "AsyncModbusUdpClient",
+    "AsyncModbusUnixClient",
     "ModbusBaseClient",
     "ModbusSerialClient",
     "ModbusTcpClient",

pymodbus/client/unix.py

@@ -0,0 +1,143 @@
+"""Modbus client async unix domain socket communication."""
+import asyncio
+import logging
+import select
+import socket
+import time
+import typing
+
+from pymodbus.client.base import ModbusBaseClient, ModbusClientProtocol
+from pymodbus.constants import Defaults
+from pymodbus.exceptions import ConnectionException
+from pymodbus.framer import ModbusFramer
+from pymodbus.framer.socket_framer import ModbusSocketFramer
+from pymodbus.utilities import ModbusTransactionState
+
+
+_logger = logging.getLogger(__name__)
+
+
+class AsyncModbusUnixClient(ModbusBaseClient):
+    """**AsyncModbusUnixClient**.
+
+    :param path: unix domain socket path
+    :param framer: (optional) Framer class
+    :param kwargs: (optional) Experimental parameters
+
+    Example::
+
+        from pymodbus.client import AsyncModbusUnixClient
+
+        async def run():
+            client = AsyncModbusTcpClient("/tmp/unix_socket")
+
+            await client.connect()
+            ...
+            await client.close()
+    """
+
+    def __init__(
+        self,
+        path: str,
+        framer: ModbusFramer = ModbusSocketFramer,
+        **kwargs: any,
+    ) -> None:
+        """Initialize Asyncio Modbus TCP Client."""
+        self.protocol = None
+        super().__init__(framer=framer, **kwargs)
+        self.params.host = path
+        self.loop = None
+        self.connected = False
+        self.delay_ms = self.params.reconnect_delay
+
+    async def connect(self):  # pylint: disable=invalid-overridden-method
+        """Initiate connection to start client."""
+        # force reconnect if required:
+        self.loop = asyncio.get_running_loop()
+
+        txt = f"Connecting to {self.params.host}:{self.params.port}."
+        _logger.debug(txt)
+        return await self._connect()
+
+    async def close(self):  # pylint: disable=invalid-overridden-method
+        """Stop client."""
+
+        # prevent reconnect:
+        self.delay_ms = 0
+        if self.connected:
+            if self.protocol.transport:
+                self.protocol.transport.close()
+            if self.protocol:
+                await self.protocol.close()
+            await asyncio.sleep(0.1)
+
+    def _create_protocol(self):
+        """Create initialized protocol instance with factory function."""
+        protocol = ModbusClientProtocol(
+            framer=self.params.framer,
+            xframer=self.framer,
+            timeout=self.params.timeout,
+            retries=self.params.retries,
+            retry_on_empty=self.params.retry_on_empty,
+            close_comm_on_error=self.params.close_comm_on_error,
+            strict=self.params.strict,
+            broadcast_enable=self.params.broadcast_enable,
+            reconnect_delay=self.params.reconnect_delay,
+            reconnect_delay_max=self.params.reconnect_delay_max,
+            **self.params.kwargs,
+        )
+        protocol.factory = self
+        return protocol
+
+    async def _connect(self):
+        """Connect."""
+        _logger.debug("Connecting.")
+        try:
+            transport, protocol = await asyncio.wait_for(
+                self.loop.create_unix_connection(
+                    self._create_protocol, path=self.params.host),
+                timeout=self.params.timeout,
+            )
+        except Exception as exc:  # pylint: disable=broad-except
+            txt = f"Failed to connect: {exc}"
+            _logger.warning(txt)
+            if self.delay_ms > 0:
+                asyncio.ensure_future(self._reconnect())
+        else:
+            txt = f"Connected to {self.params.host}:{self.params.port}."
+            _logger.info(txt)
+            self.reset_delay()
+            return transport, protocol
+
+    def protocol_made_connection(self, protocol):
+        """Notify successful connection."""
+        _logger.info("Protocol made connection.")
+        if not self.connected:
+            self.connected = True
+            self.protocol = protocol
+        else:
+            _logger.error("Factory protocol connect callback called while connected.")
+
+    def protocol_lost_connection(self, protocol):
+        """Notify lost connection."""
+        _logger.info("Protocol lost connection.")
+        if protocol is not self.protocol:
+            _logger.error(
+                "Factory protocol callback called from unexpected protocol instance."
+            )
+
+        self.connected = False
+        if self.protocol is not None:
+            del self.protocol
+            self.protocol = None
+        if self.delay_ms > 0:
+            asyncio.ensure_future(self._reconnect())
+
+    async def _reconnect(self):
+        """Reconnect."""
+        txt = f"Waiting {self.delay_ms} ms before next connection attempt."
+        _logger.debug(txt)
+        await asyncio.sleep(self.delay_ms / 1000)
+        self.delay_ms = min(2 * self.delay_ms, self.params.reconnect_delay_max)
+
+        return await self._connect()

pymodbus/server/async_io.py

@@ -484,7 +484,7 @@ def _send_(self, data):
 
 
 class ModbusUnixServer:
-    """A modbus threaded Unix socket server.
+    """A modbus Unix domain socket server.
 
     We inherit and overload the socket server so that we
     can control the client threads as well as have a single
@@ -1107,7 +1107,7 @@ async def StartAsyncUnixServer(  # pylint: disable=invalid-name,dangerous-defaul
     defer_start=False,
     **kwargs,
 ):
-    """Start and run a tcp modbus server.
+    """Start and run a unix domain socket modbus server.
 
     :param context: The ModbusServerContext datastore
     :param identity: An optional identify structure