Python与Snap7连接西门子PLC：开源数据采集工程实例详解-开发者社区

简介：这是一份面向工业自动化与Python开发初学者的轻量级PLC通信实践资源，聚焦使用Snap7库实现与西门子S7系列PLC的稳定连接、实时数据读取与写入控制，解决工业现场Python侧快速接入PLC的核心技术门槛。压缩包仅含1个Python源文件（.py），体积精简至2KB，代码结构清晰，已内嵌连接参数配置、DB块/内存区读写示例及异常处理逻辑，开箱即用，适合嵌入教学实验、产线数据采集原型或小型SCADA模块开发。资源已获679人学习下载，作者ksthen提供了完整可运行的工程实例，无需额外依赖复杂框架，仅需执行pip install python-snap7即可部署验证，对理解底层S7通信协议、调试PLC交互流程及规避常见连接超时/地址错误等典型问题具有直接参考价值。

1. 项目缘起：为什么选择Snap7与Python来连接PLC？

在工业自动化领域，数据采集是第一步，也是至关重要的一步。无论是监控产线状态、分析设备能耗，还是实现预测性维护，都需要将现场PLC（可编程逻辑控制器）中的数据“搬”到上位机或服务器上。传统的做法可能是用组态软件、OPC服务器，或者厂商提供的专用通讯库，但这些方案往往成本高、灵活性差，或者与现代化的数据分析栈（如Python生态）格格不入。

几年前，我在一个设备远程监控项目中就遇到了这个痛点。客户现场有几十台西门子S7-1200 PLC，需要实时读取设备运行参数并上传到云平台进行分析。如果采用传统的OPC方案，光是授权费用就是一笔不小的开支，而且后续的数据处理流程会变得非常笨重。正是在这种背景下，我发现了Snap7这个开源宝藏，并决定用Python来搭建整个数据采集链路。snap7连接PLCpython工程实例.zip这个压缩包，就是我基于多个实战项目沉淀下来的一个“样板间”工程。它不仅仅是一段代码，更包含了我从环境搭建、通讯测试到稳定运行整个过程中踩过的坑和总结的经验。无论你是想快速验证一个想法，还是为正式项目搭建原型，这个工程都能提供一个清晰的起点。

简单来说，Snap7是一个用C++编写的、跨平台的开源通讯库，它实现了西门子S7系列PLC（200, 300, 400, 1200, 1500等）的以太网通讯协议（ISO-on-TCP，常被称为S7comm）。而Python-snap7则是它的Python封装。这套组合拳的优势非常明显：免费、开源、跨平台（Windows/Linux均可运行）、且能与NumPy, Pandas, Flask, Django等Python生态无缝集成。你可以用几行代码就读到PLC里的一个温度值，也可以用这个框架构建起一个复杂的分布式数据采集系统。

2. 工程全景：`snap7连接PLCpython工程实例.zip`里有什么？

当你解压这个工程包后，看到的不是一个孤零零的脚本，而是一个结构清晰、即拿即用的项目脚手架。这样的设计是为了模拟真实项目的开发环境，避免“一次性脚本”带来的混乱。让我们看看核心目录和文件：

snap7_plc_demo/ ├── requirements.txt # Python依赖包列表 ├── config/ # 配置文件目录 │ ├── plc_config.yaml # PLC连接参数配置文件 │ └── db_layout.json # DB块数据布局定义 ├── src/ # 源代码目录 │ ├── __init__.py │ ├── plc_client.py # 核心通讯客户端类 │ ├── data_models.py # 数据结构定义（如从DB块映射的类） │ └── utils/ # 工具函数 │ ├── logger.py # 日志配置 │ └── helpers.py # 字节解析等辅助函数 ├── examples/ # 使用示例 │ ├── read_single_value.py # 读取单个变量示例 │ ├── read_db_block.py # 读取整个DB块示例 │ ├── write_value.py # 写入数据示例 │ └── async_read.py # 异步读取示例（高级） ├── tests/ # 单元测试（可选） │ └── test_plc_client.py └── main_demo.py # 一个综合演示的主程序

为什么需要这样的结构？在真实的工业场景中，连接参数（IP地址、机架号、槽位）和数据点（DB块编号、偏移量、数据类型）经常会变。如果把它们硬编码在脚本里，每次修改都要翻代码，极易出错。因此，我特意将配置抽离出来，使用YAML和JSON文件进行管理。plc_config.yaml让你可以轻松管理多个PLC的连接信息，而db_layout.json则以一种声明式的方式定义DB块的结构，使得数据读写像操作对象属性一样直观。

例如，plc_config.yaml可能长这样：

plc_rack_1: ip: '192.168.1.100' rack: 0 slot: 1 port: 102 # S7标准端口 timeout_ms: 2000 # 连接超时

db_layout.json则定义了DB1块的数据结构：

{ "db_number": 1, "layout": [ {"name": "motor_speed", "offset": 0, "type": "INT"}, {"name": "temperature", "offset": 2, "type": "REAL"}, {"name": "running", "offset": 6, "type": "BOOL", "bit": 0} ] }

这种配置与代码分离的设计，是工程化思维的体现，也是本项目实例超越简单Demo的关键价值。

3. 从零开始：环境搭建与核心依赖安装详解

万事开头难，环境配置是第一个拦路虎。很多人卡在这一步，不是因为步骤多复杂，而是因为一些细节没注意到。下面我以Windows系统为主（Linux思路类似），手把手带你走通。

3.1 安装Snap7的动态链接库（DLL）这是最核心、也最容易出错的一步。Python-snap7只是一个“外壳”，它底层需要调用用C++编写的Snap7核心库（snap7.dll或libsnap7.so）。

下载库文件：访问Snap7的官方GitHub仓库（https://github.com/stephan182/Snap7）的 Releases 页面。不要下错，你需要的是对应你操作系统的“二进制发布包”，比如Windows下就下载snap7-full-1.4.2-win32.zip（版本号可能更新）。
放置DLL文件：解压下载的zip包，你会看到一堆文件。关键是把snap7.dll（64位系统用snap7.dll，32位系统用snap7.dll，通常下载包里有区分）复制到系统能够找到的位置。我强烈推荐以下两种方法之一：
- 方法A（推荐，一劳永逸）：将snap7.dll复制到Python解释器所在的根目录（例如C:\Python39\）或者你的Python脚本的工作目录。这是最直接的方法。
- 方法B（系统级）：将snap7.dll复制到C:\Windows\System32（64位DLL）或C:\Windows\SysWOW64（32位DLL）。但这种方法可能会因系统权限问题带来麻烦。

注意：很多初学者在这一步会失败，报错信息通常是OSError: [WinError 126] 找不到指定的模块。这几乎100%是因为snap7.dll没有放在正确的位置，或者你的Python环境是32位的却用了64位的DLL（反之亦然）。请务必检查Python位数（在命令行输入python看提示）与DLL位数是否匹配。

3.2 安装Python-snap7包这一步相对简单。打开你的命令行（CMD或PowerShell），激活你的Python虚拟环境（如果你使用的话，强烈建议使用），然后执行：

pip install python-snap7

如果下载慢，可以使用国内镜像源，例如：

pip install python-snap7 -i https://pypi.tuna.tsinghua.edu.cn/simple

3.3 验证安装创建一个简单的Python脚本test_install.py：

import snap7 client = snap7.client.Client() print(f"Snap7 客户端创建成功，版本信息：{snap7.common.get_library_version()}")

运行这个脚本，如果不报错并打印出版本号（如b'1.4.2'），那么恭喜你，环境搭建成功了！如果出现DLL load failed之类的错误，请返回上一步仔细检查DLL的放置位置和位数匹配。

4. 核心通讯类深度解析：`plc_client.py`是如何工作的？

工程中的src/plc_client.py是这个项目的“发动机”。它不是一个简单的函数集合，而是一个封装了连接管理、错误重试、数据解析等功能的类。我们来拆解它的核心方法。

4.1 连接与断开：不只是connect()

import snap7 from snap7.util import get_int, get_real, set_int, set_real import time import logging class PLCClient: def __init__(self, ip, rack=0, slot=1, port=102): self.ip = ip self.rack = rack self.slot = slot self.port = port self.client = snap7.client.Client() self._logger = logging.getLogger(__name__) # 设置连接超时和读取超时（单位：毫秒） self.client.set_connection_timeout(2000) self.client.set_read_timeout(3000) def connect(self, max_retries=3): """带重试机制的连接方法""" for attempt in range(max_retries): try: self.client.connect(self.ip, self.rack, self.slot, self.port) self._logger.info(f"成功连接到PLC {self.ip}:{self.port}") return True except Exception as e: self._logger.warning(f"连接PLC失败 (尝试 {attempt+1}/{max_retries}): {e}") if attempt < max_retries - 1: time.sleep(2) # 等待2秒后重试 self._logger.error(f"无法连接到PLC {self.ip}，已达最大重试次数") return False def disconnect(self): if self.client.get_connected(): self.client.disconnect() self._logger.info(f"已断开与PLC {self.ip} 的连接")

为什么设置超时？工业网络环境复杂，PLC可能因维护、重启或网络波动暂时不可达。设置connection_timeout和read_timeout可以防止程序在连接阶段或读取数据时无限期挂起，提高程序的健壮性。
为什么需要重试机制？网络闪断在工业现场并不罕见。一个简单的重试逻辑，可以自动处理这种瞬时故障，避免人工干预。重试间隔（如2秒）要给PLC和网络足够的恢复时间。

4.2 数据读取：从字节到有意义的数值读取数据是核心操作。Snap7读取回来的是原始的字节流（bytearray），我们需要根据PLC中变量的数据类型（INT, REAL, BOOL等）和偏移量来解析它。

def read_db_real(self, db_number, start_offset, amount=1): """读取一个或多个REAL（浮点数）值""" try: # 计算需要读取的字节长度：一个REAL占4个字节 byte_length = amount * 4 # 从DB块读取原始字节 raw_data = self.client.db_read(db_number, start_offset, byte_length) values = [] for i in range(amount): # 计算当前值在字节流中的起始位置 byte_index = i * 4 # 从字节流中切片出4个字节，并解析为REAL value = get_real(raw_data, byte_index) values.append(value) return values if amount > 1 else values[0] except Exception as e: self._logger.error(f"读取DB{db_number}浮点数失败 (偏移{start_offset}): {e}") raise def read_db_int(self, db_number, start_offset, amount=1): """读取一个或多个INT（整数）值""" try: byte_length = amount * 2 # 一个INT占2个字节 raw_data = self.client.db_read(db_number, start_offset, byte_length) values = [get_int(raw_data, i*2) for i in range(amount)] return values if amount > 1 else values[0] except Exception as e: self._logger.error(f"读取DB{db_number}整数失败 (偏移{start_offset}): {e}") raise def read_db_bool(self, db_number, byte_offset, bit_offset): """读取一个BOOL（布尔）值""" try: # 读取包含目标位的那一个字节 raw_data = self.client.db_read(db_number, byte_offset, 1) # 使用snap7.util的get_bool函数解析指定位 value = snap7.util.get_bool(raw_data, 0, bit_offset) return value except Exception as e: self._logger.error(f"读取DB{db_number}布尔值失败 (字节{byte_offset} 位{bit_offset}): {e}") raise

偏移量（Offset）是关键：这是与PLC编程侧约定好的“地址”。在西门子PLC的DB块中，每个变量都有其固定的偏移地址。例如，DB1.DBW0（一个INT）的偏移量是0，DB1.DBD2（一个REAL）的偏移量是2。你必须从PLC程序员那里获得准确的数据布局表，或者通过TIA Portal等软件查看DB块结构。猜偏移量是行不通的。
字节序问题：西门子PLC使用的字节序（Byte Order）与常见的Intel处理器（小端序）不同，它使用“大端序”（Big-Endian）。snap7.util中的get_int,get_real等函数已经帮我们处理好了这个转换，所以直接使用即可，不必自己手动转换。

4.3 数据写入：将控制指令下发给PLC写入操作是读取的逆过程，需要将Python中的值转换为正确的字节格式，并写入到PLC的指定地址。

def write_db_real(self, db_number, start_offset, value): """写入一个REAL（浮点数）值""" try: # 创建一个4字节的缓冲区 data = bytearray(4) # 将Python的float值设置到缓冲区 set_real(data, 0, value) # 写入到PLC self.client.db_write(db_number, start_offset, data) self._logger.debug(f"已向DB{db_number}.{start_offset}写入REAL值: {value}") except Exception as e: self._logger.error(f"写入DB{db_number}浮点数失败 (偏移{start_offset}): {e}") raise def write_db_bool(self, db_number, byte_offset, bit_offset, value): """写入一个BOOL（布尔）值""" try: # 先读取整个字节，避免影响其他位 raw_data = self.client.db_read(db_number, byte_offset, 1) # 修改指定位的值 snap7.util.set_bool(raw_data, 0, bit_offset, value) # 将修改后的字节写回 self.client.db_write(db_number, byte_offset, raw_data) self._logger.debug(f"已向DB{db_number} (字节{byte_offset} 位{bit_offset})写入BOOL值: {value}") except Exception as e: self._logger.error(f"写入DB{db_number}布尔值失败: {e}") raise

重要提示：写入操作，尤其是写入BOOL值控制设备启停时，务必谨慎！在工业环境中，错误的写入可能导致设备误动作，甚至引发安全事故。务必在测试环境中充分验证，并确保有完善的权限管理和操作确认流程。写入BOOL时先读取再修改，是为了确保不改变同一字节内其他位的状态，这是一种安全的编程习惯。

5. 实战演练：基于配置文件的优雅数据读写

理解了核心类之后，我们来看看工程中更高级的用法：如何利用配置文件，实现声明式的数据读写。这是将项目从“脚本”升级为“工程”的关键。

5.1 加载数据布局配置我们使用db_layout.json来定义DB块。data_models.py中的DBLayout类负责加载和解析这个配置。

import json from typing import List, Dict, Any class DBLayout: def __init__(self, layout_file_path: str): with open(layout_file_path, 'r') as f: config = json.load(f) self.db_number = config['db_number'] self.layout = config['layout'] # 一个字典列表，每个字典定义一个变量 def get_variable_info(self, var_name: str) -> Dict[str, Any]: """根据变量名，获取其在DB块中的地址信息""" for var in self.layout: if var['name'] == var_name: return var raise KeyError(f"变量 '{var_name}' 未在布局文件中定义") def read_all(self, plc_client) -> Dict[str, Any]: """一次性读取此DB布局中定义的所有变量（优化版）""" results = {} # 按数据类型和连续区域分组读取，减少通讯次数 # 这里是一个简化示例，实际可以优化为合并连续地址的读取请求 for var in self.layout: try: if var['type'] == 'REAL': value = plc_client.read_db_real(self.db_number, var['offset']) elif var['type'] == 'INT': value = plc_client.read_db_int(self.db_number, var['offset']) elif var['type'] == 'BOOL': value = plc_client.read_db_bool(self.db_number, var['offset'], var.get('bit', 0)) # ... 可以扩展其他数据类型 else: value = None results[var['name']] = value except Exception as e: print(f"读取变量 {var['name']} 失败: {e}") results[var['name']] = None return results

5.2 在示例中使用看看examples/read_db_block.py是如何变得简洁的：

from src.plc_client import PLCClient from src.data_models import DBLayout # 1. 创建PLC客户端并连接 plc = PLCClient(ip='192.168.1.100') if not plc.connect(): print("连接失败，退出。") exit(1) # 2. 加载DB1的数据布局定义 db1_layout = DBLayout('config/db_layout.json') # 3. 一键读取所有定义的变量 all_data = db1_layout.read_all(plc) print("DB1 数据快照：") for name, value in all_data.items(): print(f" {name}: {value}") # 4. 精准读取单个变量 motor_info = db1_layout.get_variable_info('motor_speed') if motor_info['type'] == 'INT': speed = plc.read_db_int(db1_layout.db_number, motor_info['offset']) print(f"\n电机转速: {speed} RPM") plc.disconnect()

这种方式的好处是巨大的：

可维护性：当PLC数据点变更时，只需修改JSON配置文件，无需改动代码。
可读性：代码中直接使用有意义的变量名（如motor_speed），而不是晦涩的数字偏移量。
减少错误：避免了在代码中硬编码偏移量和数据类型可能带来的错误。

6. 避坑指南与性能优化：来自实战的经验之谈

纸上得来终觉浅，绝知此事要躬行。下面这些坑，都是我或我的同事用真金白银的停机时间换来的教训。

6.1 连接不稳定与断线重连工业网络环境不是实验室的纯净网络。交换机故障、网线松动、PLC重启都可能导致连接中断。一个健壮的客户端必须具备断线检测与自动重连能力。

class RobustPLCClient(PLCClient): def __init__(self, ip, rack=0, slot=1, port=102, ping_interval=30): super().__init__(ip, rack, slot, port) self.ping_interval = ping_interval # 心跳检测间隔（秒） self.last_ping_time = time.time() def _check_connection(self): """检查连接状态，如果断开则尝试重连""" current_time = time.time() if current_time - self.last_ping_time > self.ping_interval: try: # 尝试一个轻量级操作，如读取一个字节，来探测连接 self.client.db_read(1, 0, 1) self.last_ping_time = current_time except: self._logger.error("连接心跳检测失败，尝试重连...") self.client.disconnect() time.sleep(1) self.connect(max_retries=5) def safe_read(self, db_number, offset, size): """带连接检查的安全读取""" self._check_connection() return super().read_db_raw(db_number, offset, size) # 假设父类有这个方法

在你的主循环中，定期调用_check_connection方法，或者在每次读写操作前调用safe_read。这能确保在网络波动后，程序能自动恢复，而不是彻底崩溃。

6.2 读取性能优化：批量读取与异步频繁地读取单个变量（比如在循环中每秒读100个不同的偏移量）会产生大量的小数据包，效率极低，会给PLC和网络带来不必要的负担。

策略一：批量读取：尽可能一次读取一个连续的内存区域，然后在Python端进行分割解析。例如，如果你需要DB1中从偏移0到100的100个字节，不要循环调用100次db_read(..., 1)，而是调用一次db_read(..., 100)。
```
# 低效做法 for i in range(100): value = client.read_db_int(1, i*2) # 高效做法 raw_data = client.client.db_read(1, 0, 200) # 一次读200字节 values = [get_int(raw_data, i*2) for i in range(100)] # 在内存中解析
```
工程中的DBLayout.read_all()方法可以进一步优化，将布局中连续的、同类型的变量合并为一个读取请求，这需要更复杂的地址合并算法，但能极大提升效率。
策略二：异步读取：对于需要同时监控多个PLC，或者读写操作不能阻塞主线程（例如在GUI应用中）的场景，可以考虑异步。Python-snap7本身是同步的，但你可以将其放在单独的线程中，或者使用concurrent.futures的线程池来管理。
```
from concurrent.futures import ThreadPoolExecutor import asyncio def read_plc_data_sync(plc_client, db_info): # 这里是同步的读取操作 return plc_client.read_db_real(db_info['db'], db_info['offset']) async def read_multiple_plcs_async(plc_clients_configs): loop = asyncio.get_event_loop() with ThreadPoolExecutor() as pool: tasks = [] for config in plc_clients_configs: # 将同步函数提交到线程池执行 task = loop.run_in_executor(pool, read_plc_data_sync, config['client'], config['db_info']) tasks.append(task) # 等待所有任务完成 results = await asyncio.gather(*tasks, return_exceptions=True) return results
```
examples/async_read.py提供了一个基础的异步示例。注意，Snap7库本身不是线程安全的，如果你在多线程中共享同一个client对象进行读写，需要自己加锁（threading.Lock）来保证操作串行化，否则可能引发不可预知的问题。更安全的做法是为每个线程创建独立的client实例。

6.3 数据类型与偏移量的“坑”

BOOL的位偏移：BOOL变量在DB块中不单独占用一个字节，而是与其他BOOL共享一个字节。snap7.util.get_bool(data, byte_index, bit_index)中的bit_index范围是0-7。务必确认PLC程序中该BOOL变量在字节中的具体位置。
REAL的精度：Python的float是双精度，而PLC的REAL是单精度。在频繁的读写中，可能会遇到极细微的精度损失，对于高精度控制场合需要留意。通常过程监控数据没有问题。
字符串（STRING）：西门子PLC中的STRING类型有特殊的格式：前两个字节是最大长度和当前长度，后面才是字符数据。读取时需要按照这个格式解析。工程中可以扩展read_db_string方法来处理。
时间与日期：PLC中的时间、日期格式（如S5TIME, DATE_AND_TIME）也需要特殊转换。这些在snap7.util中也有相应的函数（如get_s5time,get_dt），但需要清楚原始数据的格式。

6.4 防火墙与网络配置这是连接失败的常见原因。确保：

PC的防火墙允许Python程序或相关端口（102）的出入站连接。
PLC的IP地址与你的PC在同一网段，且没有IP冲突。
对于S7-1200/1500，需要确保PLC属性中的“连接机制”已勾选“允许来自远程对象的PUT/GET通信访问”。这是安全设置，默认可能关闭。

7. 工程扩展：从数据采集到应用集成

一个稳定的数据采集客户端只是起点。本工程实例为你搭好了架子，你可以轻松地将其集成到更大的系统中。

写入数据库：在main_demo.py的主循环中，读取到数据后，可以很容易地插入到MySQL、PostgreSQL、InfluxDB或TimescaleDB中，用于历史数据存储和趋势分析。

import pandas as pd from sqlalchemy import create_engine # 读取数据 data = db1_layout.read_all(plc) data['timestamp'] = pd.Timestamp.now() # 转换为DataFrame并写入数据库 df = pd.DataFrame([data]) engine = create_engine('postgresql://user:pass@localhost/dbname') df.to_sql('plc_data', engine, if_exists='append', index=False)

提供Web API：使用Flask或FastAPI，将PLC数据以RESTful API的形式暴露出去，供其他系统（如MES、看板系统）调用。

from flask import Flask, jsonify app = Flask(__name__) plc_client = PLCClient('192.168.1.100') plc_client.connect() @app.route('/api/plc/status') def get_status(): data = db1_layout.read_all(plc_client) return jsonify(data)