一、背景与目标
在工业物联网场景中，经常需要把 PLC 数据采集到边缘网关，再上报到云端。本文记录一套已实测跑通的全链路方案：
-
数据源：S7-1200（CPU1214 DC/DC/DC），DB1 中两个 Array[0..999] of INT
-
采集端：IoT2050 边缘网关，Python + python-snap7
-
本地归档：SQLite，每秒 1000 行，文件上限 700MB，FIFO 自动清理
-
上报：MQTT 发布 JSON（1Hz），订阅回环验证
二、测试拓扑
| CPU1214(G1) | 192.168.x.12 | PLC，数据源 |
| IoT2050 | 192.168.x.29 | 边缘网关，Python 采集 + MQTT 上报 |
| Win11 (Docker Mosquitto) | 192.168.x.176 | 本地测试 MQTT Broker，匿名访问 |
三、设备组态要点
CPU1214：
Device configuration 中激活 S7 PUT/GET
创建非仅优化寻址的 DB1，含两个 Array[0..999] of INT，每个 Array[0] 每秒自加一
IoT2050：
-
Python venv：/opt/plc_poller/s71200
-
依赖：python-snap7、paho-mqtt
四、采集脚本（plc_mqtt_loop.py）
#!/opt/plc_poller/s71200/bin/python
\”\”\”
plc_mqtt_loop.py – IoT2050 上 S7-1200 → MQTT 1Hz 测试脚本
功能:
1. 每秒从 CPU1214 DC/DC/DC 的 DB1.DBW0 起读取 1000 个 INT (2000 字节)
– 超过 S7-1200 PDU 上限 240B，由 Snap7 底层自动分包（实测已确认）
2. 本地归档：1000 行/秒写入 SQLite（带时间戳），文件上限 700MB，FIFO 清理
3. 打包成 JSON 发布到 MQTT broker
4. 订阅同一 topic，回环打印（只打印第 1 个 INT）
部署:
scp plc_mqtt_loop.py root@192.168.x.29:/opt/plc_poller/s71200/
ssh root@192.168.x.29 \”/opt/plc_poller/s71200/bin/python /opt/plc_poller/s71200/plc_mqtt_loop.py\”
依赖 (已在 venv 中):
pip install python-snap7 paho-mqtt
\”\”\”
import os
import json
import time
import signal
import sqlite3
import logging
import snap7
from snap7.util import get_int
import paho.mqtt.client as mqtt
# ———- 配置 ———-
PLC_IP = \”192.168.x.12\” # CPU1214 IP
DB_NUM = 1 # DB1
START_B = 0 # DBW0
INT_N = 1000 # 连续读 1000 个 INT (= 2000 字节)
PERIOD_S = 1.0 # 1Hz 采样
MQTT_HOST = \”192.168.x.176\” # Win11 上的 docker mosquitto
MQTT_PORT = 1883
TOPIC = \”Node-red\”
# ———- SQLite 归档配置 ———-
DB_PATH = \”/opt/plc_poller/s71200/plc_data.db\”
MAX_DB_MB = 700 # 归档文件上限 700MB
ROW_BYTES = 60 &
