# Modbus协议实战：如何用Python快速读取PLC寄存器数据（附代码） 在工业自动化领域，数据是驱动一切决策和控制的血液。无论是监控生产线上的温度、压力，还是统计设备的运行时长、能耗，其核心都离不开与底层可编程逻辑控制器（PLC）的稳定、高效通信。对于众多开发者而言，尤其是那些具备Python技能栈的工程师，如何绕过复杂的工业软件，直接使用自己熟悉的工具与PLC“对话”，获取关键的寄存器数据，是一个既充满挑战又极具价值的课题。Modbus协议，作为工业通信领域事实上的“普通话”，为我们架起了这座桥梁。本文将从一线开发者的实战视角出发，抛开繁复的理论堆砌，直击要害，手把手带你用Python库（如pymodbus）打通与PLC的通信链路，涵盖从连接建立、功能码选择到寄存器地址解析的全流程，并提供可直接复制、修改并投入生产的代码示例。无论你是希望快速搭建一个数据采集原型，还是优化现有的监控系统，这里都有你需要的“干货”。 ## 1. 理解Modbus通信的基石：数据模型与地址映射 在动手写代码之前，我们必须先厘清几个核心概念，否则很容易在后续的调试中陷入“地址对不上”、“数据读不对”的泥潭。Modbus协议定义了一个清晰但稍显抽象的数据模型，它是对PLC内部存储区域的逻辑划分。 简单来说，你可以把PLC的内存想象成一个巨大的表格，Modbus协议为这个表格的不同区域贴上了四种标签： * **线圈（Coils）**：可读可写的开关量。每个线圈代表一个比特（bit），状态非0即1。它通常对应实际的数字量输出（DO），比如控制一个继电器吸合（1）或断开（0）。 * **离散输入（Discrete Inputs）**：只读的开关量。同样是一个比特，用于反映外部数字量输入（DI）的状态，例如一个限位开关是否被触发。 * **输入寄存器（Input Registers）**：只读的模拟量或数据。每个寄存器占16位（2个字节），通常用于存储来自传感器等设备的只读数据，如温度、压力等模拟量输入（AI）的瞬时值。 * **保持寄存器（Holding Registers）**：可读可写的模拟量或数据。这是我们在数据交互中最常打交道的区域，常用于存储可修改的参数（如设定值）、或作为中间计算结果存储区。 这四种数据模型在协议中用不同的“地址区域”来标识。我们常看到的诸如“40001”、“30004”这类地址，就是一种“PLC地址”或“协议数据单元（PDU）地址”的表示方法。理解其映射规则至关重要： | 数据模型 | 读写属性 | 数据类型 | 常见地址范围（PLC地址表示法） | 对应协议地址偏移量 | | :--- | :--- | :--- | :--- | :--- | | **线圈** | 读/写 | 位（Bit） | 000001 - 099999 | 0 - 65535 | | **离散输入** | 只读 | 位（Bit） | 100001 - 199999 | 0 - 65535 | | **输入寄存器** | 只读 | 字（Word, 16-bit） | 300001 - 399999 | 0 - 65535 | | **保持寄存器** | 读/写 | 字（Word, 16-bit） | 400001 - 499999 | 0 - 65535 | > **注意**：上表中的“PLC地址”（如40001）是一种便于人类识别的表示方式，其前导数字（4）代表寄存器类型，后五位（00001）代表在该类型区域内的偏移地址（从1开始）。而在实际的Modbus协议帧中，我们使用的是从0开始的偏移地址。例如，PLC地址40001对应协议地址0，40002对应协议地址1，依此类推。几乎所有Modbus库函数都需要你传入这个从0开始的协议地址。 **功能码（Function Code）** 则是告诉PLC“你想干什么”的指令。例如，读取线圈用功能码`0x01`，读取保持寄存器用`0x03`，写入单个保持寄存器用`0x06`，写入多个保持寄存器用`0x10`。在Python的pymodbus库中，这些功能码被封装成了直观的方法名，我们通常不需要直接记忆这些十六进制数字。 ## 2. 搭建Python开发环境与连接PLC 工欲善其事，必先利其器。我们选择`pymodbus`这个强大的第三方库，它同时支持Modbus RTU（串行）和TCP/IP（网络）两种传输方式，API设计也较为友好。 首先，通过pip安装pymodbus： ```bash pip install pymodbus ``` 接下来，根据你的PLC连接方式（网络或串口）选择不同的客户端。这里我们以更常见的Modbus TCP为例。 假设你的PLC IP地址是`192.168.1.100`，Modbus TCP端口默认为`502`。创建一个连接并读取数据的示例代码如下： ```python from pymodbus.client import ModbusTcpClient # 1. 创建客户端并建立连接 PLC_IP = '192.168.1.100' PLC_PORT = 502 client = ModbusTcpClient(PLC_IP, port=PLC_PORT) if client.connect(): # 尝试连接 print("成功连接到PLC") # 2. 尝试读取数据（例如：从保持寄存器地址0开始，读取10个寄存器） # 注意：这里的地址0对应PLC地址40001 start_address = 0 register_count = 10 try: response = client.read_holding_registers(address=start_address, count=register_count, slave=1) # slave参数是从站地址，对于多数TCP设备是1，具体需查阅PLC手册 if not response.isError(): # 读取成功，response.registers是一个包含读取值的列表 data_list = response.registers print(f"读取到的寄存器值（地址 {start_address} 到 {start_address+register_count-1}）: {data_list}") # 3. 数据处理示例：假设第一个寄存器（地址0）是一个16位有符号整数 # 注意：Modbus寄存器本身是无符号16位整数，需要根据实际含义转换 raw_value = data_list[0] # 如果实际是有符号整数（范围-32768 ~ 32767） if raw_value >= 0x8000: # 判断最高位是否为1（负数） signed_value = raw_value - 0x10000 else: signed_value = raw_value print(f"地址0的原始值：{raw_value}，解释为有符号整数：{signed_value}") else: print(f"读取寄存器时发生错误：{response}") except Exception as e: print(f"通信过程中发生异常：{e}") finally: # 4. 关闭连接 client.close() print("连接已关闭") else: print("无法连接到PLC，请检查网络和地址") ``` 这段代码完成了从连接到读取再到安全关闭的完整流程。关键点在于`read_holding_registers`方法，它封装了功能码`0x03`的请求。你需要根据PLC的配置手册，确定正确的**从站地址（slave）** 和**寄存器起始地址**。 ## 3. 核心操作解析：读写寄存器与数据处理实战 仅仅建立连接和简单读取是远远不够的。在实际项目中，我们会遇到各种复杂的数据类型和操作需求。 ### 3.1 写入寄存器数据 向PLC写入设定值或控制命令是常见操作。写入分为写入单个寄存器和写入多个寄存器。 ```python # 接续上面的连接成功部分 # 写入单个保持寄存器（功能码 0x06） # 向地址为5的寄存器（PLC地址40006）写入值1234 write_address = 5 write_value = 1234 response = client.write_register(address=write_address, value=write_value, slave=1) if not response.isError(): print(f"成功向地址 {write_address} 写入值 {write_value}") else: print(f"写入失败：{response}") # 写入多个保持寄存器（功能码 0x10） # 向地址10开始连续写入3个值 [100, 200, 300] start_write_address = 10 values_to_write = [100, 200, 300] response = client.write_registers(address=start_write_address, values=values_to_write, slave=1) if not response.isError(): print(f"成功从地址 {start_write_address} 开始写入多个值：{values_to_write}") ``` ### 3.2 处理复杂数据类型 PLC中的一个数据往往不止占一个16位寄存器。例如，32位整数（DINT）、浮点数（Float）需要占用两个连续的寄存器。Modbus协议通常采用“高位在前（Big-Endian）”的字节序，但不同品牌的PLC可能有差异（如西门子常用“低位在前”）。 下面是一个读取32位有符号整数和单精度浮点数的通用函数示例： ```python import struct def read_32bit_signed_int(client, start_reg_address, slave=1): """读取两个寄存器并组合成一个32位有符号整数（假设高位在前）""" response = client.read_holding_registers(start_reg_address, count=2, slave=slave) if response.isError(): raise Exception(f"读取失败：{response}") high_word, low_word = response.registers # 将两个16位无符号整数组合成4个字节 # 注意：这里假设PLC的字节序是高位字在前（即 high_word 是数据的高16位） byte_data = struct.pack('>HH', high_word, low_word) # '>' 表示大端字节序 # 解包为32位有符号整数 value, = struct.unpack('>i', byte_data) # 'i' 表示有符号整数 return value def read_float(client, start_reg_address, slave=1, byteorder='>'): """读取两个寄存器并组合成一个单精度浮点数""" response = client.read_holding_registers(start_reg_address, count=2, slave=slave) if response.isError(): raise Exception(f"读取失败：{response}") reg1, reg2 = response.registers # 根据PLC的字节序调整结构体格式字符 # '>' 大端（常见），'<' 小端（如某些西门子PLC） fmt = f'{byteorder}HH' byte_data = struct.pack(fmt, reg1, reg2) value, = struct.unpack(f'{byteorder}f', byte_data) # 'f' 表示浮点数 return value # 使用示例 try: int32_value = read_32bit_signed_int(client, start_reg_address=20, slave=1) print(f"从地址20读取的32位整数：{int32_value}") float_value = read_float(client, start_reg_address=30, slave=1, byteorder='>') print(f"从地址30读取的浮点数：{float_value}") except Exception as e: print(e) ``` > **提示**：字节序（Endianness）是跨平台数据交换的常见坑点。务必查阅你所连接PLC的技术文档，确认其Modbus数据映射的字节顺序。如果上述大端模式读取的数据不对，尝试将`byteorder`参数改为`'<'`（小端）。 ### 3.3 错误处理与连接稳健性 工业现场网络可能不稳定。我们必须让程序具备重连和异常处理能力。 ```python import time from pymodbus.exceptions import ConnectionException, ModbusException def robust_modbus_read(client, address, count, slave=1, retries=3): """带重试机制的读取函数""" for attempt in range(retries): try: if not client.is_socket_open(): print("连接断开，尝试重连...") client.connect() time.sleep(0.5) # 连接后稍作等待 response = client.read_holding_registers(address, count=count, slave=slave) if response.isError(): # 处理Modbus协议层面的错误（如非法地址、功能码） print(f"Modbus错误（尝试{attempt+1}/{retries}）: {response}") time.sleep(1) continue return response.registers except ConnectionException as ce: print(f"连接异常（尝试{attempt+1}/{retries}）: {ce}") time.sleep(2) # 等待更长时间后重试 except ModbusException as me: print(f"Modbus通信异常（尝试{attempt+1}/{retries}）: {me}") time.sleep(1) except Exception as e: print(f"未知异常（尝试{attempt+1}/{retries}）: {e}") time.sleep(1) print(f"读取失败，已达最大重试次数{retries}") return None # 在主循环中使用 while True: data = robust_modbus_read(client, address=0, count=5, slave=1) if data is not None: print(f"实时数据: {data}") time.sleep(2) # 每2秒采集一次 ``` ## 4. 构建一个简易的PLC数据监控与记录系统 将上述知识点整合，我们可以构建一个更有实用价值的小系统。这个系统会周期性地从PLC读取一组关键数据（包括开关量、整型、浮点型），将其显示在控制台，并同时记录到CSV文件中，便于后续分析。 ```python import csv import time from datetime import datetime from pymodbus.client import ModbusTcpClient class PLCDataMonitor: def __init__(self, ip, port=502, slave_id=1): self.client = ModbusTcpClient(ip, port=port) self.slave_id = slave_id self.csv_file = None self.csv_writer = None def connect(self): if self.client.connect(): print(f"[{datetime.now()}] 已连接至PLC {self.client.params.host}") return True else: print("连接失败") return False def setup_data_logging(self, filename='plc_data_log.csv'): """初始化CSV日志文件""" self.csv_file = open(filename, 'a', newline='') self.csv_writer = csv.writer(self.csv_file) # 写入表头 headers = ['Timestamp', 'Coil_Status', 'Temp_Raw', 'Temp_Float', 'Pressure', 'Energy_Count'] self.csv_writer.writerow(headers) print(f"数据将记录到文件: {filename}") def read_sensor_data(self): """定义需要读取的数据点及其地址""" data_points = {} # 1. 读取一个线圈状态（例如：设备运行信号，地址0 -> PLC地址00001） try: response = self.client.read_coils(address=0, count=1, slave=self.slave_id) if not response.isError(): data_points['coil_status'] = response.bits[0] # 返回的是BitArray，取第一位 except Exception as e: data_points['coil_status'] = None print(f"读取线圈失败: {e}") # 2. 读取一个保持寄存器中的原始温度值（16位整数，地址10 -> 40011） try: response = self.client.read_holding_registers(address=10, count=1, slave=self.slave_id) if not response.isError(): data_points['temp_raw'] = response.registers[0] except Exception as e: data_points['temp_raw'] = None print(f"读取温度原始值失败: {e}") # 3. 读取一个浮点数温度（占用地址30,31 -> 40031, 40032） try: # 使用之前定义的read_float函数，假设大端字节序 float_val = read_float(self.client, start_reg_address=30, slave=self.slave_id, byteorder='>') data_points['temp_float'] = round(float_val, 2) # 保留两位小数 except Exception as e: data_points['temp_float'] = None print(f"读取浮点温度失败: {e}") # 4. 读取压力值（32位整数，占用地址40,41） try: int32_val = read_32bit_signed_int(self.client, start_reg_address=40, slave=self.slave_id) data_points['pressure'] = int32_val except Exception as e: data_points['pressure'] = None print(f"读取压力值失败: {e}") return data_points def log_data(self, data): """将数据写入CSV并打印""" timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') row = [ timestamp, data.get('coil_status', 'N/A'), data.get('temp_raw', 'N/A'), data.get('temp_float', 'N/A'), data.get('pressure', 'N/A'), # 可以添加更多数据项... ] if self.csv_writer: self.csv_writer.writerow(row) # 控制台输出 print(f"[{timestamp}] 状态:{row[1]}, 温度(原始):{row[2]}, 温度(℃):{row[3]}, 压力:{row[4]}") def run(self, interval_seconds=5): """主循环""" if not self.connect(): return self.setup_data_logging() try: while True: data = self.read_sensor_data() self.log_data(data) time.sleep(interval_seconds) except KeyboardInterrupt: print("\n用户中断，停止监控...") finally: self.cleanup() def cleanup(self): """清理资源""" if self.csv_file: self.csv_file.close() print("日志文件已关闭") self.client.close() print("Modbus连接已关闭") # 运行监控器 if __name__ == "__main__": # 请替换为实际的PLC IP地址和从站ID monitor = PLCDataMonitor(ip='192.168.1.100', slave_id=1) monitor.run(interval_seconds=5) ``` 这个类封装了连接、多数据类型读取、数据记录和稳健运行的功能。你可以根据实际PLC的地址映射表，轻松修改`read_sensor_data`方法中的地址和数据类型解析逻辑。运行后，你将在控制台看到实时刷新的数据，并在同目录下生成一个包含时间戳和所有数据的CSV文件，为后续的数据分析和可视化（如用Pandas、Matplotlib处理）提供了便利。 在实际部署中，你可能还需要考虑将配置（如IP、地址映射）外置到配置文件（如YAML、JSON），添加更完善的日志记录（使用logging模块），甚至结合Web框架（如Flask）提供简单的数据API。从这段基础代码出发，你已经拥有了一个功能完整、可扩展的PLC数据交互核心。