PLC与Python直连通信:无需额外网关即可批量采集车间设备数据
车间多台PLC做数据采集,不一定要上SCADA网关。Python装好协议库后,通过以太网直连Modbus TCP、S7、Ethernet/IP,或用串口直连Modbus RTU,把寄存器、DB块、标签统一成点表,定时批量轮询、本地缓存、统一上送MES/时序库,边缘盒子即可替代硬件网关。
一、Modbus TCP批量读保持寄存器
多数施耐德、台达、三菱、国产PLC都支持Modbus TCP,502端口。与其逐点读,不如按连续地址一次读整段,再用BinaryPayloadDecoder处理32位浮点、DINT的字节序。
from pymodbus.client import ModbusTcpClient
from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian
def poll_modbus(ip, slave, start, count, interval=1.0):
cli = ModbusTcpClient(ip, port=502, timeout=3)
if not cli.connect():
raise RuntimeError("PLC TCP connect fail")
try:
while True:
rr = cli.read_holding_registers(address=start, count=count, slave=slave)
if rr.isError():
print("read error", rr)
else:
dec = BinaryPayloadDecoder.fromRegisters(
rr.registers, byteorder=Endian.Big, wordorder=Endian.Little)
temp = dec.decode_32bit_float()
total = dec.decode_32bit_uint()
print({"temp": temp, "total": total})
import time
time.sleep(interval)
finally:
cli.close()
连续地址一次读40~100个寄存器最划算;模拟量1~2秒一轮,状态量0.5秒一轮,报警量可缩到200ms。RTU设备可通过工业串口服务器转TCP,Python仍用上面这套TCP客户端,免得服务器直接折腾ttyUSB0。
二、西门子S7-1200/1500用Snap7直读DB
S7走ISO-on-TCP 102端口,python-snap7直连。TIA里CPU保护勾选“Permit access with PUT/GET”,DB取消“优化的块访问”,rack=0、slot=1;优化块按绝对偏移读会全0。
import snap7
from snap7.util import get_real, get_dint, get_int
class S7Batch:
def __init__(self, ip, rack=0, slot=1):
self.c = snap7.client.Client()
self.c.connect(ip, rack, slot)
self.c.set_connection_params(3000, 3000) # 连接/读写超时
def read_db(self, db, start, size):
return self.c.db_read(db, start, size)
def parse(self, raw):
return {
"temp": get_real(raw, 0),
"pressure": get_int(raw, 4),
"total": get_dint(raw, 6)
}
def close(self):
self.c.disconnect()
把同DB的温度、压力、产量、状态字集中摆放,一次db_read取整段,比逐变量读减少大量往返。S7-1200并行S7连接数有限,单Python进程单连接轮询多DB最稳,别开很多线程各建连接。
三、统一点表与批量调度
不同协议抽象成同一点模型,采集线程只看点表,不关心底层:
POINTS = [
{"dev":"A","proto":"modbus","ip":"192.168.1.10","slave":1,"addr":100,"len":6,"cycle":1},
{"dev":"B","proto":"s7","ip":"192.168.1.20","db":1,"start":0,"size":16,"cycle":1},
]
def dispatch(pt):
if pt["proto"]=="modbus":
# 调poll_modbus单次读,按addr/len解析
...
else:
s7=S7Batch(pt["ip"])
raw=s7.read_db(pt["db"],pt["start"],pt["size"])
return s7.parse(raw)
点表放YAML/JSON,新增工位只改配置;工程量、单位、报警上下限也随点表走,上层统计、报表、MQTT转发都不用改代码。
四、断线重连与本地缓存
现场最怕网络闪断。重连用指数退避,避免疯重连把PLC连接资源打满;断网期间数据写SQLite或本地队列,恢复后按时间戳补传。
import time
def safe_loop(job, max_delay=60):
delay=1
while True:
try:
job()
delay=1
except Exception as e:
print("comm fail", e)
time.sleep(delay)
delay=min(max_delay, delay*2)
Modbus可定时读一个固定版本/心跳寄存器判活;S7可周期性get_cpu_state或读固定状态字,连续失败再重连。写库用批量提交,别每条点写一次;上云用QoS1 MQTT或HTTP批量POST,弱网降级为本地文件。
五、落地注意
权限与安全:PLC侧只开采集所需DB/寄存器区间,Python用只读账号;跨网段走VLAN,公网不直暴露502/102。
字节序:Modbus 32位浮点、DINT必须按厂家word/byte顺序试解码,Snap7用get_real/get_dint,别自己硬解。
性能:普通PLCS7轮询200~500ms、Modbus 500ms读几十寄存器较稳;要求100ms以内改PLC主动上报,别让Python硬轮。
资源:边缘盒用虚拟环境锁依赖,pymodbus、python-snap7、pandas分开;长期运行加logging,不把异常吞掉。
多协议扩展:罗克韦尔用pycomm3读标签,OMRON用FINS/Modbus,Ethernet/IP走CIP,统一进POINTS调度即可。
按“协议适配器+点表配置+单连接批量+退避重连+本地补传”做,单台ARM盒子直连十几台PLC、千级点位做车间采集完全可行,不买硬件网关也能把数据稳定送MES和时序库。





