一 背景说明

计划

通过西门子 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块创建-

），这篇文章主要描述上位机中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块创建-

中创建的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通讯-

，其中写的非常详细！）

六 结合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()

【3】效果展示：

七 参考资料

【1】

【PLC+Python】上位机通过snap7实现与西门子PLC通讯并用Tkinter可视化——（1）PLC DB块创建-

【2】

Python使用python-snap7实现西门子PLC通讯-

【3】

Python使用python-snap7实现西门子PLC通讯_python snap7-

【4】

python通过S7协议读取西门子200smart数据_python读取西门子plc数据-

【5】

C++上位软件通过Snap7开源库访问西门子S7-1200/S7-1500数据块的方法_snap7是什么-

【6】

S7 协议调试工具 & 模拟器 --snap7 demo server_partner_client-

【7】

Python + OPCUA + s7-1200 + MySql + Grafana实现工业数据可视化看板开发_grafana中mysql看板-