1. 为什么选择PythonSMLP协议连接三菱FX5U在工业自动化领域数据采集一直是核心需求。传统的做法往往需要依赖厂商提供的专用软件比如三菱的GX Works3。这类软件功能强大但体积臃肿学习成本高而且很难与其他系统集成。我遇到过不少项目客户只需要简单读取PLC的某些寄存器数据却被迫安装整套软件环境这就像为了喝杯牛奶要养头奶牛一样不划算。Python的socket编程配合SMLP协议Seamless Message Protocol就能完美解决这个问题。SMLP是三菱PLC的底层通信协议通过TCP/IP直接与PLC对话。实测下来这种方案有三大优势轻量化代码不到100行就能实现基础读写部署环境仅需Python标准库跨平台Windows/Linux都能运行甚至树莓派这类嵌入式设备也能胜任响应快绕开中间件直接通信实测读取100个寄存器仅需8ms去年我给某包装生产线做的质量追溯系统就用了这个方案。产线上有12台FX5U用Python脚本直接采集生产参数数据通过MQTT推送到云端。从开发到上线只用了3天如果走传统OPC路线至少需要两周。2. 环境搭建与PLC配置2.1 基础环境准备虽然我们不用GX Works3开发但初始配置还是需要它。这里分享个偷懒技巧去三菱官网下载GX Works3 Mini版本就行只有完整版1/10的大小足够完成基础配置。安装时注意关闭杀毒软件否则可能卡在驱动安装环节。PLC硬件连接有个细节容易踩坑FX5U的网口不支持自动翻转Auto-MDIX必须用直通网线连接。有次我在客户现场调试用交叉线折腾半小时都没反应换成直通线秒连。建议包里常备两种网线或者带个便携交换机。2.2 关键参数配置在GX Works3中配置以太网端口时这几个参数必须记牢参数项推荐值注意事项IP地址192.168.3.250不要用.1结尾的地址子网掩码255.255.255.0保持默认即可通信方式ASCII 16进制二进制模式调试更困难SLMP端口2000避免使用502等保留端口特别提醒配置完成后一定要点击反映设置并关闭然后重新上电PLC。我有次忘记断电新配置死活不生效白白浪费两小时查问题。3. SMLP协议报文解析3.1 协议帧结构详解SMLP协议的核心在于报文构造。以读取D寄存器为例完整请求报文如下# 读取D100开始的10个寄存器 request b\x50\x00 # 副头部 request b\x00 # 网络编号 request b\xFF # PLC编号 request b\xFF\x03 # 请求目标模块IO编号 request b\x00\x0E # 请求目标模块站号 request b\x00 # 请求数据长度(后续自动计算) request b\x0A\x00 # 监控定时器(10秒) request b\x01\x04 # 指令代码(批量读取) request b\x00\x00 # 子指令代码 request b\xA8\x01 # 起始地址D100(0xA8168,168800968D100) request b\x0A\x00 # 读取点数(10个)各字段含义用快递类比就很好理解副头部相当于快递单号网络/PLC编号是收件人楼层和房间号监控定时器是超时提醒指令代码说明你要寄的是文件还是包裹地址转换有个坑D寄存器实际地址显示地址800。比如D100对应968(0x3C8)但协议里要填168(0xA8)3.2 响应报文处理成功响应时第9字节开始是有效数据。但要注意三菱用的是小端序需要做字节转换import struct # 假设收到response_data字节流 values [] for i in range(0, len(response_data[9:]), 2): word response_data[9i:11i] # 将2字节数据转为16位无符号整数 value struct.unpack(H, word)[0] values.append(value)处理错误响应时更要小心。有次我收到错误代码E2查手册才发现是地址越界。后来养成了习惯每次都要先检查响应报文第7字节if response[7] ! 0: error_code response[8] raise Exception(fSMLP错误代码:{hex(error_code)})4. Python实战代码精讲4.1 连接管理类实现我习惯封装一个PLCClient类核心方法如下class PLCClient: def __init__(self, ip, port2000, timeout5): self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(timeout) self.ip ip self.port port def connect(self): try: self.sock.connect((self.ip, self.port)) return True except Exception as e: print(f连接失败: {str(e)}) return False def _send_receive(self, request): self.sock.sendall(request) response self.sock.recv(1024) if len(response) 11: raise Exception(响应报文过短) return response def read_d_registers(self, start, count): # 构造读取D寄存器的请求报文 request self._build_read_request(start, count) response self._send_receive(request) return self._parse_read_response(response) def __del__(self): self.sock.close()实际使用中发现TCP长连接容易因网络波动中断。后来我增加了自动重连机制def safe_execute(self, func, *args, max_retry3): for i in range(max_retry): try: return func(*args) except (socket.error, socket.timeout): self.connect() raise Exception(操作失败超过最大重试次数)4.2 批量读写优化当需要连续读取多个地址段时直接循环调用效率很低。我的优化方案是合并相邻地址为单个请求使用线程池并行读取非连续区块from concurrent.futures import ThreadPoolExecutor def batch_read(self, address_ranges): def read_range(start, count): return {fD{starti}: val for i, val in enumerate(self.read_d_registers(start, count))} with ThreadPoolExecutor() as executor: futures [] for start, end in address_ranges: futures.append(executor.submit( read_range, start, end-start1)) results {} for future in futures: results.update(future.result()) return results实测读取1000个分散寄存器串行需要1200ms优化后仅需280ms。这在需要高频采集的场景非常关键。5. 常见问题排查指南5.1 连接类问题症状connect()超时检查网线是否插在FX5U的以太网口不是USB口确认电脑IP与PLC同网段如PLC是192.168.3.250电脑可设192.168.3.100关闭电脑防火墙临时测试用ping 192.168.3.250测试基础连通性症状连接成功但收不到响应检查GX Works3中是否启用了SLMP协议确认端口号与程序设置一致默认2000在PLC参数中查看开放设置是否允许来自所有设备的连接5.2 数据异常处理数值错乱确认ASCII/二进制模式设置一致检查寄存器地址计算是否正确D寄存器要800用Wireshark抓包对比请求与响应随机断连增加心跳机制每30秒发送空指令设置socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)考虑在应用层实现断线重连队列去年遇到最棘手的案例是某工厂电磁干扰导致报文CRC错误。后来在交换机端口加了磁环同时代码里增加了校验和重试机制才解决。所以工业现场部署时物理环境因素也要纳入考虑。