Add ModbusFrameGenerator class to generate and parse Modbus frames without actual communication.

Author: Mateusz Frasunkiewicz

examples/frame_generator.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Modbus Frame Generator Example.
+
+This example shows how to use ModbusFrameGenerator to create and parse
+Modbus frames without actual communication.
+"""
+from pymodbus.client import ModbusFrameGenerator
+from pymodbus.framer import FramerType
+
+
+def run_frame_generator():
+    """Run frame generator example."""
+    print("### Modbus Frame Generator Example ###")
+    
+    # Create generator instance (RTU mode, slave address 1)
+    generator = ModbusFrameGenerator(framer=FramerType.RTU, slave=1)
+
+    print("\n1. Generate Request Frames:")
+    print("-" * 40)
+
+    # Read Coils Request (FC=1)
+    read_coils_frame = generator.read_coils(address=100, count=8)
+    print(f"Read Coils (addr=100, count=8):")
+    print(f"  Frame: {read_coils_frame.hex()}")
+
+    # Write Single Coil Request (FC=5)
+    write_coil_frame = generator.write_coil(address=100, value=True)
+    print(f"\nWrite Single Coil (addr=100, value=ON):")
+    print(f"  Frame: {write_coil_frame.hex()}")
+
+    # Read Holding Registers Request (FC=3)
+    read_regs_frame = generator.read_holding_registers(address=100, count=2)
+    print(f"\nRead Holding Registers (addr=100, count=2):")
+    print(f"  Frame: {read_regs_frame.hex()}")
+
+    # Write Single Register Request (FC=6)
+    write_reg_frame = generator.write_register(address=100, value=1234)
+    print(f"\nWrite Single Register (addr=100, value=1234):")
+    print(f"  Frame: {write_reg_frame.hex()}")
+
+    print("\n2. Parse Response Frames:")
+    print("-" * 40)
+
+    # Parse Read Coils Response (with CRC)
+    coils_response = b"\x01\x01\x01\xCD\x81\x88"  # slave 1, FC 1, 1 byte, value 0xCD + CRC
+    decoded = generator.parse_response(coils_response)
+    print("Read Coils Response:")
+    print(f"  Raw: {coils_response.hex()}")
+    print(f"  Decoded: {decoded}")
+
+    # Parse Read Holding Registers Response (with CRC)
+    regs_response = b"\x01\x03\x04\x00\x0A\x00\x0B\x41\x83"  # slave 1, FC 3, 4 bytes, values [10, 11] + CRC
+    decoded = generator.parse_response(regs_response)
+    print("\nRead Registers Response:")
+    print(f"  Raw: {regs_response.hex()}")
+    print(f"  Decoded: {decoded}")
+
+
+if __name__ == "__main__":
+    run_frame_generator() 

pymodbus/client/__init__.py

@@ -11,10 +11,15 @@
     "ModbusTcpClient",
     "ModbusTlsClient",
     "ModbusUdpClient",
+    "ModbusFrameGenerator",
 ]
 
 from pymodbus.client.base import ModbusBaseClient, ModbusBaseSyncClient
-from pymodbus.client.serial import AsyncModbusSerialClient, ModbusSerialClient
+from pymodbus.client.serial import (
+    AsyncModbusSerialClient,
+    ModbusSerialClient,
+    ModbusFrameGenerator,
+)
 from pymodbus.client.tcp import AsyncModbusTcpClient, ModbusTcpClient
 from pymodbus.client.tls import AsyncModbusTlsClient, ModbusTlsClient
 from pymodbus.client.udp import AsyncModbusUdpClient, ModbusUdpClient

pymodbus/client/serial.py

@@ -9,9 +9,9 @@
 
 from pymodbus.client.base import ModbusBaseClient, ModbusBaseSyncClient
 from pymodbus.exceptions import ConnectionException
-from pymodbus.framer import FramerType
+from pymodbus.framer import FramerType, FramerAscii, FramerRTU
 from pymodbus.logging import Log
-from pymodbus.pdu import ModbusPDU
+from pymodbus.pdu import ModbusPDU, DecodePDU
 from pymodbus.transport import CommParams, CommType
 
 
@@ -318,3 +318,79 @@ def __repr__(self):
             f"<{self.__class__.__name__} at {hex(id(self))} socket={self.socket}, "
             f"framer={self.framer}, timeout={self.comm_params.timeout_connect}>"
         )
+
+
+class ModbusFrameGenerator(ModbusSerialClient):
+    """A class to generate and parse Modbus frames without actual communication.
+    
+    This class provides a way to generate Modbus frames and parse responses
+    without actual serial communication.
+
+    :param framer: The framer to use (RTU or ASCII)
+    :param slave: Default slave address
+    
+    Example::
+        from pymodbus.client import ModbusFrameGenerator
+        from pymodbus.framer import FramerType
+        
+        generator = ModbusFrameGenerator(framer=FramerType.RTU, slave=1)
+        request = generator.read_coils(1, 10)
+        print(f"Generated frame: {request.hex()}")
+    """
+    
+    def __init__(
+        self,
+        framer: FramerType = FramerType.RTU,
+        slave: int = 0x00,
+    ) -> None:
+        """Initialize the frame generator."""
+        if framer not in [FramerType.ASCII, FramerType.RTU]:
+            raise TypeError("Only FramerType RTU/ASCII allowed.")
+        super().__init__(
+            port="DUMMY",
+            framer=framer,
+        )
+        self.slave = slave
+        self.socket = None  # Prevent actual serial operations
+        self.comm_params.timeout_connect = 0  # Prevent retries
+        self.retries = 0  # Disable retries
+
+    def build_request(self, request: ModbusPDU) -> bytes:
+        """Build a complete frame from a request PDU.
+        
+        :param request: The request PDU to frame
+        :returns: The framed request
+        """
+        request.dev_id = self.slave
+        return self.framer.buildFrame(request)
+
+    def execute(self, no_response_expected: bool, request: ModbusPDU) -> bytes:
+        """Override execute to only generate frame without sending."""
+        return self.build_request(request)
+
+    def send(self, request: bytes, addr: tuple | None = None) -> int:
+        """Don't actually send, just return the length of the request."""
+        return len(request) if request else 0
+
+    def recv(self, size: int | None) -> bytes:
+        """Dummy receive - always returns empty bytes."""
+        return b""
+
+    def close(self):
+        """Dummy close - does nothing."""
+        pass
+
+    def is_socket_open(self) -> bool:
+        """Always returns False since we're not actually connecting."""
+        return False
+
+    def parse_response(self, response: bytes) -> ModbusPDU:
+        """Parse a response frame into a PDU.
+        
+        :param response: The response frame to parse
+        :returns: The decoded PDU
+        """
+        used_len, pdu = self.framer.processIncomingFrame(response)
+        if not pdu:
+            raise ValueError("Failed to decode response")
+        return pdu

test_frame_generator.py

@@ -0,0 +1 @@
+