New nullmodem and transport. (#1696)

Author: janiversen

pymodbus/client/base.py

@@ -13,7 +13,7 @@
 from pymodbus.logging import Log
 from pymodbus.pdu import ModbusRequest, ModbusResponse
 from pymodbus.transaction import DictTransactionManager
-from pymodbus.transport.transport import CommParams, ModbusProtocol
+from pymodbus.transport import CommParams, ModbusProtocol
 from pymodbus.utilities import ModbusTransactionState
 
 

pymodbus/client/serial.py

@@ -10,7 +10,7 @@
 from pymodbus.framer import ModbusFramer
 from pymodbus.framer.rtu_framer import ModbusRtuFramer
 from pymodbus.logging import Log
-from pymodbus.transport.transport import CommType
+from pymodbus.transport import CommType
 from pymodbus.utilities import ModbusTransactionState
 
 

pymodbus/client/tcp.py

@@ -10,7 +10,7 @@
 from pymodbus.framer import ModbusFramer
 from pymodbus.framer.socket_framer import ModbusSocketFramer
 from pymodbus.logging import Log
-from pymodbus.transport.transport import CommType
+from pymodbus.transport import CommType
 from pymodbus.utilities import ModbusTransactionState
 
 

pymodbus/client/tls.py

@@ -7,7 +7,7 @@
 from pymodbus.framer import ModbusFramer
 from pymodbus.framer.tls_framer import ModbusTlsFramer
 from pymodbus.logging import Log
-from pymodbus.transport.transport import CommParams, CommType
+from pymodbus.transport import CommParams, CommType
 
 
 class AsyncModbusTlsClient(AsyncModbusTcpClient):

pymodbus/client/udp.py

@@ -8,7 +8,7 @@
 from pymodbus.framer import ModbusFramer
 from pymodbus.framer.socket_framer import ModbusSocketFramer
 from pymodbus.logging import Log
-from pymodbus.transport.transport import CommType
+from pymodbus.transport import CommType
 
 
 DGRAM_TYPE = socket.SOCK_DGRAM

pymodbus/server/async_io.py

@@ -19,7 +19,7 @@
     ModbusSocketFramer,
     ModbusTlsFramer,
 )
-from pymodbus.transport.transport import CommParams, CommType, ModbusProtocol
+from pymodbus.transport import CommParams, CommType, ModbusProtocol
 
 
 with suppress(ImportError):

pymodbus/transport/__init__.py

@@ -1 +1,22 @@
 """Transport."""
+__all__ = [
+    "CommParams",
+    "CommType",
+    "create_serial_connection",
+    "ModbusProtocol",
+    "NullModem",
+    "NULLMODEM_HOST",
+    "SerialTransport",
+]
+
+from pymodbus.transport.transport import (
+    NULLMODEM_HOST,
+    CommParams,
+    CommType,
+    ModbusProtocol,
+    NullModem,
+)
+from pymodbus.transport.transport_serial import (
+    SerialTransport,
+    create_serial_connection,
+)

pymodbus/transport/transport.py

@@ -1,11 +1,10 @@
 """ModbusProtocol layer."""
-# mypy: disable-error-code="name-defined,union-attr"
-# needed because asyncio.Server is not defined (to mypy) in v3.8.16
 from __future__ import annotations
 
 import asyncio
 import dataclasses
 import ssl
+from contextlib import suppress
 from enum import Enum
 from typing import Any, Callable, Coroutine
 
@@ -124,7 +123,7 @@ def __init__(
         self.is_server = is_server
         self.is_closing = False
 
-        self.transport: asyncio.BaseModbusProtocol | asyncio.Server = None
+        self.transport: asyncio.BaseTransport = None
         self.loop: asyncio.AbstractEventLoop = None
         self.recv_buffer: bytes = b""
         self.call_create: Callable[[], Coroutine[Any, Any, Any]] = lambda: None
@@ -258,7 +257,7 @@ async def transport_listen(self) -> bool:
     # ---------------------------------- #
     # ModbusProtocol asyncio standard methods #
     # ---------------------------------- #
-    def connection_made(self, transport: asyncio.BaseModbusProtocol):
+    def connection_made(self, transport: asyncio.BaseTransport):
         """Call from asyncio, when a connection is made.
 
         :param transport: socket etc. representing the connection.
@@ -298,10 +297,23 @@ def datagram_received(self, data: bytes, addr: tuple):
                 self.sent_buffer = b""
             if not data:
                 return
-        Log.debug("recv: {} addr={}", data, ":hex", addr)
+        Log.debug(
+            "recv: {} old_data: {} addr={}",
+            data,
+            ":hex",
+            self.recv_buffer,
+            ":hex",
+            addr,
+        )
         self.recv_buffer += data
         cut = self.callback_data(self.recv_buffer, addr=addr)
         self.recv_buffer = self.recv_buffer[cut:]
+        if self.recv_buffer:
+            Log.debug(
+                "recv, unused data waiting for next packet: {}",
+                self.recv_buffer,
+                ":hex",
+            )
 
     def eof_received(self):
         """Accept other end terminates connection."""
@@ -342,11 +354,11 @@ def transport_send(self, data: bytes, addr: tuple = None) -> None:
             self.sent_buffer = data
         if self.comm_params.comm_type == CommType.UDP:
             if addr:
-                self.transport.sendto(data, addr=addr)
+                self.transport.sendto(data, addr=addr)  # type: ignore[attr-defined]
             else:
-                self.transport.sendto(data)
+                self.transport.sendto(data)  # type: ignore[attr-defined]
         else:
-            self.transport.write(data)
+            self.transport.write(data)  # type: ignore[attr-defined]
 
     def transport_close(self, intern: bool = False, reconnect: bool = False) -> None:
         """Close connection.
@@ -392,26 +404,11 @@ async def create_nullmodem(self, port):
         """Bypass create_ and use null modem"""
         if self.is_server:
             # Listener object
-            self.transport = NullModem(self)
-            NullModem.listener_new_connection[port] = self.handle_new_connection
+            self.transport = NullModem.set_listener(port, self)
             return self.transport, self
 
         # connect object
-        client_protocol = self.handle_new_connection()
-        try:
-            server_protocol = NullModem.listener_new_connection[port]()
-        except KeyError as exc:
-            raise asyncio.TimeoutError(
-                f"No listener on port {self.comm_params.port} for connect"
-            ) from exc
-
-        client_transport = NullModem(client_protocol)
-        server_transport = NullModem(server_protocol)
-        client_transport.other_transport = server_transport
-        server_transport.other_transport = client_transport
-        client_protocol.connection_made(client_transport)
-        server_protocol.connection_made(server_transport)
-        return client_transport, client_protocol
+        return NullModem.set_connection(port, self)
 
     def handle_new_connection(self):
         """Handle incoming connect."""
@@ -468,46 +465,117 @@ class NullModem(asyncio.DatagramTransport, asyncio.Transport):
     (Allowing tests to be shortcut without actual network calls)
     """
 
-    listener_new_connection: dict[int, ModbusProtocol] = {}
+    listeners: dict[int, ModbusProtocol] = {}
+    connections: dict[NullModem, int] = {}
 
-    def __init__(self, protocol: ModbusProtocol):
+    def __init__(self, protocol: ModbusProtocol, listen: int = None) -> None:
         """Create half part of null modem"""
         asyncio.DatagramTransport.__init__(self)
         asyncio.Transport.__init__(self)
-        self.other: NullModem = None
-        self.protocol: ModbusProtocol | asyncio.BaseProtocol = protocol
+        self.protocol: ModbusProtocol = protocol
         self.serving: asyncio.Future = asyncio.Future()
-        self.other_transport: NullModem = None
+        self.other_modem: NullModem = None
+        self.listen = listen
+        self.manipulator: Callable[[bytes], list[bytes]] = None
+        self._is_closing = False
+
+    # -------------------------- #
+    # external nullmodem methods #
+    # -------------------------- #
+    @classmethod
+    def set_listener(cls, port: int, parent: ModbusProtocol) -> NullModem:
+        """Register listener."""
+        if port in cls.listeners:
+            raise AssertionError(f"Port {port} already listening !")
+        cls.listeners[port] = parent
+        return NullModem(parent, listen=port)
+
+    @classmethod
+    def set_connection(
+        cls, port: int, parent: ModbusProtocol
+    ) -> tuple[NullModem, ModbusProtocol]:
+        """Connect to listener."""
+        if port not in cls.listeners:
+            raise asyncio.TimeoutError(f"Port {port} not being listened on !")
+
+        client_protocol = parent.handle_new_connection()
+        server_protocol = NullModem.listeners[port].handle_new_connection()
+        client_transport = NullModem(client_protocol)
+        server_transport = NullModem(server_protocol)
+        cls.connections[client_transport] = port
+        cls.connections[server_transport] = -port
+        client_transport.other_modem = server_transport
+        server_transport.other_modem = client_transport
+        client_protocol.connection_made(client_transport)
+        server_protocol.connection_made(server_transport)
+        return client_transport, client_protocol
+
+    def set_manipulator(self, function: Callable[[bytes], list[bytes]]) -> None:
+        """Register a manipulator."""
+        self.manipulator = function
+
+    @classmethod
+    def is_dirty(cls):
+        """Check if everything is closed."""
+        dirty = False
+        if cls.connections:
+            Log.error(
+                "NullModem_FATAL missing close on port {} connect()",
+                [str(key) for key in cls.connections.values()],
+            )
+            dirty = True
+        if cls.listeners:
+            Log.error(
+                "NullModem_FATAL missing close on port {} listen()",
+                [str(value) for value in cls.listeners],
+            )
+            dirty = True
+        return dirty
 
     # ---------------- #
     # external methods #
     # ---------------- #
 
-    def close(self):
+    def close(self) -> None:
         """Close null modem"""
+        if self._is_closing:
+            return
+        self._is_closing = True
         if not self.serving.done():
             self.serving.set_result(True)
-        if self.other_transport:
-            self.other_transport.other_transport = None
-            self.other_transport.protocol.connection_lost(None)
-            self.other_transport = None
+        if self.listen:
+            del self.listeners[self.listen]
+            return
+        if self.connections:
+            with suppress(KeyError):
+                del self.connections[self]
+        if self.other_modem:
+            self.other_modem.other_modem = None
+            self.other_modem.close()
+            self.other_modem = None
+        if self.protocol:
             self.protocol.connection_lost(None)
 
-    def sendto(self, data: bytes, _addr: Any = None):
+    def sendto(self, data: bytes, _addr: Any = None) -> None:
         """Send datagrame"""
-        return self.write(data)
+        self.write(data)
 
-    def write(self, data: bytes):
+    def write(self, data: bytes) -> None:
         """Send data"""
-        self.other_transport.protocol.data_received(data)
+        if not self.manipulator:
+            self.other_modem.protocol.data_received(data)
+            return
+        data_manipulated = self.manipulator(data)
+        for part in data_manipulated:
+            self.other_modem.protocol.data_received(part)
 
-    async def serve_forever(self):
+    async def serve_forever(self) -> None:
         """Serve forever"""
         await self.serving
 
-    # ---------------- #
-    # Abstract methods #
-    # ---------------- #
+    # ------------- #
+    # Dummy methods #
+    # ------------- #
     def abort(self) -> None:
         """Abort connection."""
         self.close()
@@ -536,11 +604,10 @@ def get_protocol(self) -> ModbusProtocol | asyncio.BaseProtocol:
 
     def set_protocol(self, protocol: asyncio.BaseProtocol) -> None:
         """Set current protocol."""
-        self.protocol = protocol
 
     def is_closing(self) -> bool:
         """Return true if closing"""
-        return False
+        return self._is_closing
 
     def is_reading(self) -> bool:
         """Return true if read is active."""

test/conftest.py

@@ -16,6 +16,22 @@ def pytest_configure():
 # -----------------------------------------------------------------------#
 # Generic fixtures
 # -----------------------------------------------------------------------#
+BASE_PORTS = {
+    "TestBasicModbusProtocol": 8100,
+    "TestBasicSerial": 8200,
+    "TestCommModbusProtocol": 8300,
+    "TestCommNullModem": 8400,
+    "TestExamples": 8500,
+    "TestModbusProtocol": 8600,
+    "TestNullModem": 8700,
+    "TestReconnectModbusProtocol": 8800,
+}
+
+
+@pytest.fixture(name="base_ports", scope="package")
+def get_base_ports():
+    """Return base_ports"""
+    return BASE_PORTS
 
 
 class MockContext(ModbusBaseSlaveContext):

test/sub_examples/conftest.py

@@ -6,7 +6,7 @@
 
 from examples.server_async import run_async_server, setup_server
 from pymodbus.server import ServerAsyncStop
-from pymodbus.transport.transport import NULLMODEM_HOST
+from pymodbus.transport import NULLMODEM_HOST
 
 
 @pytest.fixture(name="port_offset")