【PLC+Python】snap7/Tkinter实现与西门子PLC通讯/可视化（2）——Python上位机_plc标签通讯与python

作者：小丑西瓜9 | 2024-06-07 12:02:44

踩

plc标签通讯与python

一背景说明

计划通过西门子 S7-1200（CPU 1212C-DCDCDC），进行PLC与设备间的数据监控。但 TIA Portal V15.1 的交互数据非专业人员很难一目了然，又不想专门购买西门子的可编程屏幕，所以拟采用 python-snap7 模块实现上位机与PLC的通信，并将运行监控抽象到 Tkinter 绘制的可视化GUI上，方便测试维护人员操作。

前文已经完成了PLC上面的组态以及DB数据块创建等操作（【PLC+Python】上位机通过snap7实现与西门子PLC通讯并用Tkinter可视化——（1）PLC DB块创建-CSDN博客），这篇文章主要描述上位机中python的通讯以及可视化相关内容。

二 snap7介绍

snap7是一个由意大利程序员开发的基于以太网与西门子S7系列PLC的通讯的开源库，类似于C#的S7.Net，但是它不单只支持Python，还支持Java、C/C++、C#等语言。官网地址如下：

snap7官网地址

而python-snap7则是snap7的python版本，有单独的文档以及使用说明，只能用于python，以下是官方文档及GitHub链接：

python-snap7官网地址

python-snap7 github链接

另外，官方还提供了支持多种操作系统的调试工具，可以方便预先进行通信调试，下载的方式如下：

三通讯建立

【1】博图TIA软件端放开PLC的通讯限制，并编译下载到PLC：

（i）“常规——防护与安全——连接机制——勾选‘允许来自远程对象的PUT/GET通信访问’”：

（ii）“数据库程序块——属性——取消勾选‘优化的块访问’”（该选项的作用可以参考TAI博图笔记知识:DB数据块的优化访问与非优化访问的区别）：

【2】python中导入snap7库，并测试一下能否连接：


import snap7
 
# 创建通讯客户端实例
plcObj = snap7.client.Client()
# 连接至PLC
plcObj.connect('192.168.5.1', 0, 1)
# 打印连接状态
print(f"连接状态：{plcObj.get_connected()}")
# 关闭连接
plcObj.disconnect()
# 打印连接状态
print(f"连接状态：{plcObj.get_connected()}")

测试结果如下：

至此能够正常连接到PLC。

四读取数据

根据【PLC+Python】上位机通过snap7实现与西门子PLC通讯并用Tkinter可视化——（1）PLC DB块创建-CSDN博客中创建的DB1，需要监控DB1中 dig_ctrl（偏移值0.0——0.2） / dig_fbk（偏移值2.0——2.1），核心代码如下：


import snap7
from snap7 import util
 
DEV_CTRL_DATA = [[False for i in range(3)] for j in range(4)]   #4台设备&3个控制数据(开/关/停)
DEV_FBK_DATA = [[False for i in range(2)] for j in range(4)]    ##4台设备&2个反馈数据(全开/全关)
 
# 创建通讯客户端实例
plcObj = snap7.client.Client()
# 连接至PLC
plcObj.connect('192.168.5.1', 0, 1)
# 读取数据
datazip = plcObj.db_read(1, 0, 4)  # 读取数据（DB序号为1，起始地址为0，读取长度4字节）
# 关闭连接
plcObj.disconnect()
# snap7解析
DEV_CTRL_DATA[0][0] = util.get_bool(datazip, 0, 0)
DEV_CTRL_DATA[0][1] = util.get_bool(datazip, 0, 1)
DEV_CTRL_DATA[0][2] = util.get_bool(datazip, 0, 2)
DEV_FBK_DATA[0][0] = util.get_bool(datazip, 2, 0)
DEV_FBK_DATA[0][1] = util.get_bool(datazip, 2, 1)
print("PLC数据解包：")
print(
    f"[设备1]：停指令：{DEV_CTRL_DATA[0][0]};开指令{DEV_CTRL_DATA[0][1]};关指令{DEV_CTRL_DATA[0][2]} / 全开位置:{DEV_FBK_DATA[0][0]};全关位置:{DEV_FBK_DATA[0][1]}\n"
)

测试结果如下：

五发送数据

发送数据即是修改DB1中 dig_ctrl（偏移值0.0——0.2）数据，核心代码如下：


import snap7
from snap7 import util
 
# 创建通讯客户端实例
plcObj = snap7.client.Client()
# 连接至PLC
plcObj.connect('192.168.5.1', 0, 1)
 
# 发送数据
boolData = bytearray(1)
util.set_bool(boolData, 0, 0, True)
util.set_bool(boolData, 0, 1, True)
util.set_bool(boolData, 0, 2, True)
plcObj.db_write(1, 0, boolData)
 
# 关闭连接
plcObj.disconnect()

其中，将DB1中偏移值0.0——0.2全部改成了TRUE，测试结果如下：

（！！！更多数据类型的读写方法可以参考：Python使用python-snap7实现西门子PLC通讯-CSDN博客，其中写的非常详细！）

六结合Tkinter界面进行操作

有了上面通讯建立/读写数据作为基础，再结合Tkinter模块做出的界面，就可以实现一个极简的GUI，方便对PLC连接的从机设备进行操作：

【1】界面设计以及主要操作方法如下：

（i）点击“连接”之后，首先读取输入的IP地址，格式校验正确之后，连接到对应的PLC；

（ii）连接成功之后，点击“轮询”开启线程，以300ms的间隔轮询PLC中DB1（偏移量2.0——2.1）数据，并将读取到的数据结果体现在全关/全开 LED灯亮灭上；

（iii）点击下面的关/停/开按钮，发送指令到PLC中DB1（偏移量0.0——0.2），以操控设备运行状态；

（iv）点击“断开”，即断开与PLC的连接；

【2】完整代码：


import tkinter as tk
from tkinter import ttk
from tkinter import messagebox
from PIL import Image, ImageTk
 
import snap7
from snap7 import util
import threading
import time
import re
 
 
# PLC连接参数
PLC_IP = '192.168.5.1'  # PLC的IP地址
IPV4_PATTERN = r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'   # IPv4地址的正则表达式
PLC_RACK = 0
PLC_SLOT = 1
PLC_DB_ID = 1   #收发数据的DB模块id
 
 
DEV_CTRL_DATA = [[False for i in range(3)] for j in range(4)]   #4台设备&3个控制数据(开/关/停)
DEV_FBK_DATA = [[False for i in range(2)] for j in range(4)]    ##4台设备&2个反馈数据(全开/全关)
 
# 全局变量，用于控制循环和连接状态
is_connected = False
is_running = False
is_cmd_st = False
is_cmd_op = False
is_cmd_cl = False
 
 
"""PLC监控系统类（PLC通信 & 主界面绘制）"""
class PlcGUI(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title('PLC监控系统')
        # 设置窗口在打开时默认最大化
        self.state('zoomed')
 
        comm_frame = tk.LabelFrame(self, text="通信")
        comm_frame.grid(row=0, column=0, padx=10, pady=10, sticky='nw')
        comm_label = tk.Label(comm_frame, text="IP地址：")
        comm_label.grid(row=0, column=0, padx=10, pady=10, sticky='ew')
        self.comm_addr = ttk.Entry(comm_frame, width=20)
        self.comm_addr.insert(0, PLC_IP)  # 插入默认IP地址
        self.comm_addr.grid(row=0, column=1, padx=10, pady=10, sticky='e')
        self.comm_start = ttk.Button(comm_frame, text="连接", command=self.start_query)
        self.comm_start.grid(row=1, column=0, padx=10, pady=10, sticky='w')
        self.comm_stop = ttk.Button(comm_frame, text="断开", command=self.stop_query)
        self.comm_stop.grid(row=1, column=1, padx=10, pady=10, sticky='e')
 
        # Create LEDFrames
        self.DEV_LISTS = [
            MonitorUnit(self, "设备1")
        ]
 
        self.DEV_LISTS[0].grid(row=2, column=0, padx=10, pady=10, sticky='news')
        self.grid_rowconfigure(0, weight=1)
        self.grid_columnconfigure(0, weight=1)
 
 
    def read_ipv4_addr(self):   #输入的IPV4地址校验
        ip_addr_input = self.comm_addr.get()    #获取文本框内容
        if re.match(IPV4_PATTERN, ip_addr_input):
            global PLC_IP
            PLC_IP = ip_addr_input
            return True
        else:
            return False
 
 
    def connect_plc(self):
        global is_connected
        try:
            client = snap7.client.Client()
            client.connect(PLC_IP, PLC_RACK, PLC_SLOT)
            is_connected = True
            self.comm_start['text'] = '轮询'
            messagebox.showinfo("连接成功", "已成功连接到PLC")
        except Exception as e:
            messagebox.showerror("连接失败", f"连接PLC时出错：{e}")
 
    def query_data(self):
        global is_connected, is_running, is_cmd_st, is_cmd_op, is_cmd_cl
        if not is_connected:
            return
        try:
            client = snap7.client.Client()
            client.connect(PLC_IP, PLC_RACK, PLC_SLOT)
            while is_running:
                # 在这里添加您的数据查询代码
                datazip = client.db_read(PLC_DB_ID, 0, 4)  # 读取数据（DB序号为1，起始地址为0，读取长度4字节）
                # snap7解析
                DEV_CTRL_DATA[0][0] = util.get_bool(datazip, 0, 0)
                DEV_CTRL_DATA[0][1] = util.get_bool(datazip, 0, 1)
                DEV_CTRL_DATA[0][2] = util.get_bool(datazip, 0, 2)
                DEV_FBK_DATA[0][0] = util.get_bool(datazip, 2, 0)
                DEV_FBK_DATA[0][1] = util.get_bool(datazip, 2, 1)
 
                # 监控界面LED灯状态刷新
                for i in range(1):
                    self.DEV_LISTS[i].chg_color_op('green') if DEV_FBK_DATA[i][0] == True else self.DEV_LISTS[i].chg_color_op('gray')
                    self.DEV_LISTS[i].chg_color_cl('red') if DEV_FBK_DATA[i][1] == True else self.DEV_LISTS[i].chg_color_cl('gray')
                # 控制命令下发
                boolData = bytearray(1)
                util.set_bool(boolData, 0, 0, is_cmd_st)
                util.set_bool(boolData, 0, 1, is_cmd_op)
                util.set_bool(boolData, 0, 2, is_cmd_cl)
                client.db_write(PLC_DB_ID, 0, boolData)
 
                time.sleep(0.3)  # 查询数据间隔时间，您可以根据需要调整这个时间间隔
        except Exception as e:
            messagebox.showerror("查询错误", f"查询PLC数据时出错：{e}")
            is_connected = False
            self.comm_start['text'] = '连接'
        finally:
            client.disconnect()
 
    def start_query(self):
        if not self.read_ipv4_addr():
            messagebox.showerror("校验失败", "输入的IP地址不符合IPv4格式要求！")
            return
 
        global is_running
        if is_connected:
            is_running = True
            self.comm_start.state(['disabled']) #与PLC建立交互之后就禁用按钮，以示正在运行
            query_thread = threading.Thread(target=self.query_data)
            query_thread.setDaemon(True)
            query_thread.start()
        else:
            self.connect_plc()
 
    def stop_query(self):
        if not self.read_ipv4_addr():
            messagebox.showerror("校验失败", "输入的IP地址不符合IPv4格式要求！")
            return
 
        global is_connected, is_running
        is_running = False
        self.comm_start.state(['!disabled'])  # 重启启用开始按钮
        if is_connected:
            client = snap7.client.Client()
            client.connect(PLC_IP, PLC_RACK, PLC_SLOT)
            client.disconnect()
            is_connected = False
            self.comm_start['text'] = '连接'
            messagebox.showinfo("已断开", "已成功断开与PLC的连接")
 
 
"""监控单元类（每个监控单元包含两个LED灯以及三个控制按钮）"""
class MonitorUnit(ttk.LabelFrame):
    def __init__(self, master, title):
        super().__init__(master, text=title)
 
        # 创建LED灯
        self.close_cv = tk.Canvas(self, width=100, height=100, highlightthickness=0)   #全关指示灯
        self.led_close = self.close_cv.create_oval(5, 5, 95, 95, fill='gray')
        self.close_cv.create_text(50, 50, text="全关", fill='black', font=("Helvetica", 20))
        self.open_cv = tk.Canvas(self, width=100, height=100, highlightthickness=0)    #全开指示灯
        self.led_open = self.open_cv.create_oval(5, 5, 95, 95, fill='gray')
        self.open_cv.create_text(50, 50, text="全开", fill='black', font=("Helvetica", 20))
 
        # 创建按钮
        self.close_button = ttk.Button(self, text="关", command=self.cmd_cl) #关命令
        self.stop_button = ttk.Button(self, text="停", command=self.cmd_st)  #停命令
        self.open_button = ttk.Button(self, text="开", command=self.cmd_op)  #开命令
 
        #控件布局
        self.close_cv.grid(row=0, column=0)
        self.open_cv.grid(row=0, column=2)
        self.close_button.grid(row=2, column=0)
        self.stop_button.grid(row=2, column=1)
        self.open_button.grid(row=2, column=2)
 
        # Configure row and column weights for expansion
        self.grid_rowconfigure(1, weight=1)
        self.grid_columnconfigure(0, weight=1)
        self.grid_columnconfigure(1, weight=1)
        self.grid_columnconfigure(2, weight=1)
 
 
    def cmd_st(self):    #停按钮操作
        global is_connected, is_running, is_cmd_st, is_cmd_op, is_cmd_cl
        if not is_connected:
            print("未连接\n")
            return
        if not is_running:
            print("未运行\n")
            return
        is_cmd_op, is_cmd_cl = False, False
        is_cmd_st = not is_cmd_st
        print(f"停按钮{is_cmd_st}\n")
 
    def cmd_op(self):    #开按钮操作
        global is_connected, is_running, is_cmd_st, is_cmd_op, is_cmd_cl
        if not is_connected:
            print("未连接\n")
            return
        if not is_running:
            print("未运行\n")
            return
        is_cmd_st, is_cmd_cl = False, False
        is_cmd_op = not is_cmd_op
        print(f"开按钮{is_cmd_op}\n")
 
    def cmd_cl(self):    #关按钮操作
        global is_connected, is_running, is_cmd_st, is_cmd_op, is_cmd_cl
        if not is_connected:
            print("未连接\n")
            return
        if not is_running:
            print("未运行\n")
            return
        is_cmd_st, is_cmd_op = False, False
        is_cmd_cl = not is_cmd_cl
        print(f"关按钮{is_cmd_cl}\n")
 
 
    def chg_color_op(self, new_color):    #修改开灯颜色
        self.open_cv.itemconfig(self.led_open, fill=new_color)
 
    def chg_color_cl(self, new_color):    #修改关灯颜色
        self.close_cv.itemconfig(self.led_close, fill=new_color)
 
if __name__ == '__main__':
    app = PlcGUI()
    app.mainloop()