从0到1：EC942边缘计算机用Python实现Modbus TCP采集+MQTT上云全记录（附踩坑实录）

前言最近在搞一个工业物联网的小项目用映翰通 EC942 边缘计算机采集现场设备的数据再通过 MQTT 把数据传到云端。EC942 自带了 DeviceSupervisor AgentDSA配置化点几下鼠标就能采集上云但我这次想要更灵活的协议处理和数据加工能力所以选择了自己写采集程序这条路——用 Python 实现一个 Modbus TCP 主站采集完再用paho-mqtt发布到云端 Broker。手头没有真实 PLC 怎么办用Modbus Slave 模拟器在 PC 上模拟一个 Modbus TCP 从站通过网线直连 EC942先把整套流程跑通之后再接真实设备。今天上午从环境搭建到最终跑通中间踩了好几个坑每一个都挺有代表性整理出来希望能帮后面遇到同样问题的人少走弯路。先上一张整体架构图看清楚今天到底在打通哪几段链路实线是采集上云的主链路虚线是后面加的云端反向控制链路。下面按今天实际操作的顺序一步步还原整个过程。环境准备EC942 跑的是 Debian 10 系统默认 ETH2 网口地址是192.168.4.100/24。我把 PC 网卡配成同网段的静态 IP192.168.4.11网线直连 EC942 的 ETH2 口——现代网卡都支持 Auto-MDIX直通线交叉线都能用不需要交换机。PC 上打开 Modbus Slave 模拟器切到 **Modbus TCP Server从站/服务端**模式监听默认的 502 端口然后在保持寄存器40001~40010里填了一组测试值1,2,3,4,5,6,7,77,8,9。中间故意插了个77当特征值后面验证解析对不对的时候特别好用——一眼就能看出来有没有对上。这里有个坑提前说一下Modbus 协议里的寄存器地址是从0开始的但模拟器界面显示的是4xxxx这种传统显示地址两者相差1。也就是说界面上的40001对应协议报文里的addr040008也就是我们的特征值77对应addr7。这个差1的换算贯穿了后面写代码的整个过程一定要记住。踩坑①ping通了端口却连不上万事俱备先验证一下网络通不通。PC 上ping 192.168.4.100没问题那反过来在 EC942 上验证到 PC 的连通性总该没问题吧结果一顿骚操作nc -zv 192.168.4.11 502bash: nc: command not found换 telnetbash: telnet: command not found那装个 mbpoll 总行了吧sudo apt install mbpollE: Unable to locate package mbpoll三连败。原因其实很简单EC942 跑的是精简版 Debian 镜像压根没预装这些网络调试工具而apt install报找不到包是因为Debian 10 (buster) 已经 EOL生命周期结束默认的软件源早就下线了装什么都得先把/etc/apt/sources.list改成指向archive.debian.org才行。这里还有个概念要理清楚ping通只能证明网络三层ICMP是通的完全不能说明 502 端口有没有程序在监听。ping 和端口连通性根本是两回事别被ping都通了怎么还连不上这种直觉误导。工具不好使不代表没法测。EC942 上现成有 Python3标准库自带的socket模块就能干这个活不用装任何东西python3 -c import socket try: socket.create_connection((192.168.4.11, 502), timeout3) print(端口可达) except OSError as e: print(连接失败:, e) 这一招后面帮了大忙记下来备用。动手写 Modbus TCP 采集代码工具的坑先放一放开始写正经代码。Modbus TCP 相比 RTU串口版本少了 CRC16 校验TCP 自身保证可靠传输但多了一个MBAP 头事务标识符(2字节) 协议标识符(2字节,恒为0) 长度(2字节) 单元标识符(1字节) PDU(功能码数据)核心是一个TcpMaster类内部维护长连接、断线自动重连class TcpMaster: def __init__(self, host, port502, timeout2.0): self.host host self.port port self.timeout timeout self._sock None self._txn_id 0 self._lock threading.Lock() def _transact(self, unit_id: int, pdu: bytes) - bytes: with self._lock: try: self._ensure_connected() self._txn_id (self._txn_id 1) 0xFFFF mbap struct.pack(HHHB, self._txn_id, 0, len(pdu) 1, unit_id) self._sock.sendall(mbap pdu) header self._recv_exact(7) txn_id, _proto_id, length, _unit_id struct.unpack( HHHB, header) if txn_id ! self._txn_id: raise ModbusError(f事务号不匹配) body self._recv_exact(length - 1) except (OSError, socket.timeout) as exc: self.close() # 出错即断开下次自动重连 raise ModbusError(fTCP 通信异常: {exc}) from exc if body[0] 0x80: raise ModbusError(f从站异常码 {body[1] if len(body) 1 else 未知}) return body def read_registers(self, unit_id, fc, addr, qty): fc3 保持寄存器, fc4 输入寄存器 pdu struct.pack(BHH, fc, addr, qty) body self._transact(unit_id, pdu) count body[1] return list(struct.unpack(f{count // 2}H, body[2:2 count]))点表配置化不把地址硬编码在代码里对应上面提到的40001→addr0换算规则points: [ {name: reg_40001, unit_id: 1, fc: 3, addr: 0, qty: 1, dtype: uint16}, {name: reg_40008, unit_id: 1, fc: 3, addr: 7, qty: 1, dtype: uint16} ]主循环里读点表、拼 JSON、发 MQTT逻辑很直白def poll_once(master, points): values, errors {}, {} for pt in points: try: regs master.read_registers(pt[unit_id], pt[fc], pt[addr], pt[qty]) values[pt[name]] round(decode(regs, pt[dtype], pt.get(scale, 1.0)), 4) except ModbusError as exc: errors[pt[name]] str(exc) return values, errors代码写完了装好依赖准备跑起来。踩坑②paho-mqtt和Python 3.7八字不合pip install paho-mqtt python3 main.py config.json一运行就崩File .../paho/mqtt/client.py, line 49, in module from typing import Literal ImportError: cannot import name Literal from typing During handling of the above exception, another exception occurred: ... ModuleNotFoundError: No module named typing_extensions排查下来发现EC942 的 Debian 10 自带Python 3.7而pip install paho-mqtt默认装的是最新的2.1.0。paho-mqtt 从 2.x 开始用到了typing.Literal这是 Python 3.8 才有的特性Python 3.7 下会尝试从typing_extensions这个补充包里兜底导入但环境里压根没装这个包两条路都走不通直接报错崩溃。而且就算装上typing_extensions解决了这个报错2.x 版本还改了mqtt.Client()的构造函数签名要求显式传入callback_api_version参数不传就直接报错——相当于换个坑接着踩。所以最省事的办法是直接把版本锁定到兼容 Python 3.7 的 1.6.1API 跟老代码完全匹配不用改一行业务代码pip uninstall paho-mqtt -y pip install paho-mqtt1.6.1这个坑的教训是装依赖库不能无脑装最新版尤其是在这种系统 EOL、Python 版本偏老的嵌入式设备上遇到诡异的 ImportError 先看看是不是库和运行环境的版本对不上。踩坑③一个数字引发的连接失败依赖装好了重新运行这次导入没问题了但是WARNING 采集点 reg_40001 失败: TCP 通信异常: [Errno 111] Connection refused WARNING 采集点 reg_40002 失败: TCP 通信异常: [Errno 111] Connection refused ...十个点位全部失败每5秒循环一次Connection refusedErrno 111这个报错的含义很明确TCP 握手包已经送到了目标主机但对方在那个端口上没有程序监听直接给拒绝了。这跟网络通不通是两码事——网络层没问题问题出在包送去的地方不对或者对方没在听。排查思路很简单先怀疑配置再怀疑对端检查config.json里tcp.host填的什么 —— 一看填的是192.168.4.1而 PC 实际的 IP 是192.168.4.11少打了最后一个1。改成正确的192.168.4.11之后重新跑。一个数字的差别让程序执着地往一个根本不存在或者存在但没监听 502 端口的地址上死磕报错信息其实已经把线索摆得明明白白只是当时没往这个方向想。这提醒我一个通用排查顺序遇到 Connection refused先把配置文件里的 IP/端口原样打印出来跟实际环境核对一遍比怀疑代码逻辑更快定位问题。曙光初现采集成功改完 IP重新运行INFO 上报 {reg_40001: 1.0, reg_40002: 2.0, reg_40003: 3.0, reg_40004: 4.0, reg_40005: 5.0, reg_40006: 6.0, reg_40007: 7.0, reg_40008: 77.0, reg_40009: 8.0, reg_40010: 9.0} rc0和模拟器里填的1,2,3,4,5,6,7,77,8,9完全对上尤其是reg_40008: 77.0这个特征值精确匹配——这就是前面特意埋一个跳出规律的值的用处一眼就能确认地址映射和字节解析全部正确不用逐个数值去比对。数值显示成77.0而不是77是正常的不是bug解码函数里scale默认值是浮点数1.0整数乘以浮点数结果自然变成浮点型不影响数据本身的正确性。踩坑④MQTT订阅看不到数据Modbus 这条链路验证完了接下来验证数据是不是真的传到了云端 MQTT Broker。用 MQTT.fx 连上 broker订阅框里填了设备的 topic 前缀factory/line1/ec942-0001点了 Subscribe0 msg/s No content in table设备端日志明明显示rc0发布调用成功订阅端却啥也收不到第一反应是不是broker连错了、账号密码不对。仔细一看订阅框里填的主题是factory/line1/ec942-0001而程序实际发布的两条消息在它的子主题下factory/line1/ec942-0001/status factory/line1/ec942-0001/telemetryMQTT 的主题匹配是精确匹配订阅父级主题不会自动收到子主题的消息除非用通配符。把订阅改成factory/line1/ec942-0001/##是多级通配符能匹配这个前缀下所有层级的子主题。改完立刻就收到了telemetry和status带online: true的保留消息两条数据跟设备端日志的内容完全一致。这个坑顺带纠正了我对rc0的一个误解client.publish()返回的rc0只表示消息成功放进了本地发送队列不代表消息已经真正送达远端 Broker。要确认真正送达必须站在接收方的角度去订阅验证不能只看发送方日志没报错就认为万事大吉。进阶让云端也能反向控制设备采集上云跑通之后又加了一个功能云端通过 MQTT 下发指令反向控制 Modbus 从站的寄存器/线圈。这意味着程序要从单纯的采集→上报升级成双向通信订阅一个命令主题收到指令后用 Modbus 的写功能码下发再回一条执行结果的 ack 消息。先给TcpMaster加上写方法FC05写线圈、FC06写单寄存器、FC16写多寄存器复用已有的加锁_transact读写操作天然线程安全def write_register(self, unit_id, addr, value): fc6 写单个保持寄存器 pdu struct.pack(BHH, 6, addr, value 0xFFFF) self._transact(unit_id, pdu)然后是命令处理回调收到{point: reg_40001, value: 100}这样的 JSON查点表、按功能码分发、执行完回 ackdef make_command_handler(master, control_index, ack_topic): def _on_message(client, _userdata, msg): point_name None try: cmd json.loads(msg.payload.decode(utf-8)) point_name cmd[point] value cmd[value] pt control_index[point_name] # 白名单不在点表里的点位名直接 KeyError fc pt[fc] if fc 6: master.write_register(pt[unit_id], pt[addr], int(value)) # ... fc5写线圈、fc16写多寄存器同理 ack {point: point_name, value: value, success: True} except (KeyError, ValueError, TypeError, ModbusError) as exc: ack {point: point_name, success: False, error: str(exc)} client.publish(ack_topic, json.dumps(ack), qos1) return _on_message测试的时候直接复用了现有环境用 MQTT.fx 往factory/line1/ec942-0001/cmd发布{point: reg_40001, value: 100}PC 上 Modbus Slave 窗口里40001立刻变成了100下一轮采集的telemetry里也同步更新——写入和读回形成了完整闭环。这里必须多说一句安全上的考虑写操作比读操作危险得多读错了顶多是数据不对写错了是真的会影响现场设备。所以代码里做了两层防护一是点位白名单control_index只认control_points里显式列出的点位其他点位名直接被拒绝二是所有写操作不管成功失败都会记日志方便事后审计。生产环境上线前MQTT Broker 那边的账号权限也得单独收紧别让所有客户端账号都能往控制指令主题发消息。总结今天的踩坑清单现象根因解决方案ping通了nc/telnet/mbpoll全部找不到apt install也装不上精简镜像没装网络调试工具Debian 10已EOL默认源失效不装工具直接用Python内置socket.create_connection()测端口ImportError: cannot import name Literal/ModuleNotFoundError: No module named typing_extensionspaho-mqtt 2.x依赖typing.Literal需Python 3.8而设备是Python 3.7降级安装paho-mqtt1.6.1[Errno 111] Connection refused循环报错config.json里IP填错了一位.1写成想要的.11核对配置文件里的IP和实际环境是否一致MQTT.fx订阅无内容但设备端日志显示rc0订阅主题是精确匹配没加通配符#收不到子主题消息订阅改成factory/line1/ec942-0001/#从踩坑的分布能看出来今天遇到的坑大多不是协议本身难而是环境细节一个精简系统缺个工具、一个库版本不兼容、一个IP多打少打一位、一个MQTT主题没加通配符。这类问题排查起来比写代码逻辑本身更磨人也更容易被忽略希望这篇记录能帮到同样在踩坑路上的朋友。完整代码结构├── config.json # 点表配置读点位 写点位 MQTT连接参数 ├── modbus_tcp.py # Modbus TCP 主站实现读写寄存器/线圈 └── main.py # 主循环采集轮询 MQTT发布 控制指令订阅如果这篇文章对你有帮助欢迎点赞收藏后续如果接入真实PLC或者对接私有协议我会继续更新踩坑记录