How to create Modbus/RTU request in Python

If you have serial Modbus/RTU slaves attached to embebbed serial modules, then you'll need to create the original Modbus/RTU requests to send. This is quite easily done with Python.
In the last weeks I have researched a lot about this subject. Many forums say to use libraries: Pymodbus, Minimal Modbus or Modbus-tk. Tested all and say: that it is easier to use and create PySerial polling messages directly than using these libraries.

Requirement: Download PySerial Library

 These tests have been compiled in python 2.7


First i created the CRC16.py

What is CRC?

 Wikipedia says:
"A cyclic redundancy check (CRC) is an error-detecting code commonly used in digital networks and storage devices to detect accidental changes to raw data. Blocks of data entering these systems get a short check value attached, based on the remainder of a polynomial division of their contents; on retrieval the calculation is repeated, and corrective action can be taken against presumed data corruption if the check values do not match." 

 CRC16 on Python:

#!/usr/bin/python   
# crc16_Init() - Initialize the CRC-16 table (crc16_Table[])
def init_table( ):
    global table

    if( (len( table) == 256) and (table[1] == 49345)):
        # print "Table already init!"
        return
   
    lst = []
    i = 0
    while( i < 256):
        data = (i << 1)
        crc = 0
        j = 8
        while( j > 0):
            data >>= 1
            if( (data ^ crc) & 0x0001):
                crc = (crc >> 1) ^ 0xA001
            else:
                crc >>= 1
            j -= 1
           
        lst.append( crc)
        # print "entry %d = %x" % ( i, table[i])
        i += 1

    table = tuple( lst)      
    return

# given a Byte, Calc a modbus style CRC-16 by look-up table
def calcByte( ch, crc):
    init_table( )
    if( type(ch) == type("c")):
        by = ord( ch)
    else:
        by = ch
    crc = (crc >> 8) ^ table[(crc ^ by) & 0xFF]
    return (crc & 0xFFFF)

def calcString( st, crc):
    init_table()
    # print "st = ", list(st)
    for ch in st:
        crc = (crc >> 8) ^ table[(crc ^ ord(ch)) & 0xFF]
        # print " crc=%x" % crc
    return crc

# EXECUTE
table = tuple()


def main():
    testCRC()

if __name__ == "__main__":
    main()

 Now you import the function CRC16 on the modbus application

MODBUS on Python

This is my modbus simulator (Slave).

#!/usr/bin/python   
#!/usr/bin/env python

import serial
import time



#Import function CRC16
from CRC16 import calcString

# SERIAL PC CONFIG.
ser = serial.Serial(port='COM2',baudrate=19200)

# FUNCTIONS
# Decimal to Hex.
def dec2hex(n):
    lo = n & 0x00FF
    hi = (n & 0xFF00) >> 8
    return chr(hi) + chr(lo)
    #return "%02x" % n

# Hex. to Decimal
def hex2dec(s):
    return int(s, 16)

# Invert byte ( LO and HI )
def swapLoHi(n):
    lo = n & 0x00FF
    hi = (n & 0xFF00) >> 8
    return  lo << 8 | hi

# Calc. CRC16
def stCRC(msg):
    crc = calcString(msg, 0xFFFF)
    crc = swapLoHi(crc)
    return dec2hex(crc)
    

def HiLo(n):
    lo = n & 0x00FF
    hi = (n & 0xFF00) >> 8
    return  hi | lo
    
# MODBUS PROTOCOL DEFINES
READ_HOLDING = 3
READ_INPUT = 4
PRESET_SINGLE = 6
PRESET_MULTIPLE = 10
    
# INPUT VALUES
nb = input('Value 1: ')
nc = input('Value 2: ')
nd = input('Value 3: ')
ne = input('Value 4: ')

# CONVERT VALUE TO STRING HEX
value1=  chr(nb)    

value2 = chr(nc)
value3 = chr(nd)

value4 = chr(ne)

### DEBUG MSG. ###
ReadHolding = "\x01\x03\x02\x00" + value1
stMsg = ReadHolding + stCRC(ReadHolding)
ReadInput1 = "\x02\x04\x02\x00" + value2 + "\x00"+ value3 +"\x84"
ReadInput2 = "\x03\x04\x02\x00" + value2+ "\x00\x0a\x84" 
PresetSingle1 = "\x01\x06\x00\x00\x00\x0A"
PresetMultiple1 = "\x01\x10\x70\x00\x06"
PresetMultiple = PresetMultiple1 + stCRC(PresetMultiple1)
PresetSingle = PresetSingle1 + stCRC(PresetSingle1)

## LOOP MODBUS
while True:
    out = ''
# let's wait one second before reading output (let's give device time to answer)
    time.sleep(1)
    
# MODBUS READ BUFFER
    while ser.inWaiting() > 0:
        out += ser.read(1)

### DEBUG ###
    if out != '':
        print "uC Response"
        print out[0].encode('hex_codec') 
        print out[1].encode('hex_codec') 
        print out[2].encode('hex_codec')
        print out[3].encode('hex_codec')
        print out[4].encode('hex_codec')
        print out[5].encode('hex_codec')
        print out[6].encode('hex_codec')
        print out[7].encode('hex_codec')

# GET PROTOCOL
    value = int(out[1].encode('hex_codec'),16)
    print value 

# READ HOLDING MSG.
    if value is READ_HOLDING:
        print "read holding"
        ReadHolding = "\x01\x03\x02\x00" + value1
        stMsg = ReadHolding + stCRC(ReadHolding)
        ser.write(stMsg)
        time.sleep(1) # for 100 millisecond delay
       
# READ INPUT MSG.
    elif value is READ_INPUT:
        print "read input"
        nc = nc + 1
        nd = nd + 1
        ne = ne + 1
             
        value2 = chr(nc)
        value3 = chr(nd)
        value4 = chr(ne)
        ReadInput1 = "\x01\x04\x22\x00" + value1+ "\x00"+ value2 + "\x00" + value3 + "\x00"            
        stMsg1 = ReadInput1 + stCRC(ReadInput1)
        ser.write (stMsg1)
        time.sleep(1) #for 100 millisecond delay

# PRESET SINGLE MSG.
    elif value is PRESET_SINGLE:
        print "preset single"
        val = int(out[5].encode('hex_codec'),16)
        power = chr(val) # debug pot
        ser.write (PresetSingle)
       
               
# PRESET MULTIPLE MSG.
    elif value is PRESET_MULTIPLE:
        print "preset multiple"
        ser.write (PresetMultiple)
        time.sleep(1) # for 100 millisecond delay
    
    
    RecievedData = ""

    #while ser.inWaiting() > 0:
    #    RecievedData = ser.read(1)
    #print RecievedData  

What is Modbus?

The best site ever to explain it : Modbus

 
 

Read More