pyton 实现modbus_tcp

import modbus_tk.defines as md
import modbus_tk.modbus_tcp as mt


def connect_plc(ip,port):
    try:

        master = mt.TcpMaster(ip, port)
        master.set_timeout(5.5)
        print("连接成功")
        return master
    except Exception:
        print("plc通讯不上")

master_sr30=connect_plc('222.222.221.11',502)
# master_sr20=connect_plc('222.222.221.10',502)

class Plc():
    '''
    功能码                         编号              含义

    READ_COILS                   H01               读线圈
    READ_DISCRETE_INPUTS         H02               读离散输入
    READ_HOLDING_REGISTERS       H03              读保持寄存器
    READ_INPUT_REGISTERS         H04              读输入寄存器（模拟量）
    WRITE_SINGLE_COIL            H05              写单一线圈
    WRITE_SINGLE_REGISTER        H06              写单一寄存器
    WRITE_MULTIPLE_COILS         H15              写多个线圈
    WRITE_MULTIPLE_REGISTERS     H16              写多个寄存器

    '''

    # VERSION 2.0 引入单例模式

    def __init__(self,master):
        self.master=master

    def get_01(self, adr=0, num=0):  # 读线圈Q区
        db = self.master.execute(slave=1, function_code=md.READ_COILS, starting_address=adr, quantity_of_x=num)
        return db

    def get_02(self, adr=0, num=0):  # 读输入信号
        db = self.master.execute(slave=1, function_code=md.READ_DISCRETE_INPUTS, starting_address=adr,
                                 quantity_of_x=num)
        return db

    def get_03(self, adr=0, num=0):  # 读保持寄存器
        db = self.master.execute(slave=1, function_code=md.READ_HOLDING_REGISTERS, starting_address=adr,
                                 quantity_of_x=num)
        return db

    def get_04(self, adr=0, num=0):  # 读输入寄存器
        db = self.master.execute(slave=1, function_code=md.READ_INPUT_REGISTERS, starting_address=adr,
                                 quantity_of_x=num)
        return db

    def set_05(self, adr=0, value=0):  # 写单个线圈
        self.master.execute(slave=1, function_code=md.WRITE_SINGLE_COIL, starting_address=adr, output_value=value)

    def set_06(self, adr=0, value=0):  # 写单个寄存器
        self.master.execute(slave=1, function_code=md.WRITE_SINGLE_REGISTER, starting_address=adr,
                                 output_value=value)

    def set_15(self, adr=0, value=[]):  # 写多个线圈
        self.master.execute(slave=1, function_code=md.WRITE_MULTIPLE_COILS, starting_address=adr,
                                 output_value=value)

    def set_16(self, adr=0, value=[]):  # 写多个寄存器
        self.master.execute(slave=1, function_code=md.WRITE_MULTIPLE_REGISTERS, starting_address=adr,
                                 output_value=value)



SR_30 = Plc(master_sr30)  # 建立一个plc对象
# SR_20=Plc(master_sr20)  # 建立一个plc对象




if __name__=='__main__':
    import time
    while True:
        data=SR_30.get_03(66,5)
        print(data)
        time.sleep(0.2)

主界面

from PyQt5.QtCore import Qt, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QMainWindow, QApplication, QFrame, QLabel, QVBoxLayout, 
    QScrollArea, QStackedWidget, QLineEdit, QPushButton, QMessageBox
from PyQt5.uic import loadUi
import images_rc
from tools import Tool

class Window(QMainWindow):
    def __init__(self):
        super().__init__()
        self.tool=Tool(self)
        self.light_red = """
    min-width:30px;
    min-height:30px;
    max-width:30px;
    max-height:30px;
    border-radius:15px;
    border:1px solid gray;
    background-color:red;"""
        self.light_green = """
    min-width:30px;
    min-height:30px;
    max-width:30px;
    max-height:30px;
    border-radius:15px;
    border:1px solid gray;
    background-color:green;"""
        self.setup_ui()

    def setup_ui(self):
        '''
        初始化控件，加载到UI
        :return:
        '''
        loadUi('./resources/ui/test.ui', self)
        self.stackedWidget.setCurrentIndex(5)
        self.init_DI(100)
        self.init_DO(100)
        self.init_AI(30)
        self.init_AO(16)

    def init_DI(self, num: int):
        self.topFiller = QFrame()
        self.topFiller.setMinimumSize(1000, 1350)
        self.DI = [QLabel(self.topFiller) for _ in range(num)]
        for i in range(len(self.DI)):
            x = i % 8
            y = i // 8
            self.DI[i].setStyleSheet(self.light_red)
            self.DI[i].move(x * 100 + 50, y * 100 + 50)
            # 创建标签
            label = QLabel(f'DI-{i}', self.topFiller)
            label.move(x * 100 + 55, y * 100 + 90)
        self.vbox = QVBoxLayout()
        self.scroll = QScrollArea()
        self.scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
        self.scroll.setWidget(self.topFiller)
        self.vbox.addWidget(self.scroll)
        self.frame_DI.setLayout(self.vbox)

    def init_DO(self, num):
        self.DO = [QPushButton(self.scrollAreaWidgetContents) for _ in range(num)]
        for i in range(len(self.DO)):
            x = i % 8
            y = i // 8

            self.DO[i].setStyleSheet("QPushButton{ min-width:30px;min-height:30px;max-width:30px;max-height:30px;
                                                   border-radius:9px;border:6px solid gray;background-color:#f50000"
                                     "}"
                                     "QPushButton:hover{background-color:#b80000}"
                                     "QPushButton:checked{background-color:#55ff00}"
                                     )
            self.DO[i].move(x * 100 + 50, y * 100 + 50)
            self.DO[i].setProperty('address',i)
            self.DO[i].setCheckable(True)
            self.DO[i].clicked.connect(self.event_btn_click)
            # 创建标签
            label = QLabel(f'DO-{i}', self.scrollAreaWidgetContents)
            label.move(x * 100 + 55, y * 100 + 95)

    def init_AI(self, num):
        # self.scrollAreaWidgetContents_2.setStyleSheet("QWidget#scrollAreaWidgetContents_2{border-image: url(:/main_ui/imanges/bg7.jpg);}")
        self.AI = [QLineEdit(self.scrollAreaWidgetContents_2) for _ in range(num)]
        for i in range(len(self.AI)):
            x = i % 8
            y = i // 8
            self.AI[i].setStyleSheet('border:2px solid blue;background-color:transparent;color:red;font-size:25px;')
            self.AI[i].resize(80, 30)
            self.AI[i].setReadOnly(True)
            self.AI[i].move(x * 100 + 50, y * 100 + 50)
            # 创建标签
            label = QLabel(f'地址-{i}', self.scrollAreaWidgetContents_2)
            label.move(x * 100 + 75, y * 100 + 90)

    def init_AO(self, num):
        area = self.scrollAreaWidgetContents_3
        self.AO = [[QLineEdit(area), QLineEdit(area), QPushButton(f'地址-{i}写入', area)] for i in range(num)]
        for i in range(len(self.AO)):
            # 布局LineEdit 设置两行两列显示
            x = i % 2
            y = i // 2

            self.AO[i][0].setStyleSheet('border:2px solid blue;background-color:transparent;color:white;')
            self.AO[i][0].resize(80, 30)
            self.AO[i][0].setToolTip(f'输入0-{i}之间整数')
            self.AO[i][0].move(x * 400 + 50, y * 100 + 50)
            # 创建标签
            label = QLabel(f'地址-{i}', area)
            label.setStyleSheet('color:white;font-size:15px;')
            label.move(x * 400 + 75, y * 100 + 90)
        for i in range(len(self.AO)):
            # 布局LineEdit
            x = i % 2
            y = i // 2
            self.AO[i][1].setStyleSheet('border:2px solid red;background-color:transparent;color:white;')
            self.AO[i][1].resize(80, 30)
            self.AO[i][1].move(x * 400 + 100 + 50, y * 100 + 50)
            self.AO[i][1].setToolTip('输入0-27648之间整数')
            # 创建标签
            label = QLabel('数值', area)
            label.setStyleSheet('color:white;font-size:15px;')
            label.move(x * 400 + 100 + 80, y * 100 + 90)
        for i in range(len(self.AO)):
            # 布局QPushButton
            x = i % 2
            y = i // 2
            self.AO[i][2].setStyleSheet('border:2px solid red;')
            self.AO[i][2].resize(80, 30)
            self.AO[i][2].move(x * 400 + 100 + 50 + 100, y * 100 + 50)
            self.AO[i][2].setStyleSheet("QPushButton{background-color:#a700fa;border:1px solid gray;border-radius:5px;
            color:white;font-size:15px;}"
                                        "QPushButton:hover{background-color:#650098;}"
                                        "QPushButton:pressed{background-color:#210031;}")
            self.AO[i][2].setProperty('role',i)
            self.AO[i][2].clicked.connect(self.event_btn_AO_click)

    def label_change_color(self, label_name, status='init'):
        color = {'init': '#0b610B', 'green': '#9ACD32', 'red': '#FF4000'}[status]
        style = """
            min-width:30px;
            min-height:30px;
            max-width:30px;
            max-height:30px;
            border-radius:15px;
            border:2px solid gray;
            background-color:{};
            
        """.format(color)
        self.__getattribute__(label_name).setStyleSheet(style)

    @pyqtSlot()
    def on_pushButton_clicked(self):
        self.stackedWidget.setCurrentIndex(0)
        self.tool.parse_do_with_DI()


    @pyqtSlot()
    def on_pushButton_2_clicked(self):
        self.stackedWidget.setCurrentIndex(1)

    @pyqtSlot()
    def on_pushButton_3_clicked(self):
        self.stackedWidget.setCurrentIndex(2)
        self.tool.parse_do_with_AI()

    @pyqtSlot()
    def on_pushButton_4_clicked(self):
        self.stackedWidget.setCurrentIndex(3)

    @pyqtSlot()
    def on_btn_connect_clicked(self):
        ip=self.le_ip_adr.text().strip()
        port=self.le_port.text().strip()
        if ip=='' or port=='':
            QMessageBox.warning(None,'info','不能为空')
            return


    def event_btn_click(self):
        sender=self.sender()
        self.tool.parse_do_with_DO()
    def event_btn_AO_click(self):
        sender=self.sender()
        self.tool.parse_do_with_AO(sender)





if __name__ == "__main__":
    import sys

    app = QApplication(sys.argv)
    win = Window()
    win.show()
    sys.exit(app.exec_())

模型处理

from PyQt5.QtCore import QObject, pyqtSignal, QThread, QTimer
from PyQt5.QtWidgets import QMessageBox, QWidget, QPushButton
from plc_modbus_v2 import SR_30 #引入单例模式



class Tool(QObject):
    '''处理逻辑'''

    def __init__(self, parent=None, *args, **kwargs):
        super().__init__(parent, *args, **kwargs)
        self.parent=parent
        self.w0=0x0000
    def parse_do_with_DO(self):
        if self.parent.DO[0].isChecked():
            self.w0=self.w0|0x0100
        else:
            self.w0=self.w0&0xfeff

        if self.parent.DO[1].isChecked():
            self.w0 = self.w0 | 0x0200
        else:
            self.w0 = self.w0 & 0xfdff

        if self.parent.DO[2].isChecked():
            self.w0 = self.w0 | 0x0400
        else:
            self.w0 = self.w0 & 0xfbff

        if self.parent.DO[3].isChecked():
            self.w0 = self.w0 | 0x0800
        else:
            self.w0 = self.w0 & 0xf7ff

        if self.parent.DO[4].isChecked():
            self.w0 = self.w0 | 0x1000
        else:
            self.w0 = self.w0 & 0xefff

        if self.parent.DO[5].isChecked():
            self.w0 = self.w0 | 0x2000
        else:
            self.w0 = self.w0 & 0xdfff

        if self.parent.DO[6].isChecked():
            self.w0 = self.w0 | 0x4000
        else:
            self.w0 = self.w0 & 0xbfff

        if self.parent.DO[7].isChecked():
            self.w0 = self.w0 | 0x8000
        else:
            self.w0 = self.w0 & 0x7fff

        if self.parent.DO[8].isChecked():
            self.w0 = self.w0 | 0x0001
        else:
            self.w0 = self.w0 & 0xfffe

        if self.parent.DO[9].isChecked():
            self.w0 = self.w0 | 0x0002
        else:
            self.w0 = self.w0 & 0xfffd

        if self.parent.DO[10].isChecked():
            self.w0 = self.w0 | 0x0004
        else:
            self.w0 = self.w0 & 0xfffb

        if self.parent.DO[10].isChecked():
            self.w0 = self.w0 | 0x0008
        else:
            self.w0 = self.w0 & 0xfff7

        if self.parent.DO[11].isChecked():
            self.w0 = self.w0 | 0x0010
        else:
            self.w0 = self.w0 & 0xffef

        if self.parent.DO[12].isChecked():
            self.w0 = self.w0 | 0x0020
        else:
            self.w0 = self.w0 & 0xffdf

        if self.parent.DO[13].isChecked():
            self.w0 = self.w0 | 0x0040
        else:
            self.w0 = self.w0 & 0xffbf

        SR_30.set_06(2,self.w0)

    def parse_do_with_DI(self):
        try:
            vw0=SR_30.get_03(66, 1)[0]
            vw2=SR_30.get_03(67,1)[0]
            vw4=SR_30.get_03(68,1)[0]

            data1=[vw0&i for i in[1,2,4,8,16,32,64,128,256,512,1024,2048,4096,8192,16384,32768]]
            for i,m in enumerate(data1):
                if m >0:
                    self.parent.DI[i].setStyleSheet(self.parent.light_green)
                else:
                    self.parent.DI[i].setStyleSheet(self.parent.light_red)

            data2 = [vw2 & i for i in [1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768]]
            for i, m in enumerate(data2):
                if m > 0:
                    self.parent.DI[i+16].setStyleSheet(self.parent.light_green)
                else:
                    self.parent.DI[i+16].setStyleSheet(self.parent.light_red)

            data3 = [vw4 & i for i in [1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768]]
            for i, m in enumerate(data3):
                if m > 0:
                    self.parent.DI[i + 32].setStyleSheet(self.parent.light_green)
                else:
                    self.parent.DI[i + 32].setStyleSheet(self.parent.light_red)
        except Exception as e:
            print(e)
    def parse_do_with_AI(self):
        try:
            data_tuple=SR_30.get_03(69, 30)
            for i,value in enumerate(data_tuple):
                self.parent.AI[i].setText(str(value))
        except Exception as e:
            print(e)
    def parse_do_with_AO(self,sender):
        # SR_30.set_06(3,2345)
        try:
            m=sender.property('role')
            adress=int(self.parent.AO[m][0].text())
            value=int(self.parent.AO[m][1].text())
            # print(adress,value)
            SR_30.set_06(adress,value)
        except Exception as e:
            print(e)












if __name__=='__main__':
    import time

    while True:

        # print(PLC.get_03(66, 5))
        # PLC.set_06(2,7)
        time.sleep(0.5)

资源文件

最后

以上就是称心世界为你收集整理的pyton 实现modbus_tcp的全部内容，希望文章能够帮你解决pyton 实现modbus_tcp所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。