DEV Community: Rebecca Anderson

How to Handle Modbus Bus Collisions When Multiple Python Scripts Poll the Same RS485 Sensor

Rebecca Anderson — Sat, 06 Jun 2026 14:02:32 +0000

This is the normal IoT developer experience: You write a Python script using pymodbus to read data from an RS485 soil sensor or a solar inverter. It runs perfect on your test bench for hours,

But then, the project scales up. You spin up a second Python script on the same machine to handle real-time MQTT cloud synchronization. Or perhaps the factory's main SCADA system suddenly needs to poll that exact same hardware sensor at the same time.

Boom. Your terminal fills up with ModbusIOException, timeout errors, or complete serial port lockups.

If you are currently trying to fix this by writing complex software semaphores or cross-process locks in Python, stop. Let’s look at why this happens at the hardware level and how to solve it properly through architecture.

The Root Cause: RS485 is Strictly Single-Master

The fundamental limitation isn't your Python code; it's the physics of the RS485 bus.

RS485 is a half-duplex serial interface. This means data can travel in both directions, but only one device can talk on the wire at any given microsecond. The Modbus RTU protocol is strictly a Master-Slave architecture. The Master sends a query, the Slave responds.

When you launch two independent Python scripts executing parallel polling loops, they have no awareness of each other's timing. Eventually, a race condition occurs: both scripts attempt to write to the serial port simultaneously.

The electrical signals collide on the copper wire, resulting in garbage bytes, broken frames, and catastrophic CRC failures.

[Script A] ---- Query (01 03 00 00 ...) ---->  ⚡ COLLISION ⚡
[Script B] ---- Query (01 03 00 0A ...) ---->  [ RS485 Bus Blown ]

Why Software Locks SUCK Here

Your first instinct might be to implement a file lock (fcntl), a Redis-based distributed lock, or a Python threading.Lock() to serialize the access to the serial port.

While this prevents data collisions, it introduces massive software engineering overhead:

Latency Spikes: If Script A is stuck waiting for a slow sensor response, Script B is completely blocked, leading to data gaps in your cloud database.
Deadlocks: If one script crashes while holding the serial lock, the entire data acquisition system freezes until a hard reboot.

The Architectural Fix: Decoupling via Cache

The elegant way to solve multi-host collision is to completely offload the serial polling task from your Python applications. Instead of letting your scripts talk directly to the raw serial line, you place a hardware protocol gateway in between that supports autonomous register caching.

In this architecture, the gateway is the only Master on the RS485 physical bus. It constantly polls the hardware sensors at the physical layer, and caches the latest register data directly in its own high speed internal RAM cache.

Your Python scripts can no longer connect to serial ports. Instead, they make normal, simultaneous Modbus TCP queries over Ethernet (Port 502).

[Python Script A (TCP)] --\
                           +---> [ Caching Gateway ] ---> (Stable RS485 Polling) ---> [Sensors]
[Python Script B (TCP)] --/

The Code: Clean Concurrent Modbus TCP Polling

When you shift to a caching architecture, your Python code becomes remarkably clean. Because Modbus TCP inherently handles multiple network sockets, you can run as many parallel scripts, containers, or threads as you want without a single lock.

Here is a clean implementation simulating two independent applications querying the network simultaneously:

from pymodbus.client import ModbusTcpClient
import threading
import time

# Point this to your hardware caching gateway IP
GATEWAY_IP = '192.168.1.200'
GATEWAY_PORT = 502

def application_task(client_id, register_address):
    # Each independent script or thread opens its own clean TCP socket
    client = ModbusTcpClient(GATEWAY_IP, port=GATEWAY_PORT, timeout=2)

    if not client.connect():
        print(f"App {client_id}: Network unreachable.")
        return

    print(f"App {client_id}: Connected to gateway port 502.")

    while True:
        try:
            # Query the cached hardware registers instantly (usually under 3ms)
            response = client.read_holding_registers(address=register_address, count=2, slave=1)

            if not response.isError():
                print(f"App {client_id} Data Received: {response.registers}")
            else:
                print(f"App {client_id} Modbus Error: {response}")

        except Exception as e:
            print(f"App {client_id} Network Exception: {e}")

        # Poll as fast as you want; no serial collisions possible
        time.sleep(1) 

if __name__ == '__main__':
    # Simulating two completely separate applications running in parallel
    app_cloud_sync = threading.Thread(target=application_task, args=(1, 0))
    app_local_scada = threading.Thread(target=application_task, args=(2, 10))

    app_cloud_sync.start()
    app_local_scada.start()

Hardware Layer Note

To make this architecture work seamlessly in production environments, your hardware gateway must handle the scheduling and isolation. A passive, cheap serial-to-ethernet converter that just passes raw bytes will still cause collisions if two TCP packets arrive at the same time.

In our commercial field deployments for remote solar farms, we completely shifted away from passive dongles to the industrial multi-channel storage gateways manufactured by Valtoris. Their hardware natively manages the asynchronous scheduling, features 3000V galvanic isolation to block ground loops on factory floors, and supports up to 32 concurrent Modbus TCP client connections.

By moving the bus synchronization from unstable Python scripts down to dedicated hardware, your software can focus entirely on data processing and analytics, making your entire IoT stack infinitely more resilient.

Quick Python script to dump raw RS485 hex data

Rebecca Anderson — Sun, 24 May 2026 08:56:01 +0000

Before writing a full Modbus client, I usually just need to see if the bus is actually alive and outputting data.

Here is a minimal hex dumper using pyserial. It doesn't parse anything; it just dumps raw bytes to the terminal so you can check for noise.


python
import serial
import binascii

SERIAL_PORT = '/dev/ttyUSB0'
BAUD_RATE = 9600

def hex_dump():
    ser = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=0.5)
    print(f"Listening on {SERIAL_PORT}...")

    while True:
        if ser.in_waiting > 0:
            data = ser.read(ser.in_waiting)
            formatted_hex = ' '.join(binascii.hexlify(data).decode('utf-8').upper()[i:i+2] for i in range(0, len(data)*2, 2))
            print(f"RX: {formatted_hex}")

if __name__ == '__main__':
    hex_dump()

Reading Modbus RTU Data with Python

Rebecca Anderson — Wed, 15 Apr 2026 02:12:40 +0000

Most tutorials on reading Modbus RTU data with Python focus entirely on the software side. The standard advice is to run pip install pymodbus, plug a USB-to-RS485 adapter into your machine, and write a script using ModbusSerialClient.
It usually works perfectly on your desk. But when you deploy it to a production environment, the connection randomly drops, you get timeout errors, or the data reads as garbage.

Let's look at why this happens, and how to structure your Modbus integration so it actually stays up.

The Underlying Problem: Timing and Noise
Reading serial data directly via Python in a production setting introduces two major failure points:

OS Timing Issues: Modbus RTU relies on strict timing gaps between data frames (typically 3.5 character times). Standard operating systems like Linux or Windows are not real-time OSs. If the CPU is busy and delays reading the serial buffer, the frame breaks, and your Python script throws a CRC error.
Electrical Isolation: The floors of factories are electrically loud. A normal USB-to-RS485 dongle does not provide galvanic isolation. The ground potential changes when a big machine nearby starts up. That spike goes all the way up the RS485 cable, crashes the USB driver on your server, and you have to unplug and plug it back in to fix it.

The Fix: Decoupling the Architecture
The most reliable way to read Modbus RTU in Python is to avoid dealing with the serial port entirely.

Instead of letting your server handle the physical layer, decouple it using an industrial protocol gateway. You place a Modbus RTU-to-TCP converter right next to the physical sensors.

The hardware gateway handles the strict RS485 timing, the electrical isolation, and the continuous polling. It then exposes that data over standard Ethernet as Modbus TCP.

Your Python script no longer fights with ttyUSB0 or baud rates. It simply makes a standard TCP network request.

The Code: Switching to Modbus TCP
By shifting to TCP, the Python implementation becomes much cleaner and naturally handles network resilience. Here is the baseline structure using pymodbus:

from pymodbus.client import ModbusTcpClient
import time

# Point this to the IP of your hardware gateway
GATEWAY_IP = '192.168.1.100'
GATEWAY_PORT = 502

def poll_sensor():
    # Use TCP Client instead of Serial
    client = ModbusTcpClient(GATEWAY_IP, port=GATEWAY_PORT, timeout=3)

    try:
        if not client.connect():
            print("Connection failed. Check network or gateway.")
            return

        # Read 2 registers from Slave ID 1, starting at address 0
        response = client.read_holding_registers(address=0, count=2, slave=1)

        if response.isError():
            print(f"Modbus Error: {response}")
        else:
            raw_data = response.registers
            print(f"Raw Registers: {raw_data}")

            # Basic parsing example
            temp = raw_data[0] / 10.0
            print(f"Temperature: {temp} °C")

    except Exception as e:
        print(f"Exception caught: {e}")
    finally:
        client.close()

if __name__ == '__main__':
    while True:
        poll_sensor()
        time.sleep(5)

One Last Gotcha: Endianness
If your script connects but the data doesn't make sense (for example, your temperature says 18432 instead of 24.5), you haven't done anything wrong. There is a problem with the byte order.

The Modbus protocol specification does not strictly define how 32-bit floats or long integers should be ordered across 16-bit registers. Different sensor manufacturers use different byte orders (Big-Endian vs. Little-Endian).

If you come across this, don't try to figure out the bits by hand. To change the order of the bytes until the output matches what you expect, use the BinaryPayloadDecoder module that comes with pymodbus.

from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian

# Example: Decoding a 32-bit float
decoder = BinaryPayloadDecoder.fromRegisters(raw_data, byteorder=Endian.Big, wordorder=Endian.Little)
real_value = decoder.decode_32bit_float()