工控领域最常用的通信协议Modbus排第二没人敢排第一。虽然协议本身很简单但工程实现中各种细节问题不少。本文用Python从零实现一套完整的Modbus TCP通信方案覆盖建连、读写、异常处理、多设备轮询等实际场景。## 为什么选Python工控现场C/C还是主流但Python在以下场景有明显优势- 快速原型验证和协议调试- 数据采集网关的轻量实现- 边缘计算设备上的数据预处理- 测试环境的自动化脚本性能方面Python处理Modbus TCP完全够用——Modbus TCP的典型轮询周期在百毫秒级Python的网络IO延迟在毫秒级不构成瓶颈。## 依赖准备bashpip install pymodbus3.6.3pymodbus是Python最成熟的Modbus实现支持RTU和TCP3.x版本API有较大变化以下代码基于3.6.3版本。## 一、基础连接和单次读写pythonfrom pymodbus.client import ModbusTcpClientimport structimport timeimport logging# 配置日志方便调试logging.basicConfig(levellogging.INFO)logger logging.getLogger(__name__)class ModbusTCPClient:Modbus TCP客户端封装def __init__(self, host, port502, timeout5, retries3):self.host hostself.port portself.timeout timeoutself.retries retriesself.client Nonedef connect(self):建立连接self.client ModbusTcpClient(hostself.host,portself.port,timeoutself.timeout,retriesself.retries)if self.client.connect():logger.info(f连接成功: {self.host}:{self.port})return Trueelse:logger.error(f连接失败: {self.host}:{self.port})return Falsedef disconnect(self):断开连接if self.client:self.client.close()logger.info(f连接已断开: {self.host}:{self.port})def read_holding_registers(self, address, count, unit1):读取保持寄存器功能码03Args:address: 寄存器起始地址count: 读取数量unit: 从站地址Returns:list: 寄存器值列表失败返回Nonetry:result self.client.read_holding_registers(addressaddress,countcount,slaveunit)if result.isError():logger.error(f读取失败 addr{address}: {result})return Nonereturn result.registersexcept Exception as e:logger.error(f读取异常 addr{address}: {e})return Nonedef write_single_register(self, address, value, unit1):写单个寄存器功能码06Args:address: 寄存器地址value: 写入值unit: 从站地址Returns:bool: 成功返回Truetry:result self.client.write_register(addressaddress,valuevalue,slaveunit)if result.isError():logger.error(f写入失败 addr{address}: {result})return Falsereturn Trueexcept Exception as e:logger.error(f写入异常 addr{address}: {e})return Falsedef write_multiple_registers(self, address, values, unit1):写多个寄存器功能码16Args:address: 起始地址values: 值列表unit: 从站地址Returns:bool: 成功返回Truetry:result self.client.write_registers(addressaddress,valuesvalues,slaveunit)if result.isError():logger.error(f批量写入失败 addr{address}: {result})return Falsereturn Trueexcept Exception as e:logger.error(f批量写入异常 addr{address}: {e})return False# 基本使用示例 if __name__ __main__:client ModbusTCPClient(192.168.1.100, port502)if client.connect():# 读取保持寄存器地址0开始读10个registers client.read_holding_registers(address0, count10, unit1)if registers:print(f读取到寄存器值: {registers})# 写单个寄存器success client.write_single_register(address0, value100, unit1)print(f写入结果: {success})# 写多个寄存器success client.write_multiple_registers(address10, values[1, 2, 3, 4, 5], unit1)print(f批量写入结果: {success})client.disconnect()## 二、32位浮点数的读写处理Modbus寄存器是16位的但工业传感器经常返回32位浮点数如温度、压力、流量需要两个连续寄存器拼接。这里的关键是字节序问题不同设备的存储顺序可能不同。pythonclass ModbusDataConverter:Modbus数据类型转换工具staticmethoddef registers_to_float(reg1, reg2, byte_orderbig, word_orderbig):两个16位寄存器转32位浮点数Args:reg1: 第一个寄存器值reg2: 第二个寄存器值byte_order: 字节序 big 或 littleword_order: 字序 big(高字在前) 或 little(低字在前)Returns:float: 解析后的浮点数if word_order big:# 高字在前reg1是高16位raw (reg1 16) | reg2else:# 低字在前reg2是高16位raw (reg2 16) | reg1# 将32位整数转为4字节if byte_order big:byte_data struct.pack(I, raw)else:byte_data struct.pack(I, raw)return struct.unpack(f, byte_data)[0]staticmethoddef float_to_registers(value, byte_orderbig, word_orderbig):32位浮点数转两个16位寄存器值Args:value: 浮点数值byte_order: 字节序word_order: 字序Returns:tuple: (reg1, reg2)byte_data struct.pack(f, value)raw struct.unpack(I, byte_data)[0]if word_order big:reg1 (raw 16) 0xFFFFreg2 raw 0xFFFFelse:reg1 raw 0xFFFFreg2 (raw 16) 0xFFFFreturn reg1, reg2staticmethoddef registers_to_int32(reg1, reg2, signedTrue, word_orderbig):两个寄存器转32位整数if word_order big:raw (reg1 16) | reg2else:raw (reg2 16) | reg1if signed:return struct.unpack(i, struct.pack(I, raw))[0]return rawstaticmethoddef registers_to_float_array(registers, byte_orderbig, word_orderbig):批量将寄存器数组转为浮点数数组每2个寄存器对应1个浮点数Args:registers: 寄存器值列表byte_order: 字节序word_order: 字序Returns:list: 浮点数列表floats []for i in range(0, len(registers) - 1, 2):val ModbusDataConverter.registers_to_float(registers[i], registers[i 1],byte_order, word_order)floats.append(val)return floats# 浮点数读写示例 def read_temperature(client, address0, unit1):读取温度值32位浮点数占2个寄存器registers client.read_holding_registers(address, count2, unitunit)if registers and len(registers) 2:# 注意不同设备字节序可能不同需要根据设备手册调整temp ModbusDataConverter.registers_to_float(registers[0], registers[1],byte_orderbig, word_orderlittle # 很多PLC用低字在前)return tempreturn None## 三、多设备轮询采集实际项目中通常需要同时采集多台设备的数据以下是带异常恢复的多设备轮询实现pythonimport threadingimport queueimport jsonfrom datetime import datetimeclass ModbusPollingManager:多设备Modbus TCP轮询管理器def __init__(self, config_list, poll_interval1.0):Args:config_list: 设备配置列表[{name: 温控仪1,host: 192.168.1.100,port: 502,unit: 1,registers: [{name: 温度, address: 0, count: 2, type: float},{name: 湿度, address: 2, count: 2, type: float},{name: 运行状态, address: 10, count: 1, type: uint16}]}]poll_interval: 轮询间隔秒self.config_list config_listself.poll_interval poll_intervalself.clients {}self.data_queue queue.Queue()self.running Falseself.poll_thread Nonedef init_connections(self):初始化所有设备连接for cfg in self.config_list:client ModbusTCPClient(hostcfg[host],portcfg.get(port, 502),timeoutcfg.get(timeout, 3),retriescfg.get(retries, 2))if client.connect():self.clients[cfg[name]] {client: client,config: cfg,fail_count: 0}logger.info(f设备 [{cfg[name]}] 连接成功)else:logger.error(f设备 [{cfg[name]}] 连接失败)self.clients[cfg[name]] {client: client,config: cfg,fail_count: 0}def _parse_register_value(self, registers, reg_type, byte_orderbig, word_orderlittle):根据数据类型解析寄存器值if not registers:return Noneif reg_type uint16:return registers[0]elif reg_type int16:val registers[0]return val - 65536 if val 32767 else valelif reg_type float:if len(registers) 2:return ModbusDataConverter.registers_to_float(registers[0], registers[1], byte_order, word_order)elif reg_type int32:if len(registers) 2:return ModbusDataConverter.registers_to_int32(registers[0], registers[1], word_orderword_order)return Nonedef _poll_single_device(self, device_name):轮询单个设备dev self.clients[device_name]client dev[client]cfg dev[config]# 检查是否需要重连if dev[fail_count] 3:logger.warning(f设备 [{device_name}] 连续失败{dev[fail_count]}次尝试重连)client.disconnect()time.sleep(1)if not client.connect():dev[fail_count] 1returndata {device: device_name,timestamp: datetime.now().isoformat(),values: {}}for reg_cfg in cfg[registers]:registers client.read_holding_registers(addressreg_cfg[address],countreg_cfg[count],unitcfg.get(unit, 1))if registers is not None:value self._parse_register_value(registers,reg_cfg[type],word_orderreg_cfg.get(word_order, little))data[values][reg_cfg[name]] valuedev[fail_count] 0else:data[values][reg_cfg[name]] Nonedev[fail_count] 1logger.warning(f设备 [{device_name}] 读取 {reg_cfg[name]} 失败 f(连续失败{dev[fail_count]}次))self.data_queue.put(data)def _polling_loop(self):轮询主循环while self.running:for device_name in list(self.clients.keys()):try:self._poll_single_device(device_name)except Exception as e:logger.error(f轮询设备 [{device_name}] 异常: {e})time.sleep(self.poll_interval)def start(self):启动轮询self.init_connections()self.running Trueself.poll_thread threading.Thread(targetself._polling_loop,daemonTrue,namemodbus-polling)self.poll_thread.start()logger.info(轮询线程已启动)def stop(self):停止轮询self.running Falseif self.poll_thread:self.poll_thread.join(timeout5)for dev in self.clients.values():dev[client].disconnect()logger.info(轮询已停止所有连接已断开)def get_data(self, timeout1.0):从队列获取一条数据try:return self.data_queue.get(timeouttimeout)except queue.Empty:return None# 多设备轮询示例 if __name__ __main__:devices [{name: 温控仪A,host: 192.168.1.100,port: 502,unit: 1,timeout: 3,registers: [{name: 当前温度, address: 0, count: 2, type: float},{name: 设定温度, address: 2, count: 2, type: float},{name: 运行状态, address: 10, count: 1, type: uint16}]},{name: 电表B,host: 192.168.1.101,port: 502,unit: 1,timeout: 3,registers: [{name: 电压, address: 0x1000, count: 2, type: float},{name: 电流, address: 0x1002, count: 2, type: float},{name: 功率, address: 0x1004, count: 2, type: float}]}]manager ModbusPollingManager(devices, poll_interval1.0)manager.start()try:# 运行60秒打印采集到的数据for _ in range(60):data manager.get_data(timeout2.0)if data:print(json.dumps(data, ensure_asciiFalse, indent2))except KeyboardInterrupt:passfinally:manager.stop()## 四、工程中踩过的坑**坑1连接超时设置。** 默认timeout太长30秒设备掉线时整个轮询卡住。建议设为3-5秒。同时pymodbus的retries参数控制重试次数设为2-3次就够了。**坑2从站地址unit/slave。** pymodbus 3.x中参数名是slave不是unit。从站地址范围1-2470是广播地址。有些设备默认从站地址不是1需要查手册确认。**坑3寄存器地址偏移。** 有些设备手册上写的地址是PLC地址从1开始Modbus协议地址是从0开始的。比如手册上写地址40001实际协议地址是0。差1容易导致读到的数据全错。**坑4浮点数字节序。** 这是Modbus最大的坑。同样读两个寄存器有的设备高字在前有的低字在前。建议先用Modbus Poll等工具手动读取确认再在代码中配置正确的word_order。**坑5并发访问。** pymodbus的client对象不是线程安全的。多线程环境下要么每个线程用独立的client实例要么加锁串行访问。上面的轮询管理器是单线程轮询不存在这个问题。**坑6TCP连接断开检测。** Modbus TCP的TCP连接可能被对端静默关闭pymodbus不一定能及时检测到。建议在read/write失败后主动close并reconnect而不是依赖库的自动重连。## 五、性能优化建议如果需要采集大量设备或高频采集可以考虑以下优化**用异步IO替代线程。** pymodbus 3.x提供了asyncio版本的客户端在高并发场景下性能远优于线程模型pythonimport asynciofrom pymodbus.client import AsyncModbusTcpClientasync def async_read(client, address, count, unit):result await client.read_holding_registers(address, count, slaveunit)if not result.isError():return result.registersreturn Noneasync def main():client AsyncModbusTcpClient(192.168.1.100)await client.connect()# 并发读取多个地址tasks [async_read(client, 0, 2, 1),async_read(client, 10, 2, 1),async_read(client, 20, 4, 1),]results await asyncio.gather(*tasks)print(results)client.close()asyncio.run(main())**连接池复用。** 避免每次读写都建立新连接保持长连接。但要注意设置心跳机制定时读取一个寄存器防止NAT或防火墙超时断开TCP连接。**批量读取优化。** 如果多个寄存器地址连续一次读多个比多次单个读取效率高得多。TCP协议帧中最大可以读125个寄存器功能码03。## 总结Modbus TCP协议本身不复杂但工程实现中数据类型转换特别是浮点数字节序、异常恢复、多设备并发这些环节需要仔细处理。建议在正式部署前用Modbus Poll等工具充分验证协议参数避免在现场调试时反复试错。
