前阵子接了个活儿,产线上几台三菱PLC的数据要汇到MES系统里做设备状态监控和产量统计。客户最初提的方案是全部走Modbus TCP,但现场设备已经稳定跑了好几年,PLC程序动不了,通信模块也没有多余的Modbus映射表,最稳妥的办法就是走三菱原生协议从以太网口直接读。我最后选了Python加pymcprotocol这套组合,两三天时间就把数据采集端跑通了,后面加上数据库写入和看板展示,整套系统到今天已经稳定跑了好几个月。
这篇文章就把整个思路写清楚:从PLC侧需要打开哪些开关,到MC协议到底是怎么回事,再到能直接复制的读数据脚本,以及生产环境里那些文档不会告诉你的坑。适合想把Python用于工业数据采集的朋友,尤其是第一次接触三菱PLC上位机开发的程序员和现场设备工程师。
1. 开工之前:PLC侧的通信开关必须打开
很多人拿到这个需求后的第一反应是赶紧写Python代码,结果代码写了一大半才发现PLC参数没设置好,连接永远超时。所以先说清楚PLC侧要做什么准备。
1.1 搞清楚你的PLC到底支持哪些通信方式
三菱PLC的家族非常庞大,不同系列支持的通信方式差别很大。以我接触比较多的几个系列为例:
| PLC系列 | 串口通信 | 以太网通信 | 常用编程软件 |
|---|---|---|---|
| FX3U/FX3G | RS422/RS485,支持MC协议 | 需加FX3U-ENET模块 | GX Works2 |
| FX5U | RS485,支持MC协议 | 内置以太网口,直连SLMP/MC协议 | GX Works3 |
| Q系列 | RS232/485,需通信模块 | QJ71E71以太网模块 | GX Works2 |
| L系列 | 同上 | 内置以太网口 | GX Works2 |
| iQ-R系列 | 同上 | 内置以太网口,性能最强 | GX Works3 |
这里最关键的一个概念是:三菱PLC原生支持的是MC协议(MELSEC Communication Protocol),不是Modbus。虽然很多新型号也内置了Modbus TCP服务器功能,但那是另一套映射,需要PLC程序里专门配置。如果你的PLC程序不能动,那就老老实实用MC协议读。
1.2 PLC参数和防火墙里容易被忽略的设置
用GX Works打开PLC参数,进入“以太网端口设置”或者“模块参数”,有几个地方必须确认:
- IP地址要和你的电脑在同一网段,比如PLC设192.168.1.10,电脑设192.168.1.50,子网掩码统一255.255.255.0。
- 通信协议要选择“TCP”,不要选成“UDP”或者“无协议”,虽然UDP也能通,但TCP在数据完整性上更省心,pymcprotocol默认就是TCP。
- 开放端口默认是6000,这是三菱SLMP/MC协议的默认端口。有些设备工程师习惯改成别的端口,那你代码里就要对应改。
- 内置以太网口的PLC一般有个“SLMP通信”或者“MC协议”的使能开关,务必确认在打开状态。
还有一点容易被忽略的是:如果PLC和电脑中间隔着公司局域网,交换机的端口隔离、防火墙策略都可能导致连接失败。我在现场遇到过PC端防火墙全关了、IP也能ping通,但TCP 6000端口就是连不上,最后查出来是交换机做了端口隔离,只放行了常用端口。如果遇到这种情况,别急着怀疑代码,先问一下网络管理员或现场IT。
1.3 先别写代码,用ping确认链路通不通
在动手写任何Python代码之前,先在命令行里ping一下PLC的IP:
ping 192.168.1.10能ping通,说明物理链路是通的。再做一个更进一步的测试,用Windows自带的telnet测一下TCP 6000端口是不是通的:
telnet 192.168.1.10 6000如果telnet能连上并且不报错,说明PLC的通信端口已经对外开放。如果telnet提示连接失败,那即便Python代码写得再对也白搭。这个检查步骤花不了两分钟,能帮你省掉一整天的排查时间。实测下来,绝大多数“代码连不上PLC”的问题,其实在这两步就能发现根源。
2. 说人话理解MC协议:读D区到底在网络上发了什么
前面反复提到MC协议,很多刚接触工控的Python程序员可能没听过。其实它没那么神秘,说白了就是三菱定义的一套“PLC和上位机之间说话的方式”。你要读哪个地址、读多少个数、用什么格式返回,这些都有固定约定。
2.1 三菱的通信协议家族
MC协议按物理链路区分主要有两种:串口版的叫MC协议,走的是RS232/RS422/RS485;以太网版的在新型号中常被称为SLMP协议,但报文格式基本沿用了MC协议的3E帧。SLMP是Seamless Message Protocol的缩写,意思是“无缝消息协议”,本质上是一套通用以太网通信协议,三菱的PLC、运动控制器、变频器、伺服都可以用这个协议去访问。
报文格式主要分两类:3E帧和4E帧。3E帧是最常见的,主要用于访问PLC的软元件数据(D寄存器、M继电器、X输入等);4E帧多用于访问智能功能模块或带帧号的通信。我们读PLC内部数据,用3E帧就足够了。pymcprotocol里的Type3E类就是对应的实现。
3E帧的报文结构大致是这样的:
- 子帧头:固定为0xD000,用来标识这是3E帧
- 网络号:一般填0
- PC号:一般填255(0xFF),表示不指定PC
- 请求目标模块IO编号:一般填0x03FF
- 请求目标模块站号:填0
- 请求数据长度:后面数据的字节数
- 监视定时器:超时时间设定
- 命令:比如0x1401表示批量读取字软元件,0x1402表示批量读取位软元件
- 子命令:一般填0
- 起始地址和软元件代码:告诉PLC要从哪个地址读,读什么类型的寄存器
还有一个重要概念是软元件代码。三菱PLC里每种软元件都有一个专属代码,D寄存器是0xA8,M继电器是0x90,X输入是0x9C,Y输出是0x9D,计数器C的触点属于位软元件也走0x90这类代码。这个代码在手动拼报文时会用到。
2.2 手拼一个3E帧也能读数据
了解协议最直接的方式是亲手拼一帧报文。我用Python的socket库实现了一个最简版读取D寄存器的函数,代码不长,但是能把3E帧的来龙去脉看得很清楚:
import socket import struct def build_read_d_request(start, count): """构造批量读取D寄存器的3E帧请求""" # 子帧头,3E帧固定标记 sub_header = b"\xD0\x00\x00\x00" # 网络号、PC号、IO编号、站号 basic_info = bytes([0x00, 0xFF, 0x03, 0xFF, 0x00]) # 监视定时器,0x10对应4096毫秒(需要结合实际) timer = b"\x10\x00" # 命令:0x1401 = 批量读取字软元件,子命令0x0000 cmd = b"\x01\x14\x00\x00" # 起始地址,D100的编号是100,3字节小端 addr = start.to_bytes(3, "little") # 软元件代码,0xA8 = D寄存器 device_code = bytes([0xA8]) # 读取点数 point = count.to_bytes(2, "little") data = timer + cmd + addr + device_code + point length = len(data).to_bytes(2, "little") return sub_header + basic_info + length + data def read_d_words(ip, port, start, count): req = build_read_d_request(start, count) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(3) sock.connect((ip, port)) sock.sendall(req) recv = sock.recv(2048) sock.close() # 响应结构:9字节固定头 + 2字节结束代码 + 数据 end_code = struct.unpack("<H", recv[9:11])[0] if end_code != 0: raise RuntimeError(f"PLC返回错误码: 0x{end_code:04X}") # 剩余数据就是读到的字软元件值 payload = recv[11:] return struct.unpack(f"<{count}H", payload) if __name__ == "__main__": values = read_d_words("192.168.1.10", 6000, 100, 4) print(values)这段代码可以把D100到D103四个寄存器的原始值读回来。为什么起始地址是100而不是"100"这个字符串?因为MC协议里所有软元件地址最终都要折算成数字编号,D100的编号就是100。如果你读的是X输入,那地址编码就要按八进制规则换算。
说实话,这个手写版本在生产环境里用起来还是太粗糙,比如没有处理TCP粘包、没有重试机制、只实现了批量读这一种命令。但理解它之后,你再去看pymcprotocol的源码,会非常容易跟上它的逻辑。
2.3 为什么最终选pymcprotocol
三菱官方其实提供了一套Windows组件叫MX Component,可以通过COM方式供Python调用,功能也很全。但我的体会是:MX Component首先需要安装三菱的授权软件,体积大、授权麻烦,而且Python通过COM读写数据的性能损耗明显;其次它绑死了Windows环境,如果采集端打算跑在Linux服务器上就完全没法用。
pymcprotocol是开源社区做的纯Python实现,底层就是socket直接发MC协议报文。它支持Q系列、L系列、iQ-R系列、FX5U等多种三菱PLC,且只依赖标准库,不需要额外编译扩展,跨平台能力强。我们把数据采集服务丢在一台Ubuntu服务器上,一台机器同时采集多台PLC,稳定性和性能都完全够用。
3. 环境搭建和第一个能跑通的读取脚本
如果说PLC侧准备是打通物理链路,那这一节就是真正把数据从PLC里拿出来的过程。
3.1 Python环境准备的三个注意点
Python本身没什么特别的,要求3.8以上就行。我测试过的版本包括3.10、3.11和3.12,均正常。安装完成后直接:
pip install pymcprotocol这里想多说几句安装时经常遇到的问题。第一,Windows上如果提示python命令找不到,大概率是安装时没有勾选“Add Python to PATH”,重装时勾上即可;第二,如果你在公司内网环境,pip从官方源容易超时,可以临时加清华镜像源:
pip install pymcprotocol -i https://pypi.tuna.tsinghua.edu.cn/simple第三,装完之后在Python交互命令行里执行import pymcprotocol不报错,就说明库已经正常进入了当前环境。确认这一步,能避免后面写了一大堆代码才发现库没装进虚拟环境的尴尬。
3.2 第一个脚本:连接并批量读取D寄存器
下面这个脚本是真正能在现场跑起来的完整示例,连的是FX5U系列,IP地址根据实际修改即可:
from pymcprotocol import Type3E # 如果是FX5U,建议指定plctype="fx5u" # 如果是Q、L、iQ-R系列,用默认参数就行 mc = Type3E(plctype="fx5u") # 连接PLC mc.connect("192.168.1.10", 6000) # 批量读取D100开始的10个字软元件 values = mc.batchread_wordunits(headaddress="D100", readsize=10) print("D100-D109 原始值:", values) # 读取PLC型号,确认连接对象正确 print("CPU型号:", mc.get_cputype()) # 断开连接 mc.close()这个脚本跑通之后,你已经完成了90%的工作。读取返回的是一个list,长度和readsize一致,每个元素是0到65535之间的整数,对应PLC里D寄存器的16位二进制值。
连接过程我要提醒一点:connect的默认超时时间可能比较短,如果你所在的网络有轻微延迟,连接很容易超时。可以用socket.setdefaulttimeout在连接前设置一个更宽松的超时:
import socket socket.setdefaulttimeout(10)这个设置对后续的读写操作同样生效。实测在比较差的工业网络环境里,设置10秒超时比默认值稳很多,不会动不动就抛连接超时异常。
3.3 16位、32位、浮点数的换算
读出来的原始值只是16位整数,但实际项目中D寄存器里存的数据往往没那么简单。PLC程序里可能用两个连续的D寄存器拼成一个32位整数,也可能用两个D寄存器存一个32位浮点数,还有可能用4个D寄存器存64位浮点或字符串。
先说最简单的16位无符号整数,直接就是读到的值。如果PLC程序里用的是16位有符号整数(比如温度的负值),需要把超过32767的值减去65536:
def to_signed16(val): if val > 0x7FFF: return val - 0x10000 return val32位整数在三菱PLC里的存储顺序是:低位字在前,高位字在后。也就是说D100存低16位,D101存高16位。从原始值换算成32位有符号整数的方法:
def to_signed32(low, high): value = (high << 16) | low if value >= 0x80000000: value -= 0x100000000 return value浮点数的换算稍微绕一点,本质上是把两个16位寄存器重新拼成4字节,再按IEEE 754解释。三菱PLC存储浮点数时,也是低位字在前:
import struct def to_float32_from_d(d0, d1): """把两个D寄存器值解析为32位浮点数""" low = d0 & 0xFFFF high = d1 & 0xFFFF packed = (high << 16) | low return struct.unpack("<f", struct.pack("<I", packed))[0]实测读取三菱PLC里用浮点指令(DMOV指令)写入的值,用这个函数解析结果和触摸屏上显示完全一致。反过来,如果你要往PLC里写一个浮点数,流程正好倒过来:先把float的4字节拆成两个16位整数,然后写入连续的两个D寄存器。
def float32_to_ds(val): """把浮点数拆成两个D寄存器值""" packed = struct.unpack("<I", struct.pack("<f", val))[0] low = packed & 0xFFFF high = (packed >> 16) & 0xFFFF return low, high这里最怕的是字节序搞反。三菱PLC和西门子PLC在很多数据存储方式上正好相反,西门子的两个字拼接时的顺序和三菱不一样,所以不能拿着西门子的解析函数直接套。
4. 位元件读写与地址换算:来自八进制的问候
D寄存器是字数软元件(16位为一个单位),但工业现场还有大量开关量信号,比如设备运行状态、故障信号、按钮按没按下,这些在PLC里是用M继电器、X输入、Y输出这类位软元件来存的。读取这类数据,用batchread_bitunits。
4.1 位软元件的读取
以M继电器为例,读取M100到M119共20个点的状态:
bits = mc.batchread_bitunits(headaddress="M100", readsize=20) print("M100-M119:", bits)每个元素是True或False,对应触点的通断状态。这个方法同样适用于X和Y,比如读取X0到X19:
bits = mc.batchread_bitunits(headaddress="X0", readsize=20)有一点必须注意:X和Y的编号是八进制的。也就是说X后面的数字没有8和9,X7的下一个是X10,X17的下一个是X20。你在写地址时如果写成"X8"或者"X15",PLC会返回错误或者读出一个不是你想要的点。这个坑几乎每个刚接触三菱PLC的人都会踩。
4.2 八进制地址的换算逻辑
为了加深印象,举个例子。如果PLC端X输入端子接了一个限位开关,图纸上标的是X15,你在Python里读取时必须写"X15"吗?不一定。要看图纸上用的是十进制习惯还是三菱的八进制习惯。三菱真正的X15端子编号,按十进制数算其实是第13个输入点(X0是第0个,X1是第1个,一直到X7是第7个,X10是第8个,X11是第9个,X12是第10个,X13是第11个,X14是第12个,X15是第13个)。
所以如果你在触摸屏或者GX Works里看到的地址是X15,Python里直接写headaddress="X15"是没问题的,因为pymcprotocol会按八进制解析字符串。但如果你自己写代码做编号换算,比如循环生成"X0"到"X20",请务必用八进制递增,而不是简单的十进制加一。
4.3 批量读和随机读怎么选
接着聊一个性能相关的话题。有时你需要读的地址在PLC内存里不是连续的,比如想读D100、D200、D300三个寄存器的值。最笨的办法是调三次batchread_wordunits,每次读一个:
v1 = mc.batchread_wordunits(headaddress="D100", readsize=1) v2 = mc.batchread_wordunits(headaddress="D200", readsize=1) v3 = mc.batchread_wordunits(headaddress="D300", readsize=1)但这意味着三次网络往返。每一轮通信大概需要几毫秒到几十毫秒,在要求高刷新率的场景下,这种写法会白白浪费大量时间。pymcprotocol提供了randomread_wordunits:
values = mc.randomread_wordunits( headaddress=["D100", "D200", "D300"], readsize=[1, 1, 1] )这个函数会拼成一个随机读取请求帧,把不同的地址和点数一起发给PLC,一次往返就能取回所有数据。同样的道理也适用于位元件:
bits = mc.randomread_bitunits( headaddress=["M100", "X0", "Y10"], readsize=[10, 5, 2] )实际项目里,如果采集的点位比较散,我会先把要读的地址和尺寸整理成列表,然后一次性发出请求,能大幅提升采集效率和稳定性。
5. 从“能读数据”到“稳定采集”:生产环境踩坑记录
脚本跑通了,不代表项目完成了。真正把采集服务放到生产环境里跑,会遇到一连串让人抓狂的问题,这里把我踩过的坑按排查顺序整理出来。
5.1 连不上PLC时,按这个顺序排查
连接失败是最常见的问题,也是新手最容易焦虑的时候。我建议按从物理层到应用层的顺序排查:
- 先用
ping确认PLC IP通了没有。如果不通,查网线、交换机、IP地址是否在同一网段。 - 用
telnet IP 6000确认端口开放。如果端口不通,回PLC侧看“SLMP通信”或“MC协议”开关,同时确认PLC参数下载后有没有重启PLC使配置生效。 - 如果以上都正常但Python连不上,检查电脑防火墙是否拦截了TCP 6000的出站连接。Windows防火墙在私网网络下通常不拦出站,但如果你装了第三方安全软件,可能会拦。
- 检查PLC是否有连接数限制。部分型号的以太网模块对同时连接数有限制,如果已有触摸屏或其他上位机占着连接,新的连接会被拒绝。这种情况可以把触摸屏那侧改成一直在线监视,或者换一个空闲的端口。
最后这条在现场很隐蔽。一次客户报障说采集SQL空数据,我远程排查了半天,最后发现是现场的触摸屏软件占用了PLC的SLMP连接,PLC的以太网模块允许的最大连接数是4,而我们这侧排在第五个,连接被拒绝。后来把触摸屏停掉或者改走自己的通道,问题立刻消失。
5.2 读写超时和断线重连机制
工业网络环境不会像办公室网络那么稳定,偶尔一次的干扰、交换机重启、PLC固件升级,都可能导致TCP连接断开。如果采集脚本不做任何容错,断一次线就退出,那生产线上就会丢数据,这是绝对不允许的。
我的做法是把采集逻辑包在一个带重连的循环里:
import time from pymcprotocol import Type3E import logging logging.basicConfig(level=logging.INFO) PLC_ADDR = "192.168.1.10" PLC_PORT = 6000 def collect_once(mc): """采集一轮数据并返回字典""" d_values = mc.batchread_wordunits(headaddress="D100", readsize=10) m_bits = mc.batchread_bitunits(headaddress="M100", readsize=8) return { "D100_D109": d_values, "M100_M107": m_bits, } def run(): mc = Type3E(plctype="fx5u") while True: try: if not mc.is_connected(): mc.connect(PLC_ADDR, PLC_PORT) data = collect_once(mc) # 这里把data写入数据库或者推送到消息队列 time.sleep(0.5) except Exception as e: logging.error(f"采集异常: {e}") try: mc.close() except Exception: pass time.sleep(3) if __name__ == "__main__": run()重点在于两点:一是每次采集前判断连接状态,连接断开时自动重连;二是出现异常时关闭旧连接,等几秒再重试。这样即便是PLC断电重启,服务也能在恢复后自动续上。
5.3 采集频率与寄存器错位的取舍
采集频率不是越高越好。有人觉得刷新越快越实时,一上来就用100ms的间隔去轮询PLC,结果发现网络开销大,CPU占用高,而且PLC本身的响应处理能力有限,反而可能导致时序延迟。
以三菱FX5U为例,一次批量读10个字的请求通常在几十毫秒内完成。但要考虑远端还有触摸屏、其他上位机同时在通信,PLC的通信处理是分时轮询的。如果每一路都高频读写,最终效果可能互相干扰。我在实际项目中一般把采集周期设置在200到500毫秒,足以满足看板刷新和MES数据统计的需求。如果确实需要毫秒级的数据同步,建议别用轮询方式,而是让PLC在数据变化时主动往上位机发送消息,但那套方案复杂度会上一个台阶,后续再单独写文章聊。
5.4 内存地址错位的防护
另一个容易出问题的点是PLC程序版本变更后,D寄存器的意义可能变了。原先D100是设备温度,升级程序后可能变成了产量计数或者被新逻辑占用。这种问题代码上查不出来,必须靠数据校验来兜底。
我在采集程序里会加一个“心跳字”的检查。让PLC工程师在固定的D0位置每100ms自增一个32位整数,采集端每次读到后跟上次对比,如果两次完全相同说明PLC程序停了或者地址映射出了问题;如果值跳变异常,说明可能存在通信错乱。这样即使PLC程序改了,也能第一时间发现数据链路异常。
6. 多设备扩展与标签通信的思考
前面聊的核心都是单台PLC的读取,但生产现场往往不止一台设备。把方案从“读一台PLC”扩展到“读多台PLC”时,有几个思路可以立刻用上。
最笨的办法是开多个线程,每个线程持有一个Type3E实例,各连各的PLC。这个方法简单直接,缺点是线程多了之后管理困难,而且每台设备的采集逻辑如果不一样,代码会越来越臃肿。另一种做法是用异步IO,把网络读写放到事件循环里。但pymcprotocol内部是同步socket,要改成异步得自己包一层线程池,复杂度偏高。
我的选择是用一个简单的调度器,把每台PLC的连接信息和采集点位配置放在一个列表里,用一个线程池并发采集:
from concurrent.futures import ThreadPoolExecutor PLC_LIST = [ {"ip": "192.168.1.10", "port": 6000, "type": "fx5u", "points": ["D100", "D200"]}, {"ip": "192.168.1.11", "port": 6000, "type": "q", "points": ["D100", "D101"]}, ] def fetch_plc(plc_config): mc = Type3E(plctype=plc_config["type"]) mc.connect(plc_config["ip"], plc_config["port"]) data = {} for point in plc_config["points"]: data[point] = mc.batchread_wordunits(headaddress=point, readsize=1) mc.close() return data with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(fetch_plc, PLC_LIST))这个方案稳定、代码量少,适配大多数中小规模的采集场景。如果PLC数量上百台,则要考虑单机瓶颈,可能需要中间加采集网关或者分布式采集器,但那是另一个话题了。
再说说标签通信。三菱比较新的PLC(比如iQ-R和FX5U)支持标签方式访问数据,也就是说通过命名标签而不是硬编码地址来读写。这样可以避免PLC程序修改后地址漂移的问题,更贴近软件工程的做法。pymcprotocol目前只支持软元件地址方式,不支持标签方式。如果你需要用标签读取,要么用三菱官方MX Component,要么通过GX Works3把标签导出后做一层地址映射表,自己维护转换关系。后者的灵活度其实更高,表格放在配置文件里,PLC程序改动时只改映射表,采集代码不用动。
7. 最后补充几点个人经验
写到这里,核心内容其实已经说完了。再分享几个零散但很实用的经验。
第一,采集程序上线前,一定要跟PLC工程师确认好D寄存器里每一条数据的格式。同一个地址,PLC里存的是整数、浮点数还是BCD码,解析方式完全不同。我在项目里吃过一次亏,对方说是整数温度值,结果读出来翻了好几倍,后来查资料才发现PLC里用的是BCD编码,温度90度在寄存器里显示的是0x0090对应的十进制144,而不是直接90。换算方式是把十六进制的每一位按十进制还原。
第二,工业环境里不建议用无线网络连接PLC。现场有电机、变频器这些大功率设备,无线信号受干扰严重,TCP重传率高,采集数据延迟和丢包都很难受。能走有线就走有线,网线用带屏蔽层的工业网线,接头做好接地处理。
第三,不要把采集服务和数据库放在同一台性能很低的机器上。采集程序本身不重,但如果你同时跑着数据库、Web服务器和看板渲染,CPU和内存很容易吃紧。生产上最好至少分开两台机器,采集端只负责收集和推送,数据库和展示端放另外一台服务器。
第四,时刻留意日志。给采集程序加上滚动日志,每天一个日志文件,记录每条采集任务的耗时、成功失败次数。上线初期多看一眼日志,能发现很多隐藏问题。一次我们发现某台PLC平均采集耗时从20ms涨到了200ms,排查之后发现是该PLC的以太网模块固件版本太旧,在数据量大时处理不过来,升级固件后恢复正常。这种事如果没有日志,很难定位。
Python读取三菱PLC数据,说到底就是理解MC协议、选对工具库、把解析规则搞对、再做好容错和监控。这篇把一个完整的拿数据的链路过了一遍,希望能帮你少走点弯路。