### Install pyModbusTCP from GitHub Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/quickstart/index.md Clone the repository from GitHub and install the package. Ensure you specify the correct Python target version. ```default git clone https://github.com/sourceperl/pyModbusTCP.git cd pyModbusTCP # here change "python" by your python target(s) version(s) (like python3.9) sudo python setup.py install ``` -------------------------------- ### Install Twine Source: https://github.com/sourceperl/pymodbustcp/blob/master/HOWTO-PyPi.md Install the twine package, a utility for uploading Python packages to PyPI. This command requires root privileges. ```bash sudo pip install twine ``` -------------------------------- ### Install pyModbusTCP using pip Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/quickstart/index.md Install the latest stable version or upgrade an existing installation using pip. You can also install a specific version. ```default sudo pip3 install pyModbusTCP sudo pip3 install pyModbusTCP --upgrade sudo pip3 install pyModbusTCP==v0.1.10 ``` -------------------------------- ### Install pyModbusTCP (Stable Release) Source: https://github.com/sourceperl/pymodbustcp/blob/master/README.rst Installs the latest stable release of the pyModbusTCP library using pip. Ensure you have pip installed and sufficient permissions. ```bash sudo pip install pyModbusTCP ``` -------------------------------- ### Install pyModbusTCP from GitHub Source: https://github.com/sourceperl/pymodbustcp/blob/master/README.rst Installs a specific version of the pyModbusTCP library directly from GitHub. This is useful for development or testing specific commits. ```bash sudo pip install git+https://github.com/sourceperl/pyModbusTCP.git@v0.1.10 ``` -------------------------------- ### Minimal Modbus Client Code Example Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/client_minimal.md This snippet shows the most basic way to instantiate a ModbusClient and read coils. Ensure the Modbus server is running on localhost. ```python #!/usr/bin/env python3 """ Minimal code example. """ from pyModbusTCP.client import ModbusClient # read 3 coils at @0 on localhost server print('coils=%s' % ModbusClient().read_coils(0, 3)) ``` -------------------------------- ### Run Basic Modbus/TCP Server Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server.md This script initializes and starts a Modbus/TCP server. It supports custom host and port configurations via command-line arguments and enables debug logging if the --debug flag is set. Run as root to listen on privileged ports. ```python #!/usr/bin/env python3 """ Modbus/TCP server ~~~~~~~~~~~~~~~~~ Run this as root to listen on TCP privileged ports (<= 1024). Add "--host 0.0.0.0" to listen on all available IPv4 addresses of the host. $ sudo ./server.py --host 0.0.0.0 """ import argparse import logging from pyModbusTCP.server import ModbusServer # init logging logging.basicConfig() # parse args parser = argparse.ArgumentParser() parser.add_argument('-H', '--host', type=str, default='localhost', help='Host (default: localhost)') parser.add_argument('-p', '--port', type=int, default=502, help='TCP port (default: 502)') parser.add_argument('-d', '--debug', action='store_true', help='set debug mode') args = parser.parse_args() # logging setup if args.debug: logging.getLogger('pyModbusTCP.server').setLevel(logging.DEBUG) # start modbus server server = ModbusServer(host=args.host, port=args.port) server.start() ``` -------------------------------- ### Install Specific pyModbusTCP Version Source: https://github.com/sourceperl/pymodbustcp/blob/master/README.rst Installs a specific version of the pyModbusTCP library using pip. This is recommended for project stability. ```bash sudo pip install pyModbusTCP==v0.1.10 ``` -------------------------------- ### Run PyModbus TCP Server with Serial Worker Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server_serial_gw.md Use this snippet to start a Modbus TCP server that communicates via a serial port. Ensure the necessary libraries are installed and serial port parameters are correctly configured. ```python args = parser.parse_args() # init logging logging.basicConfig(level=logging.DEBUG if args.debug else None) logger = logging.getLogger(__name__) try: # init serial port logger.debug('Open serial port %s at %d bauds', args.device, args.baudrate) serial_port = Serial(port=args.device, baudrate=args.baudrate) # init serial worker serial_worker = ModbusSerialWorker(serial_port, args.timeout, args.eof) # start modbus server with custom engine logger.debug('Start modbus server (%s, %d)', args.host, args.port) srv = ModbusServer(host=args.host, port=args.port, no_block=True, ext_engine=serial_worker.srv_engine_entry) srv.start() # start serial worker loop logger.debug('Start serial worker') serial_worker.loop() except serialutil.SerialException as e: logger.critical('Serial device error: %r', e) exit(1) except ModbusServer.Error as e: logger.critical('Modbus server error: %r', e) exit(2) ``` -------------------------------- ### Modbus Polling Thread Example Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/client_thread.md This script starts a Modbus polling thread that continuously reads holding registers from a server and displays the results. Ensure the Modbus server is running on localhost:502. Exit the script using Ctrl+C. ```python #!/usr/bin/env python3 """ modbus polling thread ~~~~~~~~~~~~~~~~~~~~~~ Start a thread for polling a set of registers, display result on console. Exit with ctrl+c. """ import time from threading import Lock, Thread from pyModbusTCP.client import ModbusClient SERVER_HOST = "localhost" SERVER_PORT = 502 # set global regs = [] # init a thread lock regs_lock = Lock() def polling_thread(): """Modbus polling thread.""" global regs, regs_lock c = ModbusClient(host=SERVER_HOST, port=SERVER_PORT, auto_open=True) # polling loop while True: # do modbus reading on socket reg_list = c.read_holding_registers(0, 10) # if read is ok, store result in regs (with thread lock) if reg_list: with regs_lock: regs = list(reg_list) # 1s before next polling time.sleep(1) # start polling thread tp = Thread(target=polling_thread) # set daemon: polling thread will exit if main thread exit tp.daemon = True tp.start() # display loop (in main thread) while True: # print regs list (with thread lock synchronization) with regs_lock: print(regs) # 1s before next print time.sleep(1) ``` -------------------------------- ### Enable Developer Mode in PyModbusTCP Source: https://github.com/sourceperl/pymodbustcp/blob/master/HOWTO-pkg-devel.md Activates developer mode by installing the current package in editable mode within a virtual environment. This adds the package files to the Python path, enabling direct testing of modifications. ```bash python -m venv venv && source venv/bin/activate pip install --editable . ``` -------------------------------- ### Display Debug Frames Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/quickstart/index.md Example of debug output showing transmitted (Tx) and received (Rx) Modbus frames. ```text DEBUG:pyModbusTCP.client:(localhost:502:1) Tx [8F 8A 00 00 00 06 01] 01 00 00 00 01 DEBUG:pyModbusTCP.client:(localhost:502:1) Rx [8F 8A 00 00 00 04 01] 01 01 00 ``` -------------------------------- ### Modbus Server with Schedule Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server_schedule.md This script implements a Modbus/TCP server that can be scheduled to start and stop at specific times. It also updates a holding register every 10 seconds with the current time. ```python #!/usr/bin/env python3 """ Modbus/TCP server with start/stop schedule ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Run this as root to listen on TCP privileged ports (<= 1024). Default Modbus/TCP port is 502, so we prefix call with sudo. With argument "--host 0.0.0.0", server listen on all IPv4 of the host. Instead of just open tcp/502 on local interface. $ sudo ./server_schedule.py --host 0.0.0.0 """ import argparse import time # need https://github.com/dbader/schedule import schedule from pyModbusTCP.server import ModbusServer def alive_word_job(): """Update holding register @0 with day second (since 00:00). Job called every 10s by scheduler. """ server.data_bank.set_holding_registers(0, [int(time.time()) % (24*3600) // 10]) # parse args parser = argparse.ArgumentParser() parser.add_argument('-H', '--host', type=str, default='localhost', help='Host (default: localhost)') parser.add_argument('-p', '--port', type=int, default=502, help='TCP port (default: 502)') args = parser.parse_args() # init modbus server and start it server = ModbusServer(host=args.host, port=args.port, no_block=True) server.start() # init scheduler # schedule a daily downtime (from 18:00 to 06:00) schedule.every().day.at('18:00').do(server.stop) schedule.every().day.at('06:00').do(server.start) # update life word at @0 schedule.every(10).seconds.do(alive_word_job) # main loop while True: schedule.run_pending() time.sleep(1) ``` -------------------------------- ### Read Holding Registers with PyModbusTCP Client Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/client_read_h_registers.md This script continuously reads 10 holding registers starting from address 0 every 2 seconds. Ensure the ModbusClient is initialized with auto_open=True for automatic connection. ```python #!/usr/bin/env python3 """ Read 10 holding registers and print result on stdout. """ import time from pyModbusTCP.client import ModbusClient # init modbus client c = ModbusClient(auto_open=True) # main read loop while True: # read 10 registers at address 0, store result in regs list regs_l = c.read_holding_registers(0, 10) # if success display registers if regs_l: print('reg ad #0 to 9: %s' % regs_l) else: print('unable to read registers') # sleep 2s before next polling time.sleep(2) ``` -------------------------------- ### Write Multiple Coils (Function 0x0F) Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Writes a list of boolean values to multiple Modbus coils starting from a specified address. This is more efficient than writing coils individually. Ensure the list of values matches the number of coils to be written. ```python from pyModbusTCP.client import ModbusClient client = ModbusClient(host='localhost', port=502, auto_open=True) # Write multiple coils starting at address 0 coil_values = [True, False, True, True, False, False, True, False] if client.write_multiple_coils(0, coil_values): print(f"Written {len(coil_values)} coils starting at address 0") else: print(f"Write error: {client.last_error_as_txt}") # Set output pattern for 8 digital outputs output_pattern = [True, True, False, False, True, True, False, False] client.write_multiple_coils(0, output_pattern) ``` -------------------------------- ### Read Coils from Modbus TCP Server Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/client_read_coils.md This script connects to a Modbus TCP server, reads 10 coils starting from address 0, and prints their status every 2 seconds. Ensure the Modbus server is running and accessible at 'localhost:502'. ```python #!/usr/bin/env python3 """ Read 10 coils and print result on stdout. """ import time from pyModbusTCP.client import ModbusClient # init modbus client c = ModbusClient(host='localhost', port=502, auto_open=True) # main read loop while True: # read 10 bits (= coils) at address 0, store result in coils list coils_l = c.read_coils(0, 10) # if success display registers if coils_l: print('coil ad #0 to 9: %s' % coils_l) else: print('unable to read coils') # sleep 2s before next polling time.sleep(2) ``` -------------------------------- ### Write Multiple Registers (Function 0x10) Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Writes a list of 16-bit integer values to multiple Modbus holding registers starting from a specified address. This method is efficient for updating multiple register values at once. The number of values in the list determines how many registers are written. ```python from pyModbusTCP.client import ModbusClient client = ModbusClient(host='localhost', port=502, auto_open=True) # Write multiple registers starting at address 10 register_values = [44, 55, 66, 77] if client.write_multiple_registers(10, register_values): print(f"Written values {register_values} to registers 10-13") else: print(f"Write error: {client.last_error_as_txt}") # Write PID parameters (Kp, Ki, Kd scaled by 100) pid_params = [250, 50, 10] # Kp=2.50, Ki=0.50, Kd=0.10 if client.write_multiple_registers(200, pid_params): print("PID parameters updated") ``` -------------------------------- ### Build Source and Wheel Archives Source: https://github.com/sourceperl/pymodbustcp/blob/master/HOWTO-PyPi.md Run this command in your project directory to create distribution archives. ```bash python setup.py sdist bdist_wheel ``` -------------------------------- ### Configure Twine .pypirc File Source: https://github.com/sourceperl/pymodbustcp/blob/master/HOWTO-PyPi.md Create or append to the ~/.pypirc file to configure credentials for PyPI and the test PyPI server. Replace '__token__' and 'mytoken' with your actual credentials. ```bash cat <> ~/.pypirc [distutils] index-servers = pypi pypitest [pypi] repository: https://upload.pypi.org/legacy/ username: __token__ password: mytoken [pypitest] repository: https://test.pypi.org/legacy/ username: __token__ password: mytoken EOT ``` -------------------------------- ### Initialize ModbusClient via Properties Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/quickstart/index.md Initialize the ModbusClient without parameters and set the host and port using properties. ```python from pyModbusTCP.client import ModbusClient c = ModbusClient() c.host = 'localhost' c.port = 502 ``` -------------------------------- ### Upload to PyPI Production Server Source: https://github.com/sourceperl/pymodbustcp/blob/master/HOWTO-PyPi.md Execute this command to upload your distribution archives to the main PyPI server. Verify your configuration in .pypirc before running. ```bash twine upload dist/pyModbusTCP-x.x.x* -r pypi ``` -------------------------------- ### Read Holding Registers Source: https://github.com/sourceperl/pymodbustcp/blob/master/README.rst Reads two 16-bit registers starting from Modbus address 0. Checks if the read operation was successful before printing. ```python regs = c.read_holding_registers(0, 2) if regs: print(regs) else: print("read error") ``` -------------------------------- ### ModbusClient - Basic Connection Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Demonstrates various ways to instantiate and manage a ModbusClient connection, including auto-connect, manual management, and minimal usage. ```python from pyModbusTCP.client import ModbusClient # Create client with auto-open enabled (default behavior since v0.2.0) # Connection opens automatically on first request client = ModbusClient(host='192.168.1.100', port=502, unit_id=1, auto_open=True) # For one-request-per-connection pattern (auto close after each request) client = ModbusClient(host='192.168.1.100', port=502, auto_open=True, auto_close=True) # Manual connection management client = ModbusClient(host='192.168.1.100', port=502, auto_open=False) if client.open(): print("Connected successfully") # ... perform operations ... client.close() else: print(f"Connection failed: {client.last_error_as_txt}") # Check connection status print(f"Is connected: {client.is_open}") # Minimal one-liner usage (localhost:502 defaults) coils = ModbusClient().read_coils(0, 3) print(f"Coils: {coils}") ``` -------------------------------- ### Write Multiple Registers Source: https://github.com/sourceperl/pymodbustcp/blob/master/README.rst Writes the values 44 and 55 to Modbus registers starting at address 10. Reports success or failure of the write operation. ```python if c.write_multiple_registers(10, [44,55]): print("write ok") else: print("write error") ``` -------------------------------- ### Initialize ModbusClient via Constructor Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/quickstart/index.md Initialize the ModbusClient by providing the host and port in the constructor. Catches ValueError for invalid host/port parameters. ```python from pyModbusTCP.client import ModbusClient try: c = ModbusClient(host='localhost', port=502) except ValueError: print("Error with host or port params") ``` -------------------------------- ### ModbusClient - Read Coils Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Reads multiple coil values (boolean outputs) from the Modbus server. Handles potential read errors and Modbus exceptions. Includes an example of continuous polling. ```python from pyModbusTCP.client import ModbusClient import time client = ModbusClient(host='localhost', port=502, auto_open=True) # Read 10 coils starting at address 0 coils = client.read_coils(0, 10) if coils: print(f"Coils 0-9: {coils}") # Output: Coils 0-9: [False, True, False, False, True, False, False, False, False, False] # Access individual coil values for i, value in enumerate(coils): print(f"Coil {i}: {'ON' if value else 'OFF'}") else: print(f"Read error: {client.last_error_as_txt}") if client.last_except: print(f"Modbus exception: {client.last_except_as_txt}") # Continuous polling loop while True: coils = client.read_coils(0, 10) if coils: print(f"Coil status: {coils}") else: print("Unable to read coils") time.sleep(2) ``` -------------------------------- ### Custom DataBank with Change Logging Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Implements a custom DataBank that logs changes to coils and holding registers. Requires setting up logging and instantiating the server with the custom DataBank. ```python import logging from pyModbusTCP.server import DataBank, ModbusServer class LoggingDataBank(DataBank): """A custom DataBank that logs all changes.""" def on_coils_change(self, address, from_value, to_value, srv_info): """Called when a coil value changes.""" msg = f"Coil change [{from_value!r:^5} > {to_value!r:^5}] at 0x{address:04X}" msg += f" from {srv_info.client.address}:{srv_info.client.port}" logging.info(msg) def on_holding_registers_change(self, address, from_value, to_value, srv_info): """Called when a holding register value changes.""" msg = f"Register change [{from_value:5d} > {to_value:5d}] at 0x{address:04X}" msg += f" from {srv_info.client.address}:{srv_info.client.port}" logging.info(msg) # Setup logging logging.basicConfig(format='%(asctime)s %(message)s', level=logging.INFO) # Create server with custom data bank server = ModbusServer(host='localhost', port=502, data_bank=LoggingDataBank()) server.start() # Output when client writes: # 2024-01-15 10:30:45 Coil change [False > True ] at 0x0000 from 127.0.0.1:54321 # 2024-01-15 10:30:46 Register change [ 0 > 100] at 0x000A from 127.0.0.1:54321 ``` -------------------------------- ### Upload to PyPI Test Server Source: https://github.com/sourceperl/pymodbustcp/blob/master/HOWTO-PyPi.md Use this command to upload your distribution archives to the PyPI test server for verification. Ensure you have configured your .pypirc file with test credentials. ```bash twine upload dist/pyModbusTCP-x.x.x* -r pypitest ``` -------------------------------- ### Basic Modbus Server Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Sets up a basic Modbus/TCP server with default data banks. Can be run in blocking or non-blocking mode. Initial values for registers and coils can be set. ```python import argparse import logging from pyModbusTCP.server import ModbusServer # Enable debug logging logging.basicConfig() logging.getLogger('pyModbusTCP.server').setLevel(logging.DEBUG) # Create and start server (blocking mode) server = ModbusServer(host='localhost', port=502) print("Starting Modbus server...") server.start() # Blocks until Ctrl+C # For non-blocking mode (server runs in background thread) server = ModbusServer(host='0.0.0.0', port=502, no_block=True) server.start() print(f"Server running: {server.is_run}") # Access the data bank to set initial values server.data_bank.set_holding_registers(0, [100, 200, 300, 400]) server.data_bank.set_coils(0, [True, False, True, False]) # Stop server # server.stop() ``` -------------------------------- ### Edit .pypirc File Source: https://github.com/sourceperl/pymodbustcp/blob/master/HOWTO-PyPi.md Open the ~/.pypirc file using the nano text editor to manually update your credentials or configuration. ```bash nano ~/.pypirc ``` -------------------------------- ### Configure Modbus Server Device Identification Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Set up vendor name, product code, revision, URLs, and application names for Modbus server device identification. Custom identification objects can also be added. ```python from pyModbusTCP.server import DeviceIdentification, ModbusServer # Create device identification object device_id = DeviceIdentification( vendor_name=b'ACME Industries', product_code=b'CTRL-500', major_minor_revision=b'2.1.0', vendor_url=b'https://www.example.com', product_name=b'Industrial Controller', model_name=b'CTRL-500-PRO', user_application_name=b'Temperature Monitor' ) # Add custom object IDs (0x80-0xFF for extended identification) device_id[0x80] = b'Serial: ABC123456' device_id[0x81] = b'Location: Building A' # Create server with device identification server = ModbusServer( host='localhost', port=502, device_id=device_id ) server.start() ``` -------------------------------- ### Initialize ModbusClient (TCP Open/Close per Request) Source: https://github.com/sourceperl/pymodbustcp/blob/master/README.rst Initializes a ModbusClient instance with TCP auto-connect and auto-close enabled. The connection opens for each request and closes afterward. ```python c = ModbusClient(host="127.0.0.1", auto_open=True, auto_close=True) ``` -------------------------------- ### Command Line Interface Arguments Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server_serial_gw.md Parses command-line arguments for configuring the Modbus serial worker. ```APIDOC ## Command Line Interface ### Description This section details the command-line arguments used to configure and run the Modbus serial worker. ### Arguments - **`device`** (type: str): The serial device path (e.g., `/dev/ttyUSB0`). **Required**. - **`-H`, `--host`** (type: str): The host address for network connections (default: `localhost`). - **`-p`, `--port`** (type: int): The TCP port for network connections (default: 502). - **`-b`, `--baudrate`** (type: int): The serial communication baud rate (default: 9600). - **`-t`, `--timeout`** (type: float): The timeout for serial read operations in seconds (default: 1.0). - **`-e`, `--eof`** (type: float): The end-of-frame delay for serial reads in seconds (default: 0.05). - **`-d`, `--debug`**: Enables debug mode. ``` -------------------------------- ### Write and Read Coils with PyModbusTCP Client Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/client_write_coils.md This script demonstrates writing single coils to a Modbus server and then reading them back. It includes error handling for write operations and a loop for continuous polling. Ensure the Modbus server is running on localhost:502. ```python #!/usr/bin/env python3 """Write 4 coils to True, wait 2s, write False and redo it.""" import time from pyModbusTCP.client import ModbusClient # init c = ModbusClient(host='localhost', port=502, auto_open=True) bit = True # main loop while True: # write 4 bits in modbus address 0 to 3 print('write bits') print('----------\n') for ad in range(4): is_ok = c.write_single_coil(ad, bit) if is_ok: print('coil #%s: write to %s' % (ad, bit)) else: print('coil #%s: unable to write %s' % (ad, bit)) time.sleep(0.5) print('') time.sleep(1) # read 4 bits in modbus address 0 to 3 print('read bits') print('---------\n') bits = c.read_coils(0, 4) if bits: print('coils #0 to 3: %s' % bits) else: print('coils #0 to 3: unable to read') # toggle bit = not bit # sleep 2s before next polling print('') time.sleep(2) ``` -------------------------------- ### Command-line Argument Parsing Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server_serial_gw.md Parses command-line arguments for configuring serial device, host, port, baudrate, timeouts, and debug mode. ```python if __name__ == '__main__': # parse args parser = argparse.ArgumentParser() parser.add_argument('device', type=str, help='serial device (like /dev/ttyUSB0)') parser.add_argument('-H', '--host', type=str, default='localhost', help='host (default: localhost)') parser.add_argument('-p', '--port', type=int, default=502, help='TCP port (default: 502)') parser.add_argument('-b', '--baudrate', type=int, default=9600, help='serial rate (default is 9600)') parser.add_argument('-t', '--timeout', type=float, default=1.0, help='timeout delay (default is 1.0 s)') parser.add_argument('-e', '--eof', type=float, default=0.05, help='end of frame delay (default is 0.05 s)') parser.add_argument('-d', '--debug', action='store_true', help='set debug mode') ``` -------------------------------- ### Run Modbus/TCP Serial Gateway Server Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server_serial_gw.md Execute this script with root privileges to listen on privileged TCP ports. It configures a serial port (e.g., /dev/ttyUSB0) with specified baudrate and acts as a gateway for RTU devices. ```python #!/usr/bin/env python3 """ Modbus/TCP basic gateway (RTU slave(s) attached) ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ [pyModbusTCP server] -> [ModbusSerialWorker] -> [serial RTU devices] Run this as root to listen on TCP privileged ports (<= 1024). Open /dev/ttyUSB0 at 115200 bauds and relay it RTU messages to slave(s). $ sudo ./server_serial_gw.py --baudrate 115200 /dev/ttyUSB0 """ import argparse import logging import queue import struct from queue import Queue from threading import Event # need sudo pip install pyserial==3.4 from serial import Serial, serialutil from pyModbusTCP.constants import ( EXP_GATEWAY_PATH_UNAVAILABLE, EXP_GATEWAY_TARGET_DEVICE_FAILED_TO_RESPOND ) from pyModbusTCP.server import ModbusServer from pyModbusTCP.utils import crc16 ``` -------------------------------- ### Import ModbusClient Source: https://github.com/sourceperl/pymodbustcp/blob/master/README.rst Imports the ModbusClient class from the pyModbusTCP library. This is a prerequisite for all client operations. ```python from pyModbusTCP.client import ModbusClient ``` -------------------------------- ### Test, Set, Reset, and Toggle Bits in an Integer Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Utility functions for performing bitwise operations on integer values. Use `test_bit` to check a bit's state, `set_bit` to turn it on, `reset_bit` to turn it off, and `toggle_bit` to flip its state. ```python from pyModbusTCP.utils import ( test_bit, set_bit, reset_bit, toggle_bit, get_bits_from_int) # Test if a specific bit is set value = 0b10101010 # 170 decimal print(f"Bit 1 is set: {test_bit(value, 1)}") # True print(f"Bit 0 is set: {test_bit(value, 0)}") # False # Set a bit value = 0b00000000 value = set_bit(value, 3) print(f"After setting bit 3: {bin(value)}") # 0b1000 # Reset (clear) a bit value = 0b11111111 value = reset_bit(value, 4) print(f"After resetting bit 4: {bin(value)}") # 0b11101111 # Toggle a bit value = 0b00001111 value = toggle_bit(value, 2) print(f"After toggling bit 2: {bin(value)}") # 0b00001011 # Get all bits from an integer as a list (LSB first) bits = get_bits_from_int(0xA5, val_size=8) print(f"Bits of 0xA5: {bits}") # Output: [True, False, True, False, False, True, False, True] # Alias: int2bits from pyModbusTCP.utils import int2bits bits = int2bits(0xFFFF, 16) ``` -------------------------------- ### Initialize ModbusClient (TCP Always Open) Source: https://github.com/sourceperl/pymodbustcp/blob/master/README.rst Initializes a ModbusClient instance with TCP auto-connect enabled on the first Modbus request. The connection remains open. ```python c = ModbusClient(host="localhost", port=502, unit_id=1, auto_open=True) ``` -------------------------------- ### View Current Python Path Source: https://github.com/sourceperl/pymodbustcp/blob/master/HOWTO-pkg-devel.md Prints the current Python sys.path, which includes directories where Python looks for modules. Useful for verifying that the editable package has been correctly added. ```python python -c 'import sys; print(sys.path)' ``` -------------------------------- ### Virtual Data Mode Server Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Creates a Modbus server where holding registers provide virtual, computed values based on the current date and time. Requires enabling virtual mode in the DataBank constructor. ```python from datetime import datetime from pyModbusTCP.server import DataBank, ModbusServer class VirtualDataBank(DataBank): """DataBank with virtual holding registers showing current date/time.""" def __init__(self): # Enable virtual mode (no memory allocation for standard spaces) super().__init__(virtual_mode=True) def get_holding_registers(self, address, number=1, srv_info=None): """Return virtual register values based on current datetime.""" now = datetime.now() virtual_regs = { 0: now.day, 1: now.month, 2: now.year, 3: now.hour, 4: now.minute, 5: now.second } # Build response list try: return [virtual_regs[a] for a in range(address, address + number)] except KeyError: return None # Address out of range # Create server with virtual data bank server = ModbusServer(host='localhost', port=502, data_bank=VirtualDataBank()) server.start() # Client reads will get current date/time values: # Register 0: Day (1-31) # Register 1: Month (1-12) ``` -------------------------------- ### Modbus Server with Virtual Date/Time Data Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server_virtual_data.md This Python script implements a Modbus/TCP server that maps the current system date and time to holding registers 0-5. It overrides the default DataBank to provide virtual data, allowing only read access to these registers. Run as root to listen on ports below 1024. ```python #!/usr/bin/env python3 """ Modbus/TCP server with virtual data ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Map the system date and time to @ 0 to 5 on the "holding registers" space. Only the reading of these registers in this address space is authorized. All other requests return an illegal data address except. Run this as root to listen on TCP priviliged ports (<= 1024). """ import argparse from datetime import datetime from pyModbusTCP.server import DataBank, ModbusServer class MyDataBank(DataBank): """A custom ModbusServerDataBank for override get_holding_registers method.""" def __init__(self): # turn off allocation of memory for standard modbus object types # only "holding registers" space will be replaced by dynamic build values. super().__init__(virtual_mode=True) def get_holding_registers(self, address, number=1, srv_info=None): """Get virtual holding registers.""" # populate virtual registers dict with current datetime values now = datetime.now() v_regs_d = {0: now.day, 1: now.month, 2: now.year, 3: now.hour, 4: now.minute, 5: now.second} # build a list of virtual regs to return to server data handler # return None if any of virtual registers is missing try: return [v_regs_d[a] for a in range(address, address+number)] except KeyError: return if __name__ == '__main__': # parse args parser = argparse.ArgumentParser() parser.add_argument('-H', '--host', type=str, default='localhost', help='Host (default: localhost)') parser.add_argument('-p', '--port', type=int, default=502, help='TCP port (default: 502)') args = parser.parse_args() # init modbus server and start it server = ModbusServer(host=args.host, port=args.port, data_bank=MyDataBank()) server.start() ``` -------------------------------- ### Modbus Server with Change Logger Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server_change_log.md This Python script implements a Modbus/TCP server that logs changes to coils and holding registers. It requires root privileges to listen on ports below 1024. The server uses a custom DataBank to intercept and log these changes. ```python #!/usr/bin/env python3 """ An example of Modbus/TCP server with a change logger. Run this as root to listen on TCP privileged ports (<= 1024). """ import argparse import logging from pyModbusTCP.server import DataBank, ModbusServer class MyDataBank(DataBank): """A custom ModbusServerDataBank for override on_xxx_change methods.""" def on_coils_change(self, address, from_value, to_value, srv_info): """Call by server when change occur on coils space.""" msg = 'change in coil space [{0!r:^5} > {1!r:^5}] at @ 0x{2:04X} from ip: {3:<15}' msg = msg.format(from_value, to_value, address, srv_info.client.address) logging.info(msg) def on_holding_registers_change(self, address, from_value, to_value, srv_info): """Call by server when change occur on holding registers space.""" msg = 'change in hreg space [{0!r:^5} > {1!r:^5}] at @ 0x{2:04X} from ip: {3:<15}' msg = msg.format(from_value, to_value, address, srv_info.client.address) logging.info(msg) if __name__ == '__main__': # parse args parser = argparse.ArgumentParser() parser.add_argument('-H', '--host', type=str, default='localhost', help='Host (default: localhost)') parser.add_argument('-p', '--port', type=int, default=502, help='TCP port (default: 502)') args = parser.parse_args() # logging setup logging.basicConfig(format='%(asctime)s %(message)s', level=logging.INFO) # init modbus server and start it server = ModbusServer(host=args.host, port=args.port, data_bank=MyDataBank()) server.start() ``` -------------------------------- ### Connection Verification Before Operations Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Checks if the Modbus client connection is open and attempts to open it if not. Prints an error if connection cannot be established. ```python if not client.is_open: if not client.open(): print("Cannot establish connection") ``` -------------------------------- ### Handle Signed Integers with Two's Complement Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Utility functions for converting between unsigned register values and signed integers using two's complement representation. `get_2comp` handles single values, while `get_list_2comp` processes lists. Specify `val_size` for 16 or 32-bit integers. ```python from pyModbusTCP.utils import get_2comp, get_list_2comp, twos_c, twos_c_l # Convert unsigned 16-bit register value to signed unsigned_value = 65535 # 0xFFFF in register signed_value = get_2comp(unsigned_value, val_size=16) print(f"Signed value: {signed_value}") # -1 # Convert signed to unsigned for writing signed_input = -100 unsigned_output = get_2comp(signed_input, val_size=16) print(f"Unsigned for register: {unsigned_output}") # 65436 # Handle 32-bit signed values signed_32 = get_2comp(0xFFFFFFFF, val_size=32) print(f"32-bit signed: {signed_32}") # -1 # Convert a list of values unsigned_list = [65535, 65534, 100, 0] signed_list = get_list_2comp(unsigned_list, val_size=16) print(f"Signed list: {signed_list}") # [-1, -2, 100, 0] # Short aliases signed = twos_c(65000) signed_list = twos_c_l([65535, 32768, 100]) ``` -------------------------------- ### ModbusClient with Auto Open and Auto Close Enabled Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/quickstart/index.md Configure the ModbusClient to open the TCP connection before each request and close it afterwards. ```python c = ModbusClient(host="localhost", auto_open=True, auto_close=True) ``` -------------------------------- ### Modbus Server with IP Allow List (Python) Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/examples/server_allow.md This Python script implements a Modbus/TCP server that filters client access based on predefined IP address lists for read and write operations. It uses a custom DataHandler to enforce these restrictions, returning an illegal function exception for unauthorized clients. ```python #!/usr/bin/env python3 """ An example of Modbus/TCP server which allow modbus read and/or write only from specific IPs. Run this as root to listen on TCP privileged ports (<= 1024). """ import argparse from pyModbusTCP.constants import EXP_ILLEGAL_FUNCTION from pyModbusTCP.server import DataHandler, ModbusServer # some const ALLOW_R_L = ['127.0.0.1', '192.168.0.10'] ALLOW_W_L = ['127.0.0.1'] # a custom data handler with IPs filter class MyDataHandler(DataHandler): def read_coils(self, address, count, srv_info): if srv_info.client.address in ALLOW_R_L: return super().read_coils(address, count, srv_info) else: return DataHandler.Return(exp_code=EXP_ILLEGAL_FUNCTION) def read_d_inputs(self, address, count, srv_info): if srv_info.client.address in ALLOW_R_L: return super().read_d_inputs(address, count, srv_info) else: return DataHandler.Return(exp_code=EXP_ILLEGAL_FUNCTION) def read_h_regs(self, address, count, srv_info): if srv_info.client.address in ALLOW_R_L: return super().read_h_regs(address, count, srv_info) else: return DataHandler.Return(exp_code=EXP_ILLEGAL_FUNCTION) def read_i_regs(self, address, count, srv_info): if srv_info.client.address in ALLOW_R_L: return super().read_i_regs(address, count, srv_info) else: return DataHandler.Return(exp_code=EXP_ILLEGAL_FUNCTION) def write_coils(self, address, bits_l, srv_info): if srv_info.client.address in ALLOW_W_L: return super().write_coils(address, bits_l, srv_info) else: return DataHandler.Return(exp_code=EXP_ILLEGAL_FUNCTION) def write_h_regs(self, address, words_l, srv_info): if srv_info.client.address in ALLOW_W_L: return super().write_h_regs(address, words_l, srv_info) else: return DataHandler.Return(exp_code=EXP_ILLEGAL_FUNCTION) if __name__ == '__main__': # parse args parser = argparse.ArgumentParser() parser.add_argument('-H', '--host', type=str, default='localhost', help='Host (default: localhost)') parser.add_argument('-p', '--port', type=int, default=502, help='TCP port (default: 502)') args = parser.parse_args() # init modbus server and start it server = ModbusServer(host=args.host, port=args.port, data_hdl=MyDataHandler()) server.start() ``` -------------------------------- ### ModbusClient with Float Support Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Extends ModbusClient to handle IEEE 754 floating-point values. Requires importing utility functions for encoding and decoding. ```python from pyModbusTCP.client import ModbusClient from pyModbusTCP.utils import (decode_ieee, encode_ieee, long_list_to_word, word_list_to_long) class FloatModbusClient(ModbusClient): """A ModbusClient class with float support.""" def read_float(self, address, number=1): """Read float(s) from holding registers (2 registers per float).""" reg_list = self.read_holding_registers(address, number * 2) if reg_list: return [decode_ieee(f) for f in word_list_to_long(reg_list)] return None def write_float(self, address, floats_list): """Write float(s) to holding registers.""" b32_list = [encode_ieee(f) for f in floats_list] b16_list = long_list_to_word(b32_list) return self.write_multiple_registers(address, b16_list) # Usage client = FloatModbusClient(host='localhost', port=502, auto_open=True) # Write floats client.write_float(0, [3.14159, 2.71828, 1.41421]) # Read floats (10 floats = 20 registers) float_values = client.read_float(0, 10) if float_values: print(f"Float values: {float_values}") # Output: Float values: [3.14159, 2.71828, 1.41421, 0.0, ...] client.close() ``` -------------------------------- ### Enable Debug Logging for ModbusClient Source: https://github.com/sourceperl/pymodbustcp/blob/master/docs/quickstart/index.md Enable debug logging for pyModbusTCP.client to observe frame exchanges. Set the logging level for 'pyModbusTCP.client' to DEBUG. ```python import logging from pyModbusTCP.client import ModbusClient # set debug level for pyModbusTCP.client to see frame exchanges logging.basicConfig() logging.getLogger('pyModbusTCP.client').setLevel(logging.DEBUG) c = ModbusClient(host="localhost", port=502) ``` -------------------------------- ### Write Multiple Registers (Function 0x10) Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Write multiple 16-bit holding registers in a single request. ```APIDOC ## Write Multiple Registers (Function 0x10) ### Description Write multiple 16-bit holding registers in a single request. ### Method POST (or equivalent Modbus function) ### Endpoint Not applicable (direct client method call) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from pyModbusTCP.client import ModbusClient client = ModbusClient(host='localhost', port=502, auto_open=True) # Write multiple registers starting at address 10 register_values = [44, 55, 66, 77] if client.write_multiple_registers(10, register_values): print(f"Written values {register_values} to registers 10-13") else: print(f"Write error: {client.last_error_as_txt}") ``` ### Response #### Success Response (200) Boolean indicating success or failure of the write operation. #### Response Example ``` True ``` ``` -------------------------------- ### Write Multiple Coils (Function 0x0F) Source: https://context7.com/sourceperl/pymodbustcp/llms.txt Write multiple coil values in a single request, more efficient than individual writes. ```APIDOC ## Write Multiple Coils (Function 0x0F) ### Description Write multiple coil values in a single request, more efficient than individual writes. ### Method POST (or equivalent Modbus function) ### Endpoint Not applicable (direct client method call) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from pyModbusTCP.client import ModbusClient client = ModbusClient(host='localhost', port=502, auto_open=True) # Write multiple coils starting at address 0 coil_values = [True, False, True, True, False, False, True, False] if client.write_multiple_coils(0, coil_values): print(f"Written {len(coil_values)} coils starting at address 0") else: print(f"Write error: {client.last_error_as_txt}") ``` ### Response #### Success Response (200) Boolean indicating success or failure of the write operation. #### Response Example ``` True ``` ```