- 监测数据采集物联网应用开发步骤(9.2)
Modbus rtu协议开发
本章节在《监测数据采集物联网应用开发步骤(7)》基础上实现可参考《...开发步骤(7)》调试工具，本章节代码需要调用modbus_tk组件，阅读本章节前建议baidu熟悉modbus rtu协议内容
组件安装modbus_tk
pip3 install modbus_tk
验证是否安装成功，python中运行下列代码无异常则安装成功：
import modbus_tk
新建modbus协议管理类com.zxy.comport.ComModBus.py
#! python3
# -*- coding: utf-8 -
'''
Created on 2020年05月10日
@author: zxyong 13738196011
'''
import time,struct
from com.zxy.common import Com_Para
from com.zxy.common.Com_Fun import Com_Fun
from com.zxy.z_debug import z_debug
import modbus_tk.defines as cst
import modbus_tk.modbus_rtu as modbus_rtu
#监测数据采集物联网应用--modbus协议管理类
class ComModBus(z_debug):
def __init__(self):
pass
@staticmethod
def get_objAryRtuMaster(inputComPort):
master = Com_Fun.GetHashTableNone(Com_Para.objAryRtuMaster, inputComPort)
if master is None:
com_at = Com_Fun.GetHashTableNone(Com_Para.htComPort, inputComPort)
if com_at is not None:
master = modbus_rtu.RtuMaster(com_at.attSerial)
# 设定串口为从站
master.set_timeout(5.0)
master.set_verbose(True)
Com_Fun.SetHashTable(Com_Para.objAryRtuMaster, inputComPort, master)
return master
#字符串转16进制字节并+crc16校验，传入参数无crc校验
@staticmethod
def get_data_com_nocrc(inputComPort, CmdStr):
temReturn = None
try:
com_at = Com_Fun.GetHashTable(Com_Para.htComPort, inputComPort)
inputByte = bytes().fromhex(CmdStr)
inputByte = inputByte + ComModBus._getCrc16(inputByte)
if com_at.WritePortData(inputByte) > 0:
temReturn = com_at.attReturnValue
com_at.attReturnValue = None
except Exception as e:
temReturn = None
return temReturn
#字符串转字节发送,ascii发送
@staticmethod
def get_data_com_ascii(inputComPort, CmdStr):
temReturn = None
try:
com_at = Com_Fun.GetHashTable(Com_Para.htComPort, inputComPort)
inputByte = bytes(CmdStr, encoding="utf8")
if com_at.WritePortDataImmed(inputByte) > 0:
temReturn = com_at.attReturnValue
com_at.attReturnValue = None
except Exception as e:
temReturn = None
return temReturn
#字符串转16进制字节发送
@staticmethod
def get_data_com_hex(inputComPort, CmdStr):
temReturn = None
try:
com_at = Com_Fun.GetHashTable(Com_Para.htComPort, inputComPort)
inputByte = bytes().fromhex(CmdStr)
if com_at.WritePortData(inputByte) > 0:
comValue = com_at.attReturnValue
if comValue is None:
return None
temReturn = comValue
com_at.attReturnValue = None
except Exception as e:
temReturn = None
return temReturn
#字节发送
@staticmethod
def get_data_com_byte(inputComPort, inputByte):
temReturn = None
try:
com_at = Com_Fun.GetHashTable(Com_Para.htComPort, inputComPort)
if com_at.WritePortData(inputByte) > 0:
comValue = com_at.attReturnValue
if comValue is None:
return None
temReturn = comValue
com_at.attReturnValue = None
except Exception as e:
temReturn = None
return temReturn
#Modbus 04功能码发送
@staticmethod
def get_data_rtu_04(inputComPort, inputModbusAddr, inputModbusBegin, inputModbusLength):
red = []
try:
master = ComModBus.get_objAryRtuMaster(inputComPort)
if master is not None:
red = master.execute(int(inputModbusAddr), cst.READ_INPUT_REGISTERS, int(inputModbusBegin), int(inputModbusLength)) # 这里可以修改需要读取的功能码
time.sleep(0.1)
if isinstance(red, list) or isinstance(red, tuple):
return red
else:
return [""]
except Exception as e:
return [""]
#Modbus 03功能码发送
@staticmethod
def get_data_rtu_03(inputComPort, inputModbusAddr, inputModbusBegin, inputModbusLength):
read = []
try:
master = ComModBus.get_objAryRtuMaster(inputComPort)
if master is not None:
read = master.execute(inputModbusAddr, cst.READ_HOLDING_REGISTERS, inputModbusBegin, inputModbusLength) # 这里可以修改需要读取的功能码
time.sleep(0.1)
if isinstance(read, list) or isinstance(read, tuple):
return read
else:
return [""]
except Exception as e:
return [""]
#Modbus 写寄存器数据
@staticmethod
def set_data_rtu(inputComPort, inputModbusAddr, inputModbusBegin, inputValue):
read = []
try:
master = ComModBus.get_objAryRtuMaster(inputComPort)
if master is not None:
# 这里可以修改需要读取的功能码
if isinstance(inputValue, list) or isinstance(read, tuple):
read = master.execute(inputModbusAddr, cst.WRITE_MULTIPLE_REGISTERS, inputModbusBegin, output_value=inputValue)
else:
read = master.execute(inputModbusAddr, cst.WRITE_SINGLE_REGISTER, inputModbusBegin, output_value=inputValue)
if isinstance(read, list) or isinstance(read, tuple):
return read
else:
return [""]
except Exception as e:
return [""]
@staticmethod
def set_data_rtu2(inputComPort, inputModbusAddr, inputModbusBegin, inputValue):
read = []
try:
master = ComModBus.get_objAryRtuMaster(inputComPort)
if master is not None:
# 这里可以修改需要读取的功能码
if isinstance(inputValue, list) or isinstance(read, tuple):
read = master.execute(inputModbusAddr, cst.ANALOG_INPUTS, inputModbusBegin, output_value=inputValue)
else:
read = master.execute(inputModbusAddr, cst.ANALOG_INPUTS, inputModbusBegin, output_value=inputValue)
if isinstance(read, list) or isinstance(read, tuple):
return read
else:
return [""]
except Exception as e:
return [""]
@staticmethod
def _getCrc16(RtuStr):
b = 0xA001
# 16位寄存器
a = 0xFFFF
for byte in RtuStr:
# 1、把数据帧中的第一个8位字节与CRC寄存器中的低字节进行异或运算
a = a ^ byte
for i in range(8):
# 3、如果最低位为1：将CRC寄存器与一个预设的固定值（0A001H）进行异或运算
if a & 0x0001:
a = a >> 1
a = a ^ b
# 2、将CRC寄存器向右移一位，最高位填以0，最低位移出并检测
else:
a = a >> 1
aa = '0' * (6 - len(hex(a))) + hex(a)[2:]
# 获取低和高位
lo, hh = int(aa[:2], 16), int(aa[2:], 16)
hexbytes = bytes([hh, lo])
return hexbytes
#高低位 reverse: true高位在前 false低位在前
@staticmethod
def ReadFloat(n1, n2, reverse=False):
n = '%04x' % n1
m = '%04x' % n2
if reverse:
v = n + m
else:
v = m + n
y_bytes = bytes.fromhex(v)
y = struct.unpack('!f', y_bytes)[0]
y = round(y, 6)
return y
@staticmethod
def WriteFloat(value, reverse=False):
y_bytes = struct.pack('!f', value)
y_hex = ''.join(['%02x' % i for i in y_bytes])
n, m = y_hex[:-4], y_hex[-4:]
n, m = int(n, 16), int(m, 16)
if reverse:
v = [n, m]
else:
v = [m, n]
return v
@staticmethod
def ReadDint(n1,m1, reverse=False):
n ='%04x' % n1
m = '%04x' % m1
if reverse:
v = n + m
else:
v = m + n
y_bytes = bytes.fromhex(v)
y = struct.unpack('!i', y_bytes)[0]
return y
@staticmethod
def WriteDint(value, reverse=False):
y_bytes = struct.pack('!i', value)
y_hex = ''.join(['%02x' % i for i in y_bytes])
n, m = y_hex[:-4], y_hex[-4:]
n, m = int(n, 16), int(m, 16)
if reverse:
v = [n, m]
else:
v = [m, n]
return v
modbus rtu测试案例MonitorDataCmd.py主文件中编写：
from com.zxy.comport.ComModBus import ComModBus在 if __name__ == '__main__':下添加
#串口配置参数
Com_Para.ComPortList = "COM2,9600,8,0,A;COM4,9600,8,2,B"
#串口连接初始化
Init_Page.Start_ComPort()
#Modbus-rtu协议功能测试
temA01modbus = ComModBus()
#利用modbus_tk组件获取数据，参数:COM索引,modbus站地址,modbus起始位,modbus长度
read = temA01modbus.get_data_rtu_03("A",1,0,10)
print("获取10进制原始返回值=>"+str(read))
modValue = []
iIndex = 0.0
n1 = 0
n2 = 0
for temSV in read:
if iIndex % 2 != 0:
n2 = int(temSV)
#16进制单精转浮点
temMV = temA01modbus.ReadFloat(n1,n2,True)
modValue.append(temMV)
else:
n1 = int(temSV)
iIndex = iIndex + 1
print("获取读到并解析的寄存器浮点数=>"+str(modValue))
#利用modbus指令协议直接通过com口读取数据
temCmd = "010300A1000A942F"
bhex = temA01modbus.get_data_com_hex("A",temCmd)
read = str(binascii.b2a_hex(bhex).decode(Com_Para.U_CODE)).upper()
print("获取16进制返回值=>"+str(read))运行测试结果如下图:
测试案例中例举了2种方法都可以读到数据，任取其一即可。
针对
#利用modbus指令协议直接通过com口读取数据
temCmd = "010300A1000A942F"
指令解释如下:
01 站地址
03 功能码
00A1 寄存器开始地址(16进制)
000A 读取数据长度10个(16进制转10进制)
942F CRC校验码
获取16进制返回值=>010314429DA8F643060A3D420070A44343000043554CCD49A8返回值解释如下:
01 站地址
03 功能码
14 数据长度20(16进制转10进制)
429DA8F643060A3D420070A44343000043554CCD 数据值16进制
49A8 CRC校验码
429DA8F643060A3D420070A44343000043554CCD 数据值解析要依据相关对方开发说明文档，假设该数据为寄存器浮点数则按如下解析：
429D A8F6 ==>10进制浮点数 78.83
4306 0A3D ==>10进制浮点数 134.039
4200 70A4 ==>10进制浮点数 32.11
4343 0000 ==>10进制浮点数 195
4355 4CCD ==>10进制浮点数 213.300
可以利用该工具计算IEEE 754浮点数十六进制相互转换(32位,四字节,单精度)
在线进制转换-IEE754浮点数16进制转换
文章来源地址https://uudwc.com/A/Dy0Nk文章来源：https://uudwc.com/A/Dy0Nk