一、背景与目标在工业物联网场景中，经常需要把 PLC 数据采集到边缘网关，再上报到云端。本文记录一套已实测跑通的全链路方案：数据源：S7-1200（CPU1214 DC/DC/DC），DB1 中两个Array[0..999] of INT采集端：IoT2050 边缘网关，Python + python-snap7本地归档：SQLite，每秒 1000 行，文件上限 700MB，FIFO 自动清理上报：MQTT 发布 JSON（1Hz），订阅回环验证二、测试拓扑设备IP说明CPU1214(G1)192.168.x.12PLC，数据源IoT2050192.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