1. 監(jiān)測(cè)數(shù)據(jù)采集物聯(lián)網(wǎng)應(yīng)用開發(fā)步驟(9.2)

Modbus rtu協(xié)議開發(fā)

本章節(jié)在《監(jiān)測(cè)數(shù)據(jù)采集物聯(lián)網(wǎng)應(yīng)用開發(fā)步驟(7)》基礎(chǔ)上實(shí)現(xiàn)可參考《...開發(fā)步驟(7)》調(diào)試工具，本章節(jié)代碼需要調(diào)用modbus_tk組件，閱讀本章節(jié)前建議baidu熟悉modbus rtu協(xié)議內(nèi)容

組件安裝modbus_tk

pip3 install modbus_tk

驗(yàn)證是否安裝成功，python中運(yùn)行下列代碼無(wú)異常則安裝成功：

import?modbus_tk

新建modbus協(xié)議管理類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

#監(jiān)測(cè)數(shù)據(jù)采集物聯(lián)網(wǎng)應(yīng)用--modbus協(xié)議管理類
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)           
                # 設(shè)定串口為從站
                master.set_timeout(5.0)
                master.set_verbose(True)
                Com_Fun.SetHashTable(Com_Para.objAryRtuMaster, inputComPort, master)
        return master
    
    #字符串轉(zhuǎn)16進(jìn)制字節(jié)并+crc16校驗(yàn)，傳入?yún)?shù)無(wú)crc校驗(yàn)
    @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
    
    #字符串轉(zhuǎn)字節(jié)發(fā)送,ascii發(fā)送
    @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
    
    #字符串轉(zhuǎn)16進(jìn)制字節(jié)發(fā)送
    @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
    
    #字節(jié)發(fā)送
    @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功能碼發(fā)送
    @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功能碼發(fā)送 
    @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 寫寄存器數(shù)據(jù)
    @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、把數(shù)據(jù)幀中的第一個(gè)8位字節(jié)與CRC寄存器中的低字節(jié)進(jìn)行異或運(yùn)算
            a = a ^ byte
            for i in range(8):
                # 3、如果最低位為1：將CRC寄存器與一個(gè)預(yù)設(shè)的固定值（0A001H）進(jìn)行異或運(yùn)算
                if a & 0x0001:
                    a = a >> 1
                    a = a ^ b 
                # 2、將CRC寄存器向右移一位，最高位填以0，最低位移出并檢測(cè)
                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測(cè)試案例MonitorDataCmd.py主文件中編寫：

from com.zxy.comport.ComModBus import ComModBus

在????if?__name__ == '__main__':下添加

        #串口配置參數(shù)
        Com_Para.ComPortList = "COM2,9600,8,0,A;COM4,9600,8,2,B"
        #串口連接初始化
        Init_Page.Start_ComPort()
        #Modbus-rtu協(xié)議功能測(cè)試
        temA01modbus = ComModBus()
        #利用modbus_tk組件獲取數(shù)據(jù)，參數(shù):COM索引,modbus站地址,modbus起始位,modbus長(zhǎng)度
        read = temA01modbus.get_data_rtu_03("A",1,0,10)
        print("獲取10進(jìn)制原始返回值=>"+str(read))
        modValue = []
        iIndex = 0.0
        n1 = 0
        n2 = 0
        for temSV in read:
            if iIndex % 2 != 0:
                n2 = int(temSV)
                #16進(jìn)制單精轉(zhuǎn)浮點(diǎn)
                temMV = temA01modbus.ReadFloat(n1,n2,True)
                modValue.append(temMV)
            else:
                n1 = int(temSV)
            iIndex = iIndex + 1
        print("獲取讀到并解析的寄存器浮點(diǎn)數(shù)=>"+str(modValue))
        
        #利用modbus指令協(xié)議直接通過(guò)com口讀取數(shù)據(jù)
        temCmd = "010300A1000A942F"
        bhex = temA01modbus.get_data_com_hex("A",temCmd)
        read = str(binascii.b2a_hex(bhex).decode(Com_Para.U_CODE)).upper()
        print("獲取16進(jìn)制返回值=>"+str(read))

運(yùn)行測(cè)試結(jié)果如下圖:

測(cè)試案例中例舉了2種方法都可以讀到數(shù)據(jù)，任取其一即可。

針對(duì) ???????

#利用modbus指令協(xié)議直接通過(guò)com口讀取數(shù)據(jù)

??????temCmd = "010300A1000A942F"

指令解釋如下:

01 站地址

03 功能碼

00A1 寄存器開始地址(16進(jìn)制)

000A ?讀取數(shù)據(jù)長(zhǎng)度10個(gè)(16進(jìn)制轉(zhuǎn)10進(jìn)制)

942F CRC校驗(yàn)碼

獲取16進(jìn)制返回值=>010314429DA8F643060A3D420070A44343000043554CCD49A8返回值解釋如下:

01 站地址

03 功能碼

14 數(shù)據(jù)長(zhǎng)度20(16進(jìn)制轉(zhuǎn)10進(jìn)制)

429DA8F643060A3D420070A44343000043554CCD 數(shù)據(jù)值16進(jìn)制

49A8 CRC校驗(yàn)碼

429DA8F643060A3D420070A44343000043554CCD 數(shù)據(jù)值解析要依據(jù)相關(guān)對(duì)方開發(fā)說(shuō)明文檔，假設(shè)該數(shù)據(jù)為寄存器浮點(diǎn)數(shù)則按如下解析：

429D A8F6 ==>10進(jìn)制浮點(diǎn)數(shù) 78.83

4306 0A3D ==>10進(jìn)制浮點(diǎn)數(shù) 134.039

4200 70A4 ==>10進(jìn)制浮點(diǎn)數(shù) 32.11

4343 0000 ==>10進(jìn)制浮點(diǎn)數(shù) 195

4355 4CCD ==>10進(jìn)制浮點(diǎn)數(shù) 213.300

可以利用該工具計(jì)算IEEE 754浮點(diǎn)數(shù)十六進(jìn)制相互轉(zhuǎn)換(32位,四字節(jié),單精度)

在線進(jìn)制轉(zhuǎn)換-IEE754浮點(diǎn)數(shù)16進(jìn)制轉(zhuǎn)換文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-685532.html

1. 監(jiān)測(cè)數(shù)據(jù)采集物聯(lián)網(wǎng)應(yīng)用開發(fā)步驟(11)

到了這里，關(guān)于14、監(jiān)測(cè)數(shù)據(jù)采集物聯(lián)網(wǎng)應(yīng)用開發(fā)步驟(10)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！