ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Python读取三菱PLC实战:从MC协议到生产环境稳定采集

Python读取三菱PLC实战:从MC协议到生产环境稳定采集 前阵子接了个活儿产线上几台三菱PLC的数据要汇到MES系统里做设备状态监控和产量统计。客户最初提的方案是全部走Modbus TCP但现场设备已经稳定跑了好几年PLC程序动不了通信模块也没有多余的Modbus映射表最稳妥的办法就是走三菱原生协议从以太网口直接读。我最后选了Python加pymcprotocol这套组合两三天时间就把数据采集端跑通了后面加上数据库写入和看板展示整套系统到今天已经稳定跑了好几个月。这篇文章就把整个思路写清楚从PLC侧需要打开哪些开关到MC协议到底是怎么回事再到能直接复制的读数据脚本以及生产环境里那些文档不会告诉你的坑。适合想把Python用于工业数据采集的朋友尤其是第一次接触三菱PLC上位机开发的程序员和现场设备工程师。1. 开工之前PLC侧的通信开关必须打开很多人拿到这个需求后的第一反应是赶紧写Python代码结果代码写了一大半才发现PLC参数没设置好连接永远超时。所以先说清楚PLC侧要做什么准备。1.1 搞清楚你的PLC到底支持哪些通信方式三菱PLC的家族非常庞大不同系列支持的通信方式差别很大。以我接触比较多的几个系列为例PLC系列串口通信以太网通信常用编程软件FX3U/FX3GRS422/RS485支持MC协议需加FX3U-ENET模块GX Works2FX5URS485支持MC协议内置以太网口直连SLMP/MC协议GX Works3Q系列RS232/485需通信模块QJ71E71以太网模块GX Works2L系列同上内置以太网口GX Works2iQ-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的IPping 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帧网络号一般填0PC号一般填2550xFF表示不指定PC请求目标模块IO编号一般填0x03FF请求目标模块站号填0请求数据长度后面数据的字节数监视定时器超时时间设定命令比如0x1401表示批量读取字软元件0x1402表示批量读取位软元件子命令一般填0起始地址和软元件代码告诉PLC要从哪个地址读读什么类型的寄存器还有一个重要概念是软元件代码。三菱PLC里每种软元件都有一个专属代码D寄存器是0xA8M继电器是0x90X输入是0x9CY输出是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的编号是1003字节小端 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(fPLC返回错误码: 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建议指定plctypefx5u # 如果是Q、L、iQ-R系列用默认参数就行 mc Type3E(plctypefx5u) # 连接PLC mc.connect(192.168.1.10, 6000) # 批量读取D100开始的10个字软元件 values mc.batchread_wordunits(headaddressD100, readsize10) 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的值减去65536def 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(headaddressM100, readsize20) print(M100-M119:, bits)每个元素是True或False对应触点的通断状态。这个方法同样适用于X和Y比如读取X0到X19bits mc.batchread_bitunits(headaddressX0, readsize20)有一点必须注意X和Y的编号是八进制的。也就是说X后面的数字没有8和9X7的下一个是X10X17的下一个是X20。你在写地址时如果写成X8或者X15PLC会返回错误或者读出一个不是你想要的点。这个坑几乎每个刚接触三菱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里看到的地址是X15Python里直接写headaddressX15是没问题的因为pymcprotocol会按八进制解析字符串。但如果你自己写代码做编号换算比如循环生成X0到X20请务必用八进制递增而不是简单的十进制加一。4.3 批量读和随机读怎么选接着聊一个性能相关的话题。有时你需要读的地址在PLC内存里不是连续的比如想读D100、D200、D300三个寄存器的值。最笨的办法是调三次batchread_wordunits每次读一个v1 mc.batchread_wordunits(headaddressD100, readsize1) v2 mc.batchread_wordunits(headaddressD200, readsize1) v3 mc.batchread_wordunits(headaddressD300, readsize1)但这意味着三次网络往返。每一轮通信大概需要几毫秒到几十毫秒在要求高刷新率的场景下这种写法会白白浪费大量时间。pymcprotocol提供了randomread_wordunitsvalues 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(levellogging.INFO) PLC_ADDR 192.168.1.10 PLC_PORT 6000 def collect_once(mc): 采集一轮数据并返回字典 d_values mc.batchread_wordunits(headaddressD100, readsize10) m_bits mc.batchread_bitunits(headaddressM100, readsize8) return { D100_D109: d_values, M100_M107: m_bits, } def run(): mc Type3E(plctypefx5u) 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(plctypeplc_config[type]) mc.connect(plc_config[ip], plc_config[port]) data {} for point in plc_config[points]: data[point] mc.batchread_wordunits(headaddresspoint, readsize1) mc.close() return data with ThreadPoolExecutor(max_workers4) 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协议、选对工具库、把解析规则搞对、再做好容错和监控。这篇把一个完整的拿数据的链路过了一遍希望能帮你少走点弯路。
返回列表