用Python实现Modbus-RTU协议及串口调试（一）

最近由于要测试几块客户使用的现场仪表的通信(Modbus-RTU协议)，就用Python写了个Modbus-RTU协议的串口调试模块，主要涉及了bytes类型字节串的使用，串口模块pyserial的使用，循环冗余校验CRC计算模块crcmod的使用，以及struct内置模块的使用。如果没有安装以上模块请按下面命令安装。
pip install pyserial
pip install crcmod
实现CRC16校验
 
首先按照Modbus-RTU协议的规范，所有通信帧末尾都要有2个字节的CRC16字节的校验码，以保证通信的可靠性。所以首先要实现CRC16校验算法，我们可以直接使用crcmod模块的已有算法，代码如下： 
import crcmod
# CRC16校验，返回整型数
def crc16(veritydata):
    if not veritydata:
        return
    crc16 = crcmod.mkCrcFun(0x18005, rev=True, initCrc=0xFFFF, xorOut=0x0000)
    return crc16(veritydata)
其中函数参数veritydata是一个字节串bytes变量，表示要进行校验的数据。
 当然你也可以自己按Modbus-RTU协议给出的c语言算法自己用Python写一个。
 由于客户仪表主要使用了Modbus_RTU协议的03和04功能号，通过03和04功能号主机读取仪表从机的实时数据，用Python主要实现这两个功能号的协议，03和04功能号协议具体可以参考Modbus-RTU标准文档。 
用函数实现Modbus-RTU的03和04功能主机->从机的命令帧
 
代码如下： 
def mmodbus03or04(add, startregadd, regnum, funcode=3):
    if add < 0 or add > 0xFF or startregadd < 0 or startregadd > 0xFFFF or regnum < 1 or regnum > 0x7D:
        print("Error: parameter error")
        return
    if funcode != 3 and funcode != 4:
        print("Error: parameter error")
        return
    sendbytes = add.to_bytes(1, byteorder="big", signed=False)
    sendbytes = sendbytes + funcode.to_bytes(1, byteorder="big", signed=False) + startregadd.to_bytes(2, byteorder="big", signed=False) + \
                regnum.to_bytes(2, byteorder="big", signed=False)
    crcres = crc16(sendbytes)
    crc16bytes = crcres.to_bytes(2, byteorder="little", signed=False)
    sendbytes = sendbytes + crc16bytes
    return sendbytes
此函数一共有4个参数，含义如下。
 add：Modbus从站地址（仪表地址）
 startregadd：要读取的开始寄存器地址（从0开始的绝对地址）
 regnum：要读取的寄存器个数
 funcode：功能号，默认值是3
 此函数首先判断了从站地址是否超限，开始寄存器地址是否超限，寄存器个数是否超限，功能号是否正确等。
 然后将几个参数以及CRC16的校验码形成一个可以发送串口的字节串后返回。注意int整形的to_bytes成员函数，此函数用于将整形数转换为字节码，第一个参数是转换字节码的字节个数，第二个参数是转换是按大端模式（big）还是小端模式（little）。第三个参数标识整形数按有符号整形转换还是无符号整形转换。 
将Modbus从机返回的数据帧的解析为数值的函数
 
代码如下： 
def smodbus03or04(recvdata, valueformat=0, intsigned=False):
    if not recvdata:
        print("Error: data error")
        return
    if not checkcrc(recvdata):
        print("Error: crc error")
        return
    datalist = list(recvdata)
    if datalist[1] != 0x3 and datalist[1] != 0x4:
        print("Error: recv data funcode error")
        return
    bytenums = datalist[2]
    if bytenums % 2 != 0:
        print("Error: recv data reg data error")
        return
    retdata = []
    if valueformat == 0:
        floatnums = bytenums / 4
        print("float nums: ", str(floatnums))
        floatlist = [0, 0, 0, 0]
        for i in range(int(floatnums)):
            floatlist[1] = datalist[3+i*4]
            floatlist[0] = datalist[4+i*4]
            floatlist[3] = datalist[5+i*4]
            floatlist[2] = datalist[6+i*4]
            bfloatdata = bytes(floatlist)
            [fvalue] = struct.unpack('f', bfloatdata)
            retdata.append(fvalue)
            print(f'Data{i+1}: {fvalue:.3f}')
    elif valueformat == 1:
        shortintnums = bytenums / 2
        print("short int nums: ", str(shortintnums))
        for i in range(int(shortintnums)):
            btemp = recvdata[3+i*2:5+i*2]
            shortvalue = int.from_bytes(btemp, byteorder="big", signed=intsigned)
            retdata.append(shortvalue)
            print(f"Data{i+1}: {shortvalue}")
    return retdata
此函数的3个参数含义如下：
 recvdata：Modbus-RTU从站在接收03和04功能号命令帧后的发回主站的数据帧，bytes字节串类型。
 valueformat：寄存器中值的格式，0代表用2个寄存器4个字节表示一个单精度浮点数，1代表1个寄存器（2字节）存放1个16位整形值，默认为0,（仪表用的是单精度浮点数）
 intsigned：当寄存器数据值格式是整形时，true则按照有符号整形转换，false则按无符号整形转换。
 首先是对接收的数据帧的各种错误判断，CRC16校验值核对等判断。如果数据值格式是单精度浮点数则按照仪表规定的浮点数格式进行转换，先将接收的字节串转换为整形列表，然后根据浮点数个数进行迭代，将接收的数据转换为浮点数，并放入列表中返回，要注意的是如何将bytes类型转换为一个浮点数，这里使用了struct模块的unpack函数。将浮点数转换为bytes类型则使用struct模块的pack函数。具体使用方法请参考说明文档。当数据值格式是16位整形时，则将切片出来每个值对应的bytes字节串转换为整形数，并放到列表中返回。 
用串口读取仪表数据
 
串口读写使用pyserial模块，代码如下： 
if __name__ == '__main__':
    slaveadd = 1 	# modbus从站地址




    

    startreg = 0 	# 开始寄存器地址
    regnums = 40 	# 寄存器个数
    send_data = mmodbus03or04(slaveadd, startreg, regnums)
    print("send data : ", send_data.hex())
    com = serial.Serial("com3", 9600, timeout=0.8)
    starttime = time.time()
    com.write(send_data)
    recv_data = com.read(regnums*2+5)
    endtime = time.time()
    if len(recv_data) > 0:
        print("recv: ", recv_data.hex())
    print(f"used time: {endtime-starttime:.3f}")
    com.close()
    smodbus03or04(recv_data)
serial模块的Serial函数常用的3个参数是：串口号、波特率、超时时间(s)。这里在发送了命令帧后，根据命令帧中的寄存器个数计算出需要读取多少个字节的串口数据。读取后用解析函数得到具体数值。 
完整的代码如下： 
import serial
import crcmod
import time
import struct
# CRC16校验，返回整型数
def crc16(veritydata):
    if not veritydata:
        return
    crc16 = crcmod.mkCrcFun(0x18005, rev=True, initCrc=0xFFFF, xorOut=0x0000)
    return crc16(veritydata)
# 校验数据帧的CRC码是否正确
def checkcrc(data):
    if not data:
    	return False
    if len(data) <= 2:
        return False
    nocrcdata = data[:-2]
    oldcrc16 = data[-2:]
    oldcrclist = list(oldcrc16)
    crcres = crc16(nocrcdata)
    crc16byts = crcres.to_bytes(2, byteorder="little", signed=False)
    # print("CRC16:", crc16byts.hex())
    crclist = list(crc16byts)
    if oldcrclist[0] != crclist[0] or oldcrclist[1] != crclist[1]:
        return False
    return True
# Modbus-RTU协议的03或04读取保存或输入寄存器功能主-》从命令帧
def mmodbus03or04(add, startregadd, regnum, funcode=3):
    if add < 0 or add > 0xFF or startregadd < 0 or startregadd > 0xFFFF or regnum < 1 or regnum > 0x7D:
        print("Error: parameter error")
        return
    if funcode != 3 and funcode != 4:
        print("Error: parameter error")
        return
    sendbytes = add.to_bytes(1, byteorder="big", signed=False)
    sendbytes = sendbytes + funcode.to_bytes(1, byteorder="big", signed=False) + startregadd.to_bytes(2, byteorder="big", signed=False) + \
                regnum.to_bytes(2, byteorder="big", signed=False)
    crcres = crc16(sendbytes)
    crc16bytes = crcres.to_bytes(2, byteorder="little", signed=False)
    sendbytes = sendbytes + crc16bytes
    return sendbytes
# Modbus-RTU协议的03或04读取保持或输入寄存器功能从-》主的数据帧解析（浮点数2,1,4,3格式，16位短整形（定义正负数））
def smodbus03or04(recvdata, valueformat=0, intsigned=False):
    if not recvdata:
        print("Error: data error")
        return
    if not checkcrc(recvdata):
        print("Error: crc error")
        return
    datalist = list(recvdata)
    if datalist[1] != 0x3 and datalist[1] != 0x4:
        print("Error: recv data funcode error")
        return
    bytenums = datalist[2]
    if bytenums % 2 != 0:
        print("Error: recv data reg data error")
        return
    retdata = []
    if valueformat == 0:
        floatnums = bytenums / 4
        print("float nums: ", str(floatnums))
        floatlist = [0, 0, 0, 0]
        for i in range(int(floatnums)):
            floatlist[1] = datalist[3+i*4]
            floatlist[0] = datalist[4+i*4]
            floatlist[3] = datalist[5+i*4]
            floatlist[2] = datalist[6+i*4]
            bfloatdata = bytes(floatlist)
            [fvalue] = struct.unpack('f', bfloatdata)
            retdata.append(fvalue)
            print(f'Data{i+1}: {fvalue:.3f}')
    elif valueformat == 1:
        shortintnums = bytenums / 2
        print("short int nums: ", str(shortintnums))
        for i in range(int(shortintnums)):
            btemp = recvdata[3+i*2:5+i*2]
            shortvalue = int.from_bytes(btemp, byteorder="big", signed=intsigned)
            retdata.append(shortvalue)
            print(f"Data{i+1}: {shortvalue}")
    return retdata
if __name__ == '__main__':
    slaveadd = 1
    startreg = 0
    regnums = 40
    send_data = mmodbus03or04(slaveadd, startreg, regnums)
    print("send data : ", send_data.hex())
    com = serial.Serial("com3", 9600, timeout=0.8)
    starttime = time.time()
    com.write(send_data)
    recv_data = com.read(regnums*2+5)
    endtime = time.time()
    if len(recv_data) > 0:
        print("recv: ", recv_data.hex())
    print(f"used time: {endtime-starttime:.3f}")
    com.close()
    smodbus03or04(recv_data)
码字不易，如果本文对您有用请随手点个赞，谢谢！^_^
                    用Python实现Modbus-RTU协议及串口调试最近由于要测试几块客户使用的现场仪表的通信(Modbus-RTU协议)，就用Python写了个Modbus-RTU协议的串口调试模块，主要涉及了bytes类型字节串的使用，串口模块pyserial的使用，循环冗余校验CRC计算模块crcmod的使用，以及struct内置模块的使用。如果没有安装以上模块请按下面命令安装。pip install pyserialpip install crcmod实现CRC16校验首先按照Modbus-RTU协议的
				不太重要的注意事项：这些说明是为不使用 github 的人编写的
重要的提示：
 注意：在进行维护/重新配置时...运行“modbus-stop.py”。 您可以从文件夹视图中双击它。
 注意 2：完成维护或配置后，运行“modbus-start.py”。 这使数据收集
安装说明：
 Python 3 (  ) ... 版本应该是 Python 3.XX（安装在第一台机器上是 3.4.2
 Pip安装程序（  ...用于安装minimalmodbus）
 MinimalModbus（一般参见 //pypi.python.org/pypi/MinimalModbus/0.6... pip install -U minimummodbus ）
 安装后...将自述文件所在的“ModBus”目录复制到新计算机中
配置脚本：
 转到命令行并 cd 进入“ModBus”目录。
 运行：py
202210141442：优化功能（增加自动读取和单次读取可选）和界面美化。
202210141627：优化功能（添加菜单和读取优化，连接状态显示）
202210261446：优化功能（添加CRC校验工具）
202211021208：外观美化
ser = serial.Serial('/dev/ttyUSB0', 250000, timeout=1)
print ser.isOpen()
words="gggggggggggggggg"
while (1):
        print "send 256x\""+words+"\" to rem
# 连接端口 'com6', 超时0.8，比特率9600、8字节、无校验、停止位1
com = serial.Serial(port="com6", baudrate=9600, timeout=0.8, bytesize=8, parity='N', stopbits=1)
if com.is_open:  # 检测端口
    print("端口已...
				python玩转modbus
1. modbus协议简介
Modbus协议是一项应用层报文传输协议，包括ASCII / RTU / TCP三种报文类型，协议本身不定义物理层，只定义了控制器能够认识和使用的消息结构，而不管消息是经过何种网络进行通信的。
标准的Modbus协议物理层接口主要有RS232 / RS422 / RS485和以太网。采用Master/Slave主从方式通信
关于modbus协议更多更详细的介绍，可自行查阅
2. 环境配置
python 目前主流使用操作modbus协议的库有三
				python可以利用serial模块来和串口设备进行485或者232通讯。
当然，网上这类串口调试助手的小程序有很多，不过这些程序要么是要收费，只能试用30天，要么是不好用。
况且，别人写的程序，你只能使用，无法取出其中的数据来进行处理，所以，如果可以自己写一个程序，既方便使用，又可以随时随地使用其中的数据。
软件：python3.10
pycharm2021
硬件：window10电脑
串口485设备（国产流量计）
串口转usb线（电脑不带串口，只能转接）
准备好了后，就可以开始写程序了。
使编写modbus TCP和RTU主站和从站成为可能。
 它可以用于测试目的：它带有从属模拟器和带有基于Web的HMI的主控（好的HMI需要改进：）。
 它还可以用于创建需要通过Modbus进行通信的任何应用程序。 它是一个全栈实现，用于“实际应用程序”。
 得益于Python和令人难以置信的现有库集，它可以满足许多不同的需求：数据库日志记录，HMI，报告生成...
 modbus-tk与pymodbus不同，后者是python中modbus堆栈的另一种实现。
 modbus-tk试图限制依赖关系（即使Modbus RTU需要pyserial）。
 modbus-tk与tkInter没有链接。 tk代表“ testkit”或“ too
                    weixin_38841555: 
                    博主，请问下这个代码是作为客户端，modbus slave作为服务端的吗？
我们的设备是反的，代码是服务端，不指定硬件ip，服务端先发命令硬件（硬件里写了服务端ip）会回复。
请问这种该怎么调整呢？
                嵌入式Linux中使用SQLite3数据库
                    Silent starry sky: 
                    你要把头文件拷贝到源码目录下，链接时要指定sqlite3的库目录。