当前位置:首页 > 工业控制 > 工业控制
[导读]车间多台PLC做数据采集,不一定要上SCADA网关。Python装好协议库后,通过以太网直连Modbus TCP、S7、Ethernet/IP,或用串口直连Modbus RTU,把寄存器、DB块、标签统一成点表,定时批量轮询、本地缓存、统一上送MES/时序库,边缘盒子即可替代硬件网关。


车间多台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和时序库。



本站声明: 本文章由作者或相关机构授权发布,目的在于传递更多信息,并不代表本站赞同其观点,本站亦不保证或承诺内容真实性等。需要转载请联系该专栏作者,如若文章内容侵犯您的权益,请及时联系本站删除( 邮箱:macysun@21ic.com )。
关闭