代码收藏家 Ubuntu22.04与西门子PLC S7-1200通过Python-snap7通讯，获取PLC数据并存入数据库mysql
由于需要远程抓取数据并存入数据库，考虑用Ubuntu系统+Python-snap7实现，通过Ubuntu22.04与西门子1200PLC连接通讯，读取PLC的数据，并存入数据库。查找资料，记录实施过程。
本教程采用python-snap7实现，ubuntu22.04默认自带python3。其中，python-snap7为python专用版本（其他版本详见snap7官网）。
snap7官网：Snap7 Homepagehttps://snap7.sourceforge.net/
python-snap7官网：Welcome to python-snap7’s documentation! — python-snap7 0.0rc0 documentationhttps://python-snap7.readthedocs.io/en/latest/ 安装snap7需要安装动态库和python3-snap7软件，其中snap7动态库需要在自己的系统环境中编译生成，从官网下载的snap7是.7z格式的，需要下载后解压。python3-snap7则直接通过pip安装即可。以下是实施过程。
1. 安装7z解压工具和pip3
（1）安装7z
sudo apt update
sudo apt upgrade
sudo apt install p7zip-full安装过程如下：
（2）安装pip3
sudo apt-get install python3-pip安装过程：
输入Y，继续
2. 下载snap7
去snap7官网下载最新的snap7，目前是snap7-full-1.4.2，如下：
为了方便下载，我们采用wget方式获取，指令如下：
wget "https://jaist.dl.sourceforge.net/project/snap7/1.4.2/snap7-full-1.4.2.7z"如下：
下载完，默认是7z文件，因此需要7z解压，指令如下：
7z x snap7-full-1.4.2.7z -r -o./snap7解释一下指令：
7z 解压缩文件命令
snap7-full-1.4.2.7z 表示解压缩文件名
-r 表示递归解压缩所有的子文件夹
-o 指定解压到的那个目录 注意：-o 后面没用空格
./snap7 要解压到的路径
如下：
解压完毕。
3. 编译snap7动态库
进入snap7文件夹，通过make编译生成.so文件，再复制到lib库中。
（1）编译
首先查找自己系统是什么架构，一般开机都有，如下：
我的是aarch64架构，ARM v8, 根据官网的说法，编译动态库采用：x86_64_linux.mk，其他请自行确认。
指令如下：
cd /root/snap7/snap7-full-1.4.2/build/unix
sudo make -f x86_64_linux.mk all执行如下：
可以在相关文件夹看到已经生成了.so文件，
复制libsnap7.so到lib库，指令如下：
sudo cp /root/snap7/snap7-full-1.4.2/build/bin/x86_64-linux/libsnap7.so /usr/lib/libsnap7.so
sudo cp /root/snap7/snap7-full-1.4.2/build/bin/x86_64-linux/libsnap7.so /usr/local/lib/libsnap7.so复制完毕后，执行：
sudo ldconfigldconfig是一个动态链接库管理命令，其目的为了让动态链接库为系统所共享。ldconfig通常在系统启动时运行，而当用户安装了一个新的动态链接库时，就需要手工运行这个命令。
执行如下：
4. 安装python3-snap7
指令如下：
sudo pip3 install python-snap7执行如下：
至此，snap7在Ubuntu中的安装结束，下面进行测试通讯。
5. 连接测试
执行如下：
(1)进入python3。
python3
(2)连接PLC，执行命令：
import snap7
client = snap7.client.Client()
client.connect('192.168.10.20', 0, 1)
print(f"连接状态：{client.get_connected()}")执行如下：
连接正常。
说明：connect('192.168.10.20', 0, 1)：格式如下
connect('ip', rack, slot)，其中rack是机架号，slot卡槽号，具体可参见TIA组态配置，如下图：
其中，机架号为0，卡槽为1。另外，需要将有线网卡设置成与PLC同一网段，ping的通才行。
(3)关闭连接。执行命令如下：
client.disconnect()
print(f"连接状态：{client.get_connected()}")执行如下：
连接已关闭。按Ctrl+Z退出即可。
到此通讯正常。
6. 安装数据库
请参照之前的文章，安装数据库，创建名为db_data数据库，密码自己设定，创建以下数据表DB1：
CREATE TABLE DB1 (
ID INT AUTO_INCREMENT PRIMARY KEY,
BL1 tinyint UNSIGNED NOT NULL,
BL2 tinyint UNSIGNED NOT NULL,
BL3 tinyint UNSIGNED NOT NULL,
BL4 tinyint UNSIGNED NOT NULL,
BE1 tinyint UNSIGNED NOT NULL,
BE2 tinyint UNSIGNED NOT NULL,
IN1 smallint NOT NULL,
DIN1 int NOT NULL,
RL1 float NOT NULL,
STR1 varchar(266) NOT NULL,
WD1 smallint UNSIGNED NOT NULL,
Date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);7. 执行读写PLC数据
测试代码testSnap7.py，如下：
import pymysql # 导入pymysql库，用于连接和操作MySQL数据库
import snap7 # 导入snap7库，用于和西门子S7系列PLC通信
from snap7 import util # 从snap7库中导入util模块，提供处理PLC数据的辅助函数
import time # 导入time库，用于获取时间相关信息
# 获取当前时间的函数，格式化为年-月-日 时:分:秒
def gettime():
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
# 从PLC读取数据的函数
def read_data_from_plc(plc):
bufferRead = bytearray(300) # 创建一个长度为300字节的字节数组，用于存放读取的数据
bufferRead = plc.db_read(2, 0, 300) # 从PLC的DB2块，起始地址0，读取300字节数据
data = {
"RBOOL1": util.get_bool(bufferRead, 0, 0), # 获取第0个字节的第0位布尔值
"RBOOL2": util.get_bool(bufferRead, 0, 1), # 获取第0个字节的第1位布尔值
"RBOOL3": util.get_bool(bufferRead, 0, 2), # 获取第0个字节的第2位布尔值
"RBOOL4": util.get_bool(bufferRead, 0, 3), # 获取第0个字节的第3位布尔值
"RBYTE1": util.get_byte(bufferRead, 1), # 获取第1个字节的值
"RBYTE2": util.get_byte(bufferRead, 2), # 获取第2个字节的值
"RINT1": util.get_int(bufferRead, 4), # 获取从第4个字节开始的整数
"RDINT1": util.get_dint(bufferRead, 6), # 获取从第6个字节开始的双整数
"RREAL1": util.get_real(bufferRead, 10), # 获取从第10个字节开始的浮点数
"RSTRING1": util.get_string(bufferRead, 14), # 获取从第14个字节开始的字符串
"RWORD1": util.get_word(bufferRead, 270) # 获取从第270个字节开始的字
}
return data
# 向PLC写入数据的函数
def write_data_to_plc(plc, data):
bufferWrite = bytearray(300) # 创建一个长度为300字节的字节数组，用于存放要写入的数据
# 将多个布尔值组合放入第0个字节
bufferWrite[0] = (int(data["WBOOL1"]) << 0) | (int(data["WBOOL2"]) << 1) | (int(data["WBOOL3"]) << 2) | (
int(data["WBOOL4"]) << 3)
bufferWrite[1] = data["WBYTE1"] # 设置第1个字节的值
bufferWrite[2] = data["WBYTE2"] # 设置第2个字节的值
util.set_int(bufferWrite, 4, data["WINT1"]) # 在第4个字节开始设置整数
util.set_dint(bufferWrite, 6, data["WDINT1"]) # 在第6个字节开始设置双整数
util.set_real(bufferWrite, 10, data["WREAL1"]) # 在第10个字节开始设置浮点数
util.set_string(bufferWrite, 14, data["WSTRING1"], 254) # 在第14个字节开始设置字符串
util.set_word(bufferWrite, 270, data["WWORD1"]) # 在第270个字节开始设置字
plc.db_write(2, 300, bufferWrite) # 将数据写入PLC的DB2块，起始地址300，写入300字节数据
if __name__ == "__main__":
bufferRead = bytearray(300) # 预先定义读取数据的缓冲区
bufferWrite = bytearray(300) # 预先定义写入数据的缓冲区
PLC1 = snap7.client.Client() # 创建一个snap7客户端实例，用于连接PLC
try:
PLC1.connect('192.168.10.20', 0, 1) # 连接到指定IP地址的PLC，rack号为0，slot号为1
read_result = read_data_from_plc(PLC1) # 调用函数从PLC读取数据
print("读取结果：")
print(
f"BOOL1: {read_result['RBOOL1']};\nBOOL2: {read_result['RBOOL2']};\nBOOL3: {read_result['RBOOL3']};\nBOOL4: {read_result['RBOOL4']};\nBYTE1: {read_result['RBYTE1']};\nBYTE2: {read_result['RBYTE2']};\nINT1: {read_result['RINT1']};\nDINT1: {read_result['RDINT1']};\nREAL1: {'%.2f' % read_result['RREAL1']};\nSTRING1: {read_result['RSTRING1']};\nWORD1: {'%#x' % read_result['RWORD1']}")
db = pymysql.connect(host="localhost", user="root", password="数据库密码，自己设定的", database="db_data", charset='utf8mb4')
# 连接到本地MySQL数据库，指定用户名、密码、数据库名和字符集
cursor = db.cursor() # 创建数据库游标，用于执行SQL语句
try:
sql = "INSERT INTO DB1 (BL1, BL2, BL3, BL4, BE1, BE2, IN1, DIN1, RL1, STR1, WD1, Date) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
# 定义插入数据的SQL语句，有12个占位符
values = (
read_result["RBOOL1"], read_result["RBOOL2"], read_result["RBOOL3"], read_result["RBOOL4"],
read_result["RBYTE1"], read_result["RBYTE2"], read_result["RINT1"], read_result["RDINT1"],
read_result["RREAL1"], read_result["RSTRING1"], read_result["RWORD1"], gettime())
# 准备要插入的数据，包含从PLC读取的数据和当前时间
cursor.execute(sql, values) # 执行SQL插入语句
db.commit() # 提交数据库事务，使插入操作生效
print("数据库保存成功！")
except pymysql.Error as e:
print(f"数据库储存失败！原因: {e}")
finally:
cursor.close() # 关闭数据库游标
db.close() # 关闭数据库连接
write_data = {
"WBOOL1": True,
"WBOOL2": False,
"WBOOL3": False,
"WBOOL4": True,
"WBYTE1": 64,
"WBYTE2": 128,
"WINT1": 9999,
"WDINT1": 999999,
"WREAL1": 40.05,
"WSTRING1": "WO SHI SHUI?",
"WWORD1": 0x1000
}
write_data_to_plc(PLC1, write_data) # 将准备好的数据写入PLC
except Exception as e:
print(f"连接PLC出错: {e}")
finally:
PLC1.disconnect() # 无论是否出错，最终断开与PLC的连接注释写的很清楚，通讯连接、读取PLC数据、写入PLC数据、读取的PLC数据入库都在里面，理解不了的请咨询AI，AI时代学习也变的很轻松。
执行如下：
数据库数据如下：
S7-1200 PLC DB1地址定义如下：
write是PLC写入到ubuntu, read是PLC接收ubuntu的写入数据，测试运行正常。
作者：侠客狄仁猫
