Python玩转工业数据采集：手把手教你用python-snap7和PLCSIM Advanced搭建仿真测试环境
·
Python玩转工业数据采集：手把手教你用python-snap7和PLCSIM Advanced搭建仿真测试环境
在工业自动化领域，PLC（可编程逻辑控制器）作为核心控制设备，其数据采集与通讯一直是工程师必备技能。但对于初学者或没有硬件条件的开发者来说，如何低成本地学习和测试PLC通讯成为一大难题。本文将带你从零开始，利用西门子PLCSIM Advanced仿真软件和python-snap7库，构建一个完整的本地仿真开发环境，实现各类工业数据的读写测试。
1. 环境准备与基础配置
1.1 软件工具清单
构建PLC仿真测试环境需要以下核心组件：
- TIA Portal V16/V17 ：西门子全集成自动化工程平台
- PLCSIM Advanced V3.0+ ：高级PLC仿真软件
- Python 3.8+ ：推荐使用Anaconda发行版
- python-snap7 1.0+ ：Python与S7系列PLC通讯库
注意：PLCSIM Advanced需要单独授权，教育用户可通过学术许可获取
1.2 网络配置关键步骤
仿真环境网络配置是成功通讯的前提：
- 在Windows网络连接中创建虚拟适配器
- 为PLCSIM Advanced实例分配静态IP（如192.168.10.1）
- 配置Python所在主机的IP与仿真PLC同网段
- 关闭防火墙或添加出入站规则
# 检查网络连通性示例（Windows命令提示符）
ping 192.168.10.1
2. 仿真PLC项目搭建
2.1 TIA Portal基础配置
在TIA Portal中创建新项目时，需特别注意：
- PLC型号选择S7-1200/S7-1500系列
- 添加DB数据块并设置优化块访问为"False"
- 定义测试变量时包含多种数据类型：
| 数据类型 | 变量名 | 地址 | 初始值 |
|---|---|---|---|
| Bool | Motor_Status | DB1.DBX0.0 | False |
| Int | Temperature | DB1.DBW2 | 25 |
| Real | Pressure | DB1.DBD4 | 0.0 |
| String | Device_Name | DB1.DBB8 | "SIM_001" |
2.2 PLCSIM Advanced部署技巧
启动仿真PLC时的实用参数：
# 仿真PLC启动参数示例
{
"PLC_Type": "S7-1500",
"IP_Address": "192.168.10.1",
"Subnet_Mask": "255.255.255.0",
"Cycle_Time": 100,
"Project_File": "C:\\Projects\\PLC_Sim\\sim_project.sim"
}
常见问题排查：
- 若连接超时，检查虚拟网卡是否启用
- 出现"Access Denied"错误时确认TIA Portal已关闭
- 数据读写异常时验证DB块属性设置
3. Python通讯核心实现
3.1 python-snap7安装与初始化
推荐使用虚拟环境安装：
python -m venv plc_env
source plc_env/bin/activate # Linux/macOS
plc_env\Scripts\activate # Windows
pip install python-snap7
基础连接代码框架：
import snap7
from snap7.util import *
def connect_plc(ip, rack=0, slot=1):
client = snap7.client.Client()
try:
client.connect(ip, rack, slot)
if client.get_connected():
print(f"成功连接到PLC @ {ip}")
return client
except Exception as e:
print(f"连接失败: {str(e)}")
return None
3.2 多数据类型读写方案
针对不同数据类型的最佳实践：
Bool类型处理：
def read_bool(client, db_number, byte_offset, bit_offset):
data = client.db_read(db_number, byte_offset, 1)
return get_bool(data, 0, bit_offset)
def write_bool(client, db_number, byte_offset, bit_offset, value):
data = bytearray(1)
set_bool(data, 0, bit_offset, value)
client.db_write(db_number, byte_offset, data)
结构化数据读取模板：
def read_structured_data(client, db_number, layout):
"""
layout示例:
[
('running', 'bool', 0, 0),
('temp', 'int', 2),
('pressure', 'real', 4)
]
"""
results = {}
for item in layout:
name, dtype, offset, *bits = item
length = {
'bool': 1, 'int': 2,
'real': 4, 'string': 256
}.get(dtype, 1)
data = client.db_read(db_number, offset, length)
if dtype == 'bool':
results[name] = get_bool(data, 0, bits[0])
elif dtype == 'int':
results[name] = get_int(data, 0)
elif dtype == 'real':
results[name] = get_real(data, 0)
elif dtype == 'string':
results[name] = get_string(data, 0, 256)
return results
4. 高级应用与性能优化
4.1 批量读写效率提升
对于高频数据采集场景，建议：
- 使用
db_get/db_set替代多次单点读写 - 合理设置轮询间隔（通常100-500ms）
- 采用异步IO模式（结合asyncio）
async def async_plc_monitor(client, db_spec, interval=0.2):
import asyncio
while True:
start = time.time()
data = await asyncio.to_thread(client.db_read, *db_spec)
# 数据处理逻辑
elapsed = time.time() - start
await asyncio.sleep(max(0, interval - elapsed))
4.2 异常处理与重连机制
工业环境通讯必须考虑稳定性：
class PLCConnection:
def __init__(self, ip, rack=0, slot=1):
self.ip = ip
self.rack = rack
self.slot = slot
self.client = snap7.client.Client()
self._connect()
def _connect(self, retries=3):
for i in range(retries):
try:
self.client.connect(self.ip, self.rack, self.slot)
if self.client.get_connected():
return True
except Exception:
if i == retries - 1:
raise
time.sleep(2**i)
return False
def reconnect(self):
self.client.destroy()
self.client = snap7.client.Client()
return self._connect()
4.3 数据持久化方案
将采集数据存储到数据库的完整示例：
import sqlite3
from contextlib import closing
def init_db(db_path='plc_data.db'):
with closing(sqlite3.connect(db_path)) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS process_data (
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
name TEXT,
value REAL,
unit TEXT
)
""")
conn.commit()
def save_reading(db_path, readings):
with closing(sqlite3.connect(db_path)) as conn:
conn.executemany(
"INSERT INTO process_data(name, value, unit) VALUES (?,?,?)",
[(k, v, '°C' if 'temp' in k else 'bar' if 'press' in k else '')
for k, v in readings.items()]
)
conn.commit()
5. 典型工业场景应用案例
5.1 生产线监控系统模拟
构建完整的虚拟产线监控：
-
在TIA Portal中创建包含以下元素的DB块：
- 设备状态字（位数组）
- 生产计数（DINT）
- 良品率（REAL）
- 故障代码（INT）
-
Python端实现状态监测看板：
def monitor_dashboard(client):
plt.ion()
fig, axes = plt.subplots(2, 2, figsize=(10,8))
while True:
data = client.db_read(1, 0, 20)
# 更新各子图表
update_plot(axes[0,0], get_int(data, 0), "生产速度")
update_plot(axes[0,1], get_real(data, 4), "温度曲线")
update_plot(axes[1,0], get_int(data, 8), "故障代码")
update_bits(axes[1,1], data[12:14], "设备状态")
plt.pause(0.5)
5.2 配方参数管理系统
实现配方数据的读写管理：
class RecipeManager:
def __init__(self, client, db_num):
self.client = client
self.db_num = db_num
self.recipe_size = 1024 # 配方数据区大小
def load_recipe(self, recipe_id):
offset = recipe_id * self.recipe_size
data = self.client.db_read(self.db_num, offset, self.recipe_size)
return {
'name': get_string(data, 0, 32),
'temp': get_real(data, 32),
'time': get_int(data, 36),
'steps': [get_int(data, 40+i*2) for i in range(10)]
}
def save_recipe(self, recipe_id, params):
data = bytearray(self.recipe_size)
set_string(data, 0, params['name'], 32)
set_real(data, 32, params['temp'])
set_int(data, 36, params['time'])
for i, step in enumerate(params['steps'][:10]):
set_int(data, 40+i*2, step)
offset = recipe_id * self.recipe_size
self.client.db_write(self.db_num, offset, data)
在实际项目中，这套仿真环境已经帮助多个团队在没有物理PLC的情况下，提前完成了80%的通讯逻辑开发和测试工作。特别是在处理复杂数据结构时，先通过仿真验证数据解析逻辑，可以大幅减少现场调试时间。
更多推荐
5
0- 0
已为社区贡献27条内容
所有评论(0)