14、监测数据采集物联网应用开发步骤(10)

这篇具有很好参考价值的文章主要介绍了14、监测数据采集物联网应用开发步骤(10)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

  1. 监测数据采集物联网应用开发步骤(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))

运行测试结果如下图:

14、监测数据采集物联网应用开发步骤(10),物联网,python

测试案例中例举了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://www.toymoban.com/news/detail-685532.html

  1. 监测数据采集物联网应用开发步骤(11)

到了这里,关于14、监测数据采集物联网应用开发步骤(10)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 11、监测数据采集物联网应用开发步骤(8.2)

    监测数据采集物联网应用开发步骤(8.1) 新建TCP/IP Client线程类 com.zxy.tcp.ClientThread.py 新建tcp client数据接收插件类1 com.plugins.Usereflect.testClientReflectClass1.py 新建tcp client数据接收插件类2 com.plugins.Usereflect.testClientReflectClass2.py 在 com.zxy.main.Init_Page.py 中添加代码 TCP Client测试案例 Monit

    2024年02月10日
    浏览(43)
  • 7、监测数据采集物联网应用开发步骤(5.3)

    监测数据采集物联网应用开发步骤(5.2) 静态配置库数据库调用,新建全局变量初始化类 com.zxy.main.Init_Page.py 数据库操作测试 MonitorDataCmd.py 主文件中编写: if __name__ == \\\'__main__\\\' : 下编写 程序执行成功结果:自动生成center_data.db 打印出数据库数据 小测试:把上文的sql语句故意语法

    2024年02月10日
    浏览(42)
  • 13、监测数据采集物联网应用开发步骤(9.2)

    监测数据采集物联网应用开发步骤(9.1) TCP/IP Server开发 新建TCP/IP Server线程类 com.zxy.tcp.ServerThread.py 新建作为TCP Server接收数据拦截器插件类 com.plugins.usereflect.testServerReflectInClass1.py 新建作为TCP Server接收数据拦截器插件类 com.plugins.usereflect.testServerReflectInClass2.py 在 com.zxy.main.Init_

    2024年02月10日
    浏览(42)
  • 物联网数据采集网关在工厂数字化转型中的应用

    物联网数据采集网关能将各种传感器、执行器等设备连接在一起,通过收集、处理和传输来自各种物理设备的信息,实现数据的集成和分析,同时可通过云平台进行数据交互。它具有数据转换、数据处理、数据传输等功能,是工厂数字化转型的核心组件。随着科技的飞速发展

    2024年02月22日
    浏览(55)
  • iNeuOS工业互联网操作系统,高效采集数据配置与应用

    1. 概述 2. 通讯原理 3. 参数配置  1.   概述 某生产企业世界500强的集团能源管控平台项目建设,通过专线网络实现异地厂区数据集成, 每个终端能源仪表都有 IP 地址,总共有1000 多台能源表计,总共有将近10000 个数据点 。在集团端部署iNeuOS工业互联网操作系统,终端能源表

    2024年02月05日
    浏览(52)
  • 【雕爷学编程】MicroPython手册之 ESP32-CAM 物联网图像数据采集应用

    MicroPython是为了在嵌入式系统中运行Python 3编程语言而设计的轻量级版本解释器。与常规Python相比,MicroPython解释器体积小(仅100KB左右),通过编译成二进制Executable文件运行,执行效率较高。它使用了轻量级的垃圾回收机制并移除了大部分Python标准库,以适应资源限制的微控制

    2024年02月20日
    浏览(47)
  • 【IoT物联网】IoT小程序在展示中央空调采集数据和实时运行状态上的应用

      利用前端语言实现跨平台应用开发似乎是大势所趋,跨平台并不是一个新的概念,“一次编译、到处运行”是老牌服务端跨平台语言Java的一个基本特性。随着时代的发展,无论是后端开发语言还是前端开发语言,一切都在朝着减少工作量,降低工作成本的方向发展。  

    2024年02月16日
    浏览(40)
  • 水库安全监测方案(实时数据采集、高速数据传输)

    ​ 一、引言 水库的安全监测对于防止水灾和保障人民生命财产安全至关重要。为了提高水库安全监测的效率和准确性,本文将介绍一种使用星创易联DTU200和SG800 5g工业路由器部署的水库安全监测方案。 二、方案概述 本方案主要通过使用星创易联DTU200和SG800 5g工业路由器实现

    2024年02月08日
    浏览(50)
  • 桥梁安全监测系统中数据采集上传用 什么?

    背景 2023年7月6日凌晨时分,G5012恩广高速达万段230公里加80米处6号大桥部分桥面发生垮塌,导致造成2车受损后自燃,3人受轻伤。目前,四川省公安厅交通警察总队高速公路五支队十四大队民警已对现场进行双向管制。 作为世界第一桥梁大国,目前我国公路桥梁数量超过100万

    2024年02月12日
    浏览(42)
  • 工程监测振弦采集仪采集到的数据如何进行分析和处理

    工程监测振弦采集仪采集到的数据如何进行分析和处理 振弦采集仪是一个用于测量和记录物体振动的设备。它通过测量物体表面的振动来提取振动信号数据,然后将其转换为数字信号,以便进行分析和处理。在实际应用中,振弦采集仪是广泛应用于机械、建筑、航空航天和汽

    2024年02月12日
    浏览(50)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包