基于Python的Modbus TCP工业一体机数据采集系统开发实战

工业一体机是面向工业环境设计的集成化计算设备集成了处理器、显示屏和工业通信接口具备宽温运行和抗电磁干扰能力常作为工业现场的数据采集与控制终端使用。在数据采集应用中工业一体机常需要通过Modbus TCP协议与PLC、变频器、仪表等设备进行通信。本文从开发环境搭建到完整数据采集系统实现详细讲解基于Python的Modbus TCP客户端开发全流程。## 开发环境- 操作系统Ubuntu 22.04 LTS / Debian 12- Python版本3.10- 核心依赖库pymodbus 3.6- 测试工具Modbus Slave模拟器bash# 创建虚拟环境python3 -m venv venvsource venv/bin/activate# 安装依赖pip install pymodbus3.6.3pip install pyinstaller # 打包用## Modbus TCP协议要点Modbus TCP基于以太网TCP/IP协议栈使用502端口通信。与Modbus RTU不同TCP版本去掉了CRC校验由TCP层保证可靠性增加了MBAP头部Modbus Application Protocol Header。MBAP头部共7字节| 字段 | 长度 | 说明 ||------|------|------|| Transaction ID | 2字节 | 事务标识用于请求/响应匹配 || Protocol ID | 2字节 | 协议标识Modbus固定为0x0000 || Length | 2字节 | 后续数据长度 || Unit ID | 1字节 | 从站地址 |常用的功能码python# 功能码定义FUNCTION_CODES {read_coils: 0x01, # 读线圈read_discrete_inputs: 0x02, # 读离散输入read_holding_regs: 0x03, # 读保持寄存器read_input_regs: 0x04, # 读输入寄存器write_single_coil: 0x05, # 写单个线圈write_single_reg: 0x06, # 写单个寄存器write_multi_regs: 0x10, # 写多个寄存器}## 完整客户端实现pythonimport loggingimport timefrom datetime import datetimefrom pymodbus.client import ModbusTcpClientfrom pymodbus.exceptions import ModbusExceptionclass IndustrialDataCollector:工业一体机Modbus TCP数据采集器def __init__(self, host192.168.1.100, port502, timeout3):self.host hostself.port portself.timeout timeoutself.client Noneself.reconnect_attempts 3self.reconnect_delay 2# 配置日志logging.basicConfig(levellogging.INFO,format%(asctime)s [%(levelname)s] %(message)s,handlers[logging.FileHandler(/var/log/modbus_collector.log),logging.StreamHandler()])self.logger logging.getLogger(__name__)def connect(self):建立Modbus TCP连接支持自动重连for attempt in range(self.reconnect_attempts):try:self.client ModbusTcpClient(hostself.host,portself.port,timeoutself.timeout)if self.client.connect():self.logger.info(f已连接到 {self.host}:{self.port})return Trueexcept Exception as e:self.logger.warning(f连接失败(第{attempt1}次): {e})time.sleep(self.reconnect_delay)self.logger.error(f无法连接到Modbus设备已尝试{self.reconnect_attempts}次)return Falsedef disconnect(self):断开连接if self.client and self.client.connected:self.client.close()self.logger.info(连接已断开)def read_holding_registers(self, unit_id, start_addr, count):读取保持寄存器:param unit_id: 从站地址:param start_addr: 起始寄存器地址:param count: 读取数量:return: 寄存器值列表或Nonetry:result self.client.read_holding_registers(addressstart_addr,countcount,slaveunit_id)if result.isError():self.logger.error(f读取保持寄存器失败 - 从站:{unit_id} f地址:{start_addr} 错误:{result})return Nonereturn result.registersexcept ModbusException as e:self.logger.error(fModbus异常: {e})return Nonedef write_register(self, unit_id, address, value):写入单个保持寄存器try:result self.client.write_register(addressaddress,valuevalue,slaveunit_id)if result.isError():self.logger.error(f写入失败 - 地址:{address} 错误:{result})return Falsereturn Trueexcept ModbusException as e:self.logger.error(fModbus异常: {e})return Falsedef collect_data(self, config):按配置批量采集数据:param config: 采集配置列表[{unit_id: 1, addr: 0, count: 10, name: 温度},{unit_id: 1, addr: 10, count: 5, name: 压力}]:return: 采集结果字典results {}timestamp datetime.now().strftime(%Y-%m-%d %H:%M:%S)for item in config:unit_id item[unit_id]addr item[addr]count item[count]name item[name]values self.read_holding_registers(unit_id, addr, count)if values is not None:results[name] {values: values,timestamp: timestamp,unit_id: unit_id,addr: addr}self.logger.info(f[{name}] 从站:{unit_id} 地址:{addr} f值:{values})else:results[name] Noneself.logger.warning(f[{name}] 采集失败)return results## 数据格式转换Modbus寄存器是16位的但实际工业数据可能是32位浮点数Float32或32位整数Int32需要做字节拼接处理。pythonimport structclass DataConverter:Modbus数据格式转换工具staticmethoddef regs_to_float32(regs, byte_ordercdab):将两个寄存器值转换为32位浮点数:param regs: [reg_high, reg_low] 两个寄存器值:param byte_order: 字节序abcd - 大端序badc - 字交换大端cdab - 字交换小端dcba - 小端序:return: 浮点数值if len(regs) 2:return None# 将寄存器值转为字节b regs[0].to_bytes(2, big) regs[1].to_bytes(2, big)if byte_order abcd:passelif byte_order badc:b b[1:2] b[0:1] b[3:4] b[2:3]elif byte_order cdab:b b[2:4] b[0:2]elif byte_order dcba:b b[3:4] b[2:3] b[1:2] b[0:1]return struct.unpack(f, b)[0]staticmethoddef regs_to_int32(regs, signedTrue):将两个寄存器值转换为32位整数if len(regs) 2:return Noneraw (regs[0] 16) | regs[1]if signed and raw 0x80000000:raw - 0x100000000return rawstaticmethoddef regs_to_string(regs, encodingutf-8):将寄存器值列表转为字符串raw_bytes bfor reg in regs:raw_bytes reg.to_bytes(2, big)try:return raw_bytes.decode(encoding).strip(\x00)except UnicodeDecodeError:return None## 看门狗与自动恢复工业现场数据采集要求7×24小时不间断运行必须实现看门狗机制。pythonimport signalimport osimport threadingclass Watchdog:进程看门狗监控采集线程存活状态def __init__(self, check_interval30, max_misses3):self.check_interval check_intervalself.max_misses max_missesself.miss_count 0self.last_heartbeat time.time()self._running Truedef heartbeat(self):采集线程定期调用表示存活self.last_heartbeat time.time()self.miss_count 0def start(self, on_timeout_callback):启动看门狗监控线程def _monitor():while self._running:elapsed time.time() - self.last_heartbeatif elapsed self.check_interval:self.miss_count 1logging.warning(f采集线程心跳超时 f(已等待{elapsed:.0f}s, 第{self.miss_count}次))if self.miss_count self.max_misses:logging.error(f心跳连续超时{self.max_misses}次f触发恢复流程)on_timeout_callback()self.miss_count 0time.sleep(self.check_interval)thread threading.Thread(target_monitor, daemonTrue)thread.start()def stop(self):self._running False## 主程序入口pythondef main():# 采集配置config [{unit_id: 1, addr: 0, count: 2, name: 反应釜温度},{unit_id: 1, addr: 2, count: 2, name: 管道压力},{unit_id: 2, addr: 0, count: 10, name: 产线计数},]collector IndustrialDataCollector(host192.168.1.100,port502,timeout3)converter DataConverter()watchdog Watchdog(check_interval30, max_misses3)def on_watchdog_timeout():看门狗超时回调重连设备logging.info(正在重新连接Modbus设备...)collector.disconnect()time.sleep(5)collector.connect()watchdog.start(on_watchdog_timeout)# 主采集循环try:if not collector.connect():sys.exit(1)while True:results collector.collect_data(config)# 数据转换示例if results.get(反应釜温度):regs results[反应釜温度][values]temp converter.regs_to_float32(regs, cdab)logging.info(f反应釜温度: {temp:.2f}℃)watchdog.heartbeat()time.sleep(1) # 1秒采集间隔except KeyboardInterrupt:logging.info(收到退出信号正在关闭...)finally:collector.disconnect()watchdog.stop()if __name__ __main__:main()## 常见调试问题**连接超时**检查网络是否通502端口是否开放。用telnet测试telnet 192.168.1.100 502。**读取返回非法数据地址0x02**寄存器地址从0开始还是从1开始是常见坑。Modbus协议规范是0-based但有些设备文档用1-based地址需要减1。**数据值不对**99%是字节序问题。用Modbus Poll工具读取原始寄存器值手动拼接验证。不同厂家的字节序不同常见的有ABCD大端、CDAB字交换等。**间歇性断连**检查TCP KeepAlive设置工业网络中交换机可能主动断开空闲连接。在创建client时加上keepalive参数。## 常见问题FAQ**QModbus TCP和Modbus RTU的主要区别是什么**AModbus TCP基于以太网TCP/IP协议使用502端口去掉了CRC校验由TCP层保证增加了7字节MBAP头部Modbus RTU基于串口RS485/RS232使用CRC16校验。TCP版本传输距离更远、速度更快、支持多主机访问RTU版本成本低、布线简单适合短距离点对多通信。**QPython pymodbus库如何处理寄存器地址0-based和1-based的问题**Apymodbus库使用0-based地址即第一个寄存器地址为0。如果设备文档标注地址为1-based在代码中需要将地址减1。建议先用Modbus Poll工具测试确认地址映射关系再写入代码避免因地址偏移导致读取到错误数据。**Q工业数据采集系统如何保证7×24小时不间断运行**A需要实现三个层级的保障进程级使用看门狗监控采集线程心跳超时自动重启连接级设置自动重连机制检测到断连后延时重试系统级可配合Linux的systemd服务管理设置Restartalways实现进程崩溃后自动拉起。