One Laptop, Two Ports, Real Wires: A USB-RS232-RS485 Loopback for Modbus Testing

#!/usr/bin/env python3
"""Modbus RTU loopback test between two local serial ports.

Runs a Modbus RTU slave on one port (background thread) and a master on the
other, exercises the common function codes and verifies the results.
By default the test runs twice: RS232 side as master, then RS485 side as master.

Bench wiring:
    COM5 = RS232  (Benfei USB-RS232 -> Dtech RS232->RS485 converter)
    COM6 = RS485  (DTech 5019 USB-RS485, CP2102N)

Requires only pyserial:  pip install pyserial

Examples:
    python modbus_test.py
    python modbus_test.py --baud 19200 --parity E --cycles 500
    python modbus_test.py --direction 232-master
"""
import argparse
import random
import struct
import sys
import threading
import time

import serial


# ---------------------------------------------------------------- framing

def crc16(data: bytes) -> int:
    crc = 0xFFFF
    for b in data:
        crc ^= b
        for _ in range(8):
            crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
    return crc


def add_crc(frame: bytes) -> bytes:
    return frame + struct.pack("<H", crc16(frame))


def crc_ok(frame: bytes) -> bool:
    return len(frame) >= 4 and crc16(frame[:-2]) == struct.unpack("<H", frame[-2:])[0]


def read_exact(ser, n, deadline):
    buf = bytearray()
    while len(buf) < n and time.perf_counter() < deadline:
        buf += ser.read(n - len(buf))
    return bytes(buf)


def read_until_idle(ser, deadline):
    buf = bytearray()
    while time.perf_counter() < deadline:
        chunk = ser.read(256)
        if not chunk:
            break
        buf += chunk
    return bytes(buf)


def open_port(name, baud, parity):
    ser = serial.Serial(name, baud, bytesize=8, parity=parity, stopbits=1, timeout=0.05)
    ser.dtr = True  # DTR/RTS may power a port-powered RS232->RS485 converter
    ser.rts = True
    time.sleep(0.2)
    ser.reset_input_buffer()
    return ser


class ModbusException(Exception):
    NAMES = {1: "ILLEGAL FUNCTION", 2: "ILLEGAL DATA ADDRESS", 3: "ILLEGAL DATA VALUE"}

    def __init__(self, code):
        super().__init__(f"exception {code} ({self.NAMES.get(code, '?')})")
        self.code = code


# ---------------------------------------------------------------- slave

class Slave(threading.Thread):
    def __init__(self, ser, unit, size=200):
        super().__init__(daemon=True)
        self.ser, self.unit = ser, unit
        self.regs = [0] * size
        self.coils = [False] * size
        self.stop_event = threading.Event()
        self.crc_errors = 0

    def run(self):
        while not self.stop_event.is_set():
            first = self.ser.read(1)
            if not first:
                continue
            deadline = time.perf_counter() + 0.5
            frame = first + read_exact(self.ser, 1, deadline)
            if len(frame) < 2:
                continue
            fc = frame[1]
            if fc in (1, 2, 3, 4, 5, 6):
                frame += read_exact(self.ser, 6, deadline)
            elif fc in (15, 16):
                frame += read_exact(self.ser, 5, deadline)
                if len(frame) == 7:
                    frame += read_exact(self.ser, frame[6] + 2, deadline)
            else:
                frame += read_until_idle(self.ser, deadline)
            if not crc_ok(frame):
                self.crc_errors += 1
                self.ser.reset_input_buffer()
                continue
            if frame[0] != self.unit:
                continue  # addressed to another unit: stay silent
            self.ser.write(add_crc(bytes([self.unit]) + self.handle(frame[1:-2])))

    def _check(self, table, addr, qty):
        if qty < 1 or addr + qty > len(table):
            raise ModbusException(2)

    def handle(self, pdu):
        fc, data = pdu[0], pdu[1:]
        try:
            if fc == 1:
                addr, qty = struct.unpack(">HH", data[:4])
                self._check(self.coils, addr, qty)
                out = bytearray((qty + 7) // 8)
                for i, bit in enumerate(self.coils[addr:addr + qty]):
                    if bit:
                        out[i // 8] |= 1 << (i % 8)
                return bytes([fc, len(out)]) + out
            if fc == 3:
                addr, qty = struct.unpack(">HH", data[:4])
                if qty > 125:
                    raise ModbusException(3)
                self._check(self.regs, addr, qty)
                return bytes([fc, qty * 2]) + struct.pack(f">{qty}H", *self.regs[addr:addr + qty])
            if fc == 5:
                addr, val = struct.unpack(">HH", data[:4])
                if val not in (0xFF00, 0x0000):
                    raise ModbusException(3)
                self._check(self.coils, addr, 1)
                self.coils[addr] = val == 0xFF00
                return pdu
            if fc == 6:
                addr, val = struct.unpack(">HH", data[:4])
                self._check(self.regs, addr, 1)
                self.regs[addr] = val
                return pdu
            if fc == 16:
                addr, qty, count = struct.unpack(">HHB", data[:5])
                if count != qty * 2:
                    raise ModbusException(3)
                self._check(self.regs, addr, qty)
                self.regs[addr:addr + qty] = struct.unpack(f">{qty}H", data[5:5 + count])
                return pdu[:5]
            raise ModbusException(1)
        except ModbusException as e:
            return bytes([fc | 0x80, e.code])


# ---------------------------------------------------------------- master

class Master:
    def __init__(self, ser, unit, timeout):
        self.ser, self.unit, self.timeout = ser, unit, timeout
        self.last_rtt = 0.0

    def request(self, pdu, unit=None):
        unit = self.unit if unit is None else unit
        self.ser.reset_input_buffer()
        start = time.perf_counter()
        self.ser.write(add_crc(bytes([unit]) + pdu))
        deadline = start + self.timeout
        frame = read_exact(self.ser, 2, deadline)
        if len(frame) < 2:
            raise TimeoutError("no response")
        fc = frame[1]
        if fc & 0x80:
            frame += read_exact(self.ser, 3, deadline)
        elif fc in (1, 2, 3, 4):
            frame += read_exact(self.ser, 1, deadline)
            if len(frame) == 3:
                frame += read_exact(self.ser, frame[2] + 2, deadline)
        else:
            frame += read_exact(self.ser, 6, deadline)
        self.last_rtt = time.perf_counter() - start
        if not crc_ok(frame):
            raise IOError(f"bad/short response: {frame.hex(' ')}")
        if frame[0] != unit:
            raise IOError(f"response from unit {frame[0]}, expected {unit}")
        if fc & 0x80:
            raise ModbusException(frame[2])
        return frame[1:-2]

    def read_holding(self, addr, qty):
        resp = self.request(struct.pack(">BHH", 3, addr, qty))
        return list(struct.unpack(f">{qty}H", resp[2:2 + qty * 2]))

    def write_register(self, addr, val):
        return self.request(struct.pack(">BHH", 6, addr, val))

    def write_registers(self, addr, values):
        pdu = struct.pack(f">BHHB{len(values)}H", 16, addr, len(values), len(values) * 2, *values)
        return self.request(pdu)

    def read_coils(self, addr, qty):
        resp = self.request(struct.pack(">BHH", 1, addr, qty))
        return [bool(resp[2 + i // 8] >> (i % 8) & 1) for i in range(qty)]

    def write_coil(self, addr, on):
        return self.request(struct.pack(">BHH", 5, addr, 0xFF00 if on else 0))


# ---------------------------------------------------------------- tests

def expect_exception(fn, code):
    try:
        fn()
    except ModbusException as e:
        assert e.code == code, f"got exception {e.code}, expected {code}"
        return
    raise AssertionError(f"no exception, expected {code}")


def expect_timeout(fn):
    try:
        fn()
    except TimeoutError:
        return
    raise AssertionError("got a response, expected silence")


def run_tests(m: Master, cycles):
    vals = [random.randint(0, 0xFFFF) for _ in range(10)]
    tests = [
        ("FC06 write single register",
         lambda: m.write_register(0, 0x1234)),
        ("FC03 read back single register",
         lambda: (lambda r: None if r == [0x1234] else _fail(r))(m.read_holding(0, 1))),
        ("FC16 write 10 registers",
         lambda: m.write_registers(10, vals)),
        ("FC03 read back 10 registers",
         lambda: (lambda r: None if r == vals else _fail(r))(m.read_holding(10, 10))),
        ("FC03 read 125 registers (max frame)",
         lambda: (lambda r: None if len(r) == 125 else _fail(len(r)))(m.read_holding(0, 125))),
        ("FC05 write coil ON",
         lambda: m.write_coil(3, True)),
        ("FC01 read coils",
         lambda: (lambda r: None if r[3] and not r[2] else _fail(r))(m.read_coils(0, 8))),
        ("Exception 02: illegal address",
         lambda: expect_exception(lambda: m.read_holding(500, 1), 2)),
        ("Exception 01: illegal function",
         lambda: expect_exception(lambda: m.request(bytes([0x41])), 1)),
        ("Wrong unit id -> no response",
         lambda: expect_timeout(lambda: m.request(struct.pack(">BHH", 3, 0, 1), unit=m.unit + 1))),
    ]

    # fail fast: if the slave doesn't answer at all, the remaining tests would only time out
    try:
        m.read_holding(0, 1)
    except TimeoutError:
        print("  FAIL  no response from slave at all - check wiring/power; remaining tests skipped")
        return 0, len(tests) + (1 if cycles else 0)

    passed = 0
    for name, fn in tests:
        try:
            start = time.perf_counter()
            fn()
            print(f"  PASS  {name:<40} {(time.perf_counter() - start) * 1000:7.1f} ms")
            passed += 1
        except Exception as e:
            print(f"  FAIL  {name:<40} {e}")

    # stress: random write/read-back cycles
    errors, rtts = 0, []
    for _ in range(cycles):
        addr = random.randint(0, 190)
        data = [random.randint(0, 0xFFFF) for _ in range(random.randint(1, 10))]
        try:
            m.write_registers(addr, data)
            rtts.append(m.last_rtt)
            if m.read_holding(addr, len(data)) != data:
                raise AssertionError("mismatch")
            rtts.append(m.last_rtt)
        except Exception:
            errors += 1
    if cycles:
        ok = errors == 0
        passed += ok
        avg = sum(rtts) / len(rtts) * 1000 if rtts else 0
        mx = max(rtts) * 1000 if rtts else 0
        print(f"  {'PASS' if ok else 'FAIL'}  {cycles} write/read-back cycles, {errors} errors"
              f"   rtt avg {avg:.1f} ms, max {mx:.1f} ms")
    return passed, len(tests) + (1 if cycles else 0)


def _fail(got):
    raise AssertionError(f"unexpected value: {got}")


def run_direction(label, master_port, slave_port, args):
    print(f"\n=== {label}: master on {master_port}, slave (unit {args.unit}) on {slave_port} ===")
    ms = open_port(master_port, args.baud, args.parity)
    ss = open_port(slave_port, args.baud, args.parity)
    slave = Slave(ss, args.unit)
    slave.start()
    try:
        passed, total = run_tests(Master(ms, args.unit, args.timeout), args.cycles)
    finally:
        slave.stop_event.set()
        slave.join(1)
        ms.close()
        ss.close()
    if slave.crc_errors:
        print(f"  slave saw {slave.crc_errors} frame(s) with bad CRC")
    print(f"  --> {passed}/{total} passed")
    return passed == total


def main():
    p = argparse.ArgumentParser(description="Modbus RTU loopback test between two serial ports")
    p.add_argument("--rs232", default="COM5", help="RS232-side port (default COM5)")
    p.add_argument("--rs485", default="COM6", help="RS485-side port (default COM6)")
    p.add_argument("--baud", type=int, default=9600)
    p.add_argument("--parity", choices="NEO", default="N")
    p.add_argument("--unit", type=int, default=1, help="slave unit id (default 1)")
    p.add_argument("--timeout", type=float, default=1.0, help="response timeout, s")
    p.add_argument("--cycles", type=int, default=100, help="stress write/read cycles")
    p.add_argument("--direction", choices=["both", "232-master", "485-master"], default="both")
    args = p.parse_args()

    print(f"Modbus RTU loopback test, {args.baud} 8{args.parity}1")
    ok = True
    if args.direction in ("both", "232-master"):
        ok &= run_direction("RS232 master -> RS485 slave", args.rs232, args.rs485, args)
    if args.direction in ("both", "485-master"):
        ok &= run_direction("RS485 master -> RS232 slave", args.rs485, args.rs232, args)
    print("\nRESULT:", "ALL PASSED" if ok else "FAILURES")
    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()