"""
Raw Modbus RTU Sniffer — listens to a serial port and prints decoded Modbus RTU frames.
Now includes decoding of address/count/value for popular function codes.
"""
import serial
import argparse
import time
import logging
import struct
logging.basicConfig()
log = logging.getLogger()
log.setLevel(logging.INFO)
def calculate_crc(data: bytes):
crc = 0xFFFF
for pos in data:
crc ^= pos
for _ in range(8):
lsb = crc & 0x0001
crc >>= 1
if lsb:
crc ^= 0xA001
return crc.to_bytes(2, 'little')
def validate_crc(frame: bytes):
if len(frame) < 4:
return False
data, received_crc = frame[:-2], frame[-2:]
return calculate_crc(data) == received_crc
def decode_payload(fc, payload):
if fc in [1, 2, 3, 4]:
if len(payload) >= 4:
address, count = struct.unpack(">HH", payload[:4])
return f"Read | Addr={address} | Count={count}"
elif fc in [5, 6]:
if len(payload) >= 4:
address, value = struct.unpack(">HH", payload[:4])
return f"Write Single | Addr={address} | Value={value}"
elif fc in [15, 16]:
if len(payload) >= 5:
address, count, byte_count = struct.unpack(">HHB", payload[:5])
return f"Write Multiple | Addr={address} | Count={count} | Bytes={byte_count}"
return f"Payload: {payload.hex()}"
def print_frame_info(frame: bytes):
if not validate_crc(frame):
log.warning(f"❌ Invalid CRC: {frame.hex()}")
return
unit_id = frame[0]
function_code = frame[1]
payload = frame[2:-2]
desc = decode_payload(function_code, payload)
log.info(f"📥 Unit={unit_id} | FC={function_code} — {desc}")
def read_frames(ser):
buffer = bytearray()
last_byte_time = time.time()
inter_char_timeout = 0.01
frame_timeout = 0.1
while True:
if ser.in_waiting:
byte = ser.read(1)
now = time.time()
if now - last_byte_time > frame_timeout and buffer:
print_frame_info(bytes(buffer))
buffer.clear()
buffer.append(byte[0])
last_byte_time = now
elif buffer and time.time() - last_byte_time > frame_timeout:
print_frame_info(bytes(buffer))
buffer.clear()
else:
time.sleep(0.005)
def main():
parser = argparse.ArgumentParser(description="Raw Modbus RTU Sniffer with decoding")
parser.add_argument('--port', required=True, help='Serial port (e.g. /dev/ttyUSB0)')
parser.add_argument('--baudrate', type=int, default=9600)
parser.add_argument('--parity', choices=['N', 'E', 'O'], default='N')
parser.add_argument('--stopbits', type=int, choices=[1, 2], default=1)
args = parser.parse_args()
try:
ser = serial.Serial(
port=args.port,
baudrate=args.baudrate,
parity={'N': serial.PARITY_NONE, 'E': serial.PARITY_EVEN, 'O': serial.PARITY_ODD}[args.parity],
stopbits=args.stopbits,
bytesize=8,
timeout=0
)
log.info(f"🔍 Listening on {args.port} at {args.baudrate} bps...")
read_frames(ser)
except Exception as e:
log.error(f"Failed to open serial port: {e}")
if __name__ == "__main__":
main()
