Python Modbus库完全指南：从入门到精通

서론：为什么选择Python进行Modbus개발？

在工业物联网和自动化领域，Python凭借其简洁的语法、丰富的库生态系统和强大的社区支持，已经成为개발者的首选언어之一。对于Modbus프로토콜개발，Python提供了多个成熟的오픈소스库，能够满足从简单设备통신到复杂工业系统통합的各种需求。 本文将全面介绍Python中主流的Modbus库，包括pymodbus、minimalmodbus、uModbus等，通过实际코드 예제和最佳实践，帮助您快速掌握Python Modbus개발的核心技能。

一、主流Python Modbus库对比

1.1 pymodbus：功能最全面的选择

特点： - 支持Modbus TCP、RTU、ASCII等多种프로토콜 - 同时支持클라이언트（마스터）和服务器（슬레이브）모드 - 异步/同步两种编程모드 - 活跃的社区和持续更新 适用시나리오： - 复杂的工业控制系统 - 需要同时作为마스터和슬레이브的시나리오 - 高性能要求的적용

1.2 minimalmodbus：轻量级솔루션

特点： - 代码简洁，易于학습和使用 - 专注于Modbus RTU프로토콜 - 依赖少，部署简单 - 适合嵌入式系统和자료受限환경 适用시나리오： - 简单的设备통신 - 嵌入式系统和单板계산机 - 快速原型개발

1.3 uModbus：现代化设计

特点： - 现代化的API设计 - 支持异步编程 - 良好的유형提示 - 模块化架构 适用시나리오： - 需要异步处理的적용 - 大型项目개발 - 需要良好代码维护性的시나리오

二、pymodbus深度분석

2.1 安装和基础설정

# 安装pymodbus
pip install pymodbus

# 或者安装包含异步支持的버전
pip install pymodbus[asyncio]

2.2 Modbus TCP클라이언트예제

from pymodbus.client import ModbusTcpClient

# 创建TCP클라이언트
client = ModbusTcpClient(
    host='192.168.1.100',  # 设备IP주소
    port=502,              # Modbus TCP포트
    timeout=3,             # 超时时间（秒）
    retries=3              # 重试次数
)

try:
    # 연결设备
    if client.connect():
        print("成功연결到Modbus设备")

        # 유지 레지스터 읽기 (FC03)（주소0，数量10）
        result = client.read_holding_registers(주소=0, count=10, slave=1)
        if not result.is오답():
            print(f"읽기到的레지스터 값: {result.registers}")
        else:
            print(f"읽기失败: {result}")

        # 쓰기单个유지 레지스터（주소5，值1234）
        result = client.write_register(주소=5, value=1234, slave=1)
        if not result.is오답():
            print("레지스터쓰기成功")

        # 코일 읽기 (FC01)（주소0，数量8）
        result = client.read_coils(주소=0, count=8, slave=1)
        if not result.is오답():
            print(f"코일 상태: {result.bits}")

finally:
    # 닫기연결
    client.close()

2.3 Modbus RTU클라이언트예제

from pymodbus.client import ModbusSerialClient

# 创建RTU클라이언트
client = ModbusSerialClient(
    method='rtu',          # 프로토콜유형（rtu或ascii）
    port='/dev/ttyUSB0',   # 시리얼 포트设备
    baudrate=9600,         # 통신 속도
    bytesize=8,            # 데이터 비트
    parity='N',            # 패리티（N:无, E:偶, O:奇）
    stopbits=1,            # 정지 비트
    timeout=1              # 超时时间
)

try:
    if client.connect():
        print("成功연결到시리얼 포트设备")

        # 批量입력 레지스터 읽기 (FC04)
        result = client.read_input_registers(주소=0, count=20, slave=1)
        if not result.is오답():
            데이터 = result.registers
            # 处理데이터：假设前10个是온도值（需要除以10）
            temperatures = [value/10.0 for value in 데이터[:10]]
            print(f"온도데이터: {temperatures}")

finally:
    client.close()

2.4 异步클라이언트예제

import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def read_device_데이터():
    # 创建异步클라이언트
    client = AsyncModbusTcpClient(
        host='192.168.1.100',
        port=502
    )

    await client.connect()

    try:
        # 同时읽기多个设备的데이터
        tasks = []
        for slave_id in [1, 2, 3]:
            task = client.read_holding_registers(
                주소=0, 
                count=10, 
                slave=slave_id
            )
            tasks.append(task)

        # 并行执行所有읽기任务
        results = await asyncio.gather(*tasks)

        for i, result in enumerate(results):
            if not result.is오답():
                print(f"设备{i+1}데이터: {result.registers}")

    finally:
        client.close()

# 运行异步函数
asyncio.run(read_device_데이터())

三、minimalmodbus快速上手

3.1 基础使用

import minimalmodbus
import serial

# 创建仪器对象
instrument = minimalmodbus.Instrument(
    port='/dev/ttyUSB0',   # 시리얼 포트设备
    slave주소=1,        # 슬레이브 주소
    mode='rtu'             # 프로토콜모드
)

# 설정시리얼 포트 매개변수
instrument.serial.baudrate = 9600
instrument.serial.bytesize = 8
instrument.serial.parity = serial.PARITY_NONE
instrument.serial.stopbits = 1
instrument.serial.timeout = 1.0  # 秒

# 유지 레지스터 읽기 (FC03)
try:
    # 읽기单个레지스터（주소30001）
    temperature = instrument.read_register(
        register주소=0,  # 레지스터주소（0对应30001）
        functioncode=3,     # 기능 코드3：读유지 레지스터
        signed=True         # 有符号整数
    )
    print(f"온도: {temperature}°C")

    # 읽기多个레지스터
    values = instrument.read_registers(
        register주소=0,
        number_of_registers=5,
        functioncode=3
    )
    print(f"多个레지스터 값: {values}")

    # 단일 레지스터 쓰기 (FC06)
    instrument.write_register(
        register주소=10,
        value=100,
        functioncode=6
    )

except Exception as e:
    print(f"통신错误: {e}")

3.2 高级功能

# 自定义바이트顺序（大端/小端）
instrument.byteorder = minimalmodbus.BYTEORDER_BIG

# 设置디버깅모드（显示原始통신데이터）
instrument.debug = True

# 使用不同的精度和缩放因子
def read_temperature_with_precision():
    # 假设온도레지스터存储的是实际值*10
    raw_value = instrument.read_register(0, 3)
    temperature = raw_value / 10.0
    return temperature

# 批量읽기优化
def batch_read_registers(주소es):
    """批量읽기多个레지스터，优化통신效率"""
    results = {}
    for addr in 주소es:
        try:
            value = instrument.read_register(addr, 3)
            results[addr] = value
        except Exception as e:
            results[addr] = f"错误: {e}"
    return results

四、uModbus现代化개발

4.1 异步服务器예제

import asyncio
from umodbus.server.tcp import RequestHandler, get_server
from umodbus.conf import Modbus데이터Type

# 定义데이터存储
데이터_store = {
    'coils': {0: False, 1: True, 2: False},
    'discrete_inputs': {0: True, 1: False},
    'holding_registers': {
        0: 100,
        1: 200,
        2: 300
    },
    'input_registers': {
        0: 400,
        1: 500
    }
}

async def handle_request(slave_id, function_code, 주소, 수량):
    """处理Modbus请求"""
    if function_code == 1:  # 读코일
        return [데이터_store['coils'].get(addr, False) 
                for addr in range(주소, 주소 + 수량)]

    elif function_code == 3:  # 读유지 레지스터
        return [데이터_store['holding_registers'].get(addr, 0)
                for addr in range(주소, 주소 + 수량)]

    elif function_code == 6:  # 写单个레지스터
        # 在实际적용中，这里会更新데이터存储
        return True

    return None

async def main():
    # 创建TCP服务器
    server = get_server(
        RequestHandler,
        데이터_store=데이터_store,
        request_handler=handle_request
    )

    # 启动服务器
    await server.serve_forever()

# 运行服务器
asyncio.run(main())

五、实际적용案例

5.1 工业온도监控系统

import time
from datetime import datetime
import json
from pymodbus.client import ModbusTcpClient

class TemperatureMonitor:
    def __init__(self, device_configs):
        """
        初始化온도监控系统
        device_configs: 设备설정列表
        """
        self.devices = []
        for config in device_configs:
            client = ModbusTcpClient(
                host=config['ip'],
                port=config.get('port', 502),
                timeout=config.get('timeout', 3)
            )
            self.devices.append({
                'client': client,
                'name': config['name'],
                'slave_id': config.get('slave_id', 1),
                'registers': config['registers']
            })

        self.데이터_history = []

    def read_all_temperatures(self):
        """읽기所有设备的온도데이터"""
        readings = []

        for device in self.devices:
            client = device['client']

            if not client.connect():
                print(f"无法연결到设备: {device['name']}")
                continue

            try:
                device_readings = {}
                for reg_name, reg_config in device['registers'].items():
                    result = client.read_holding_registers(
                        주소=reg_config['주소'],
                        count=reg_config.get('count', 1),
                        slave=device['slave_id']
                    )

                    if not result.is오답():
                        raw_value = result.registers[0]
                        # 적용转换公式（如除以10得到实际온도）
                        actual_value = raw_value / reg_config.get('scale', 1.0)
                        device_readings[reg_name] = {
                            'raw': raw_value,
                            'actual': actual_value,
                            'unit': reg_config.get('unit', '°C')
                        }

                readings.append({
                    'device': device['name'],
                    'timestamp': datetime.now().isoformat(),
                    'readings': device_readings
                })

            except Exception as e:
                print(f"읽기设备{device['name']}时出错: {e}")
            finally:
                client.close()

        # 저장到历史记录
        self.데이터_history.extend(readings)

        # 只保留最近1000条记录
        if len(self.데이터_history) > 1000:
            self.데이터_history = self.데이터_history[-1000:]

        return readings

    def check_alarms(self, readings):
        """检查报警条件"""
        alarms = []

        for device_reading in readings:
            for sensor_name, sensor_데이터 in device_reading['readings'].items():
                value = sensor_데이터['actual']

                # 简单报警逻辑
                if value > 80:  # 高温报警
                    alarms.append({
                        'device': device_reading['device'],
                        'sensor': sensor_name,
                        'value': value,
                        'type': 'high_temperature',
                        'message': f"온도过高: {value}°C",
                        'timestamp': device_reading['timestamp']
                    })
                elif value < 10:  # 低温报警
                    alarms.append({
                        'device': device_reading['device'],
                        'sensor': sensor_name,
                        'value': value,
                        'type': 'low_temperature',
                        'message': f"온도过低: {value}°C",
                        'timestamp': device_reading['timestamp']
                    })

        return alarms

    def export_데이터(self, filename='temperature_데이터.json'):
        """导出데이터到JSON文件"""
        with 열기(filename, 'w', encoding='utf-8') as f:
            json.dump({
                'history': self.데이터_history,
                'export_time': datetime.now().isoformat()
            }, f, ensure_ascii=False, indent=2)

        print(f"데이터已导出到: {filename}")

# 使用예제
if __name__ == "__main__":
    # 设备설정
    device_configs = [
        {
            'name': '锅炉1',
            'ip': '192.168.1.101',
            'registers': {
                'temperature': {'주소': 0, 'scale': 10.0, 'unit': '°C'},
                'pressure': {'주소': 1, 'scale': 100.0, 'unit': 'kPa'}
            }
        },
        {
            'name': '反应釜2',
            'ip': '192.168.1.102',
            'registers': {
                'temperature': {'주소': 0, 'scale': 10.0, 'unit': '°C'},
                'ph_value': {'주소': 1, 'scale': 100.0, 'unit': 'pH'}
            }
        }
    ]

    # 创建监控器
    monitor = TemperatureMonitor(device_configs)

    # 监控循环
    try:
        while True:
            print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 开始읽기데이터...")

            # 읽기데이터
            readings = monitor.read_all_temperatures()

            # 显示데이터
            for reading in readings:
                print(f"\n设备: {reading['device']}")
                for sensor, 데이터 in reading['readings'].items():
                    print(f"  {sensor}: {데이터['actual']} {데이터['unit']}")

            # 检查报警
            alarms = monitor.check_alarms(readings)
            if alarms:
                print("\n⚠️ 报警信息:")
                for alarm in alarms:
                    print(f"  {alarm['message']}")

            # 每5秒읽기一次
            time.sleep(5)

    except KeyboardInterrupt:
        print("\n监控정지")
        # 导出데이터
        monitor.export_데이터()

5.2 데이터可视化界面

```python import tkinter as tk from tkinter import ttk import threading import time from datetime import datetime from pymodbus.client import ModbusTcpClient class ModbusMonitorGUI: def init(self, root): self.root = root self.root.title("Modbus设备监控系统") self.root.geometry("800x600")

    # 设备설정
    self.devices = [
        {'name': 'PLC_1', 'ip': '192.168.1.100', 'port': 502, 'slave_id': 1},
        {'name': 'PLC_2', 'ip': '192.168.1.101', 'port': 502, 'slave_id': 2}
    ]

    # 데이터存储
    self.데이터 = {}
    for device in self.devices:
        self.데이터[device['name']] = {
            'registers': {},
            'last_update': None,
            'status': '离线'
        }

    # 创建界面
    self.setup_ui()

    # 启动监控线程
    self.monitoring = True
    self.monitor_thread = threading.Thread(target=self.monitor_devices)
    self.monitor_thread.daemon = True
    self.monitor_thread.시작()

def setup_ui(self):
    """设置用户界面"""
    # 创建主框架
    main_frame = ttk.Frame(self.root, padding="10")
    main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))

    # 设备状态区域
    status_frame = ttk.LabelFrame(main_frame, text="设备状态", padding="10")
    status_frame.grid(row=0, column=0, sticky=(tk.W, tk.E), pady=(0, 10))

    self.status_labels = {}
    for i, device in enumerate(self.devices):
        label = ttk.Label(status_frame, text=f"{device['name']}: 离线")
        label.grid(row=i, column=0, sticky=tk.W, padx=5, pady=2)
        self.status_labels[device['name']] = label

    # 데이터展示区域
    데이터_frame = ttk.LabelFrame(main_frame, text="实时데이터", padding="10")
    데이터_frame.grid(row=1, column=0, sticky=(tk.W, tk.E, tk.N, tk.S), pady=(0, 10))

    # 创建Treeview显示데이터
    columns = ('设备', '레지스터', '주소', '值', '时间')
    self.tree = ttk.Treeview(데이터_frame, columns=columns, show='headings', height=15)

    for col in columns:
        self.tree.heading(col, text=col)
        self.tree.column(col, width=100)

    # 추가滚动条
    scrollbar = ttk.Scrollbar(데이터_frame, orient=tk.VERTICAL, command=self.tree.yview)
    self.tree.configure(yscrollcommand=scrollbar.set)

    self.tree.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
    scrollbar.grid(row=0, column=1, sticky=(tk.N, tk.S))

    # 控制버튼区域
    button_frame = ttk.Frame(main_frame)
    button_frame.grid(row=2, column=0, sticky=(tk.W, tk.E))

    ttk.Button(button_frame, text="새로고침데이터", command=self.manual_refresh).pack(side=tk.LEFT, padx=5)
    ttk.Button(button_frame, text="导出데이터", command=self.export_데이터).pack(side=tk.LEFT, padx=5)
    ttk.Button(button_frame, text="정지监控", command=self.stop_monitoring).pack(side=tk.LEFT, padx=5)

    # 설정网格权重
    main_frame.columnconfigure(0, weight=1)
    main_frame.rowconfigure(1, weight=1)
    데이터_frame.columnconfigure(0, weight=1)
    데이터_frame.rowconfigure(0, weight=1)

def read_device_데이터(self, device_config):
    """읽기单个设备的데이터"""
    client = ModbusTcpClient(
        host=device_config['ip'],
        port=device_config['port'],
        timeout=3
    )

    try:
        if client.connect():
            # 읽기레지스터0-19
            result = client.read_holding_registers(
                주소=0,
                count=20,
                slave=device_config['slave_id']
            )

            if not result.is오답():
                return {
                    'status': '在线',
                    '데이터': result.registers,
                    'timestamp': datetime.now()
                }
            else:
                return {
                    'status': '읽기错误',
                    '데이터': None,
                    'timestamp': datetime.now()
                }
        else:
            return {
                'status': '연결失败',
                '데이터': None,
                'timestamp': datetime.now()
            }

    except Exception as e:
        return {
            'status': f'异常: {str(e)}',
            '데이터': None,
            'timestamp': datetime.now()
        }
    finally:
        client.close()

def monitor_devices(self):
    """监控设备线程"""
    while self.monitoring:
        for device in self.devices:
            result = self.read_device_데이터(device)

            # 更新데이터存储
            self.데이터[device['name']].update({
                'registers': result['데이터'] or {},
                'last_update': result['timestamp'],
                'status': result['status']
            })

            # 更新UI（需要在主线程中执行）
            self.root.after(0, self.update_ui, device['name'], result)

        # 每2秒更新一次
        time.sleep(2)

def update_ui(self, device_name, result):
    """更新用户界面"""
    # 更新状态标签
    status_text = f"{device_name}: {result['status']}"
    if result['timestamp']:
        status_text += f" ({result['timestamp'].strftime('%H:%M:%S')})"

    self.status_labels[device_name].config(text=status_text)

    # 更新颜色
    if result['status'] == '在线':
        self.status_labels[device_name].config(foreground='green')
    elif '错误' in result['status'] or '失败' in result['status']:
        self.status_labels[device_name].config(foreground='red')
    else:
        self.status_labels[device_name].config(foreground='orange')

    # 更新데이터表格
    if result['데이터']:
        # 清除旧데이터
        for item in self.tree.get_children():
            if self.tree.item(item)['values'][0] == device_name:
                self.tree.delete(item)

        # 추가新데이터
        for i, value in enumerate(result['데이터']):
            self.tree.insert('', 'end', values=(
                device_name,
                f'400{i+1:02d}',
                i,
                value,
                result['timestamp'].strftime('%H:%M:%S') if result['timestamp'] else ''
            ))

def manual_refresh(self):
    """手动새로고침데이터"""
    for device in self.devices:
        result = self.read_device_데이터(device)
        self.update_ui(device['name'], result)

def export_데이터(self):
    """导出데이터到文件"""
    import json
    from datetime import datetime

    export_데이터 = {
        'export_time': datetime.now().isoformat(),
        'devices': {}
    }

    for device_name, device_데이터 in self.데이터.items():
        export_데이터['devices'][device_name] = {
            'status': device_데이터['status'],
            'last_update': device_데이터['last_update'].isoformat() if device_데이터['last_update'] else None,
            'registers': device_데이터['registers']
        }

    filename = f"modbus_데이터_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    with 열기(filename, 'w', encoding='utf-8') as f:
        json.dump(export_데이터, f, ensure_ascii=False, indent=2)

    # 显示导出成功消息
    from tkinter import messagebox
    messagebox.showinfo("导出成功", f"데이터已导出到: {filename}")

def stop_monitoring(self):
    """정지监控"""
    self.monitoring = False
    self.root.quit()

运行GUI적용

if name == "main": root = tk.Tk() app = ModbusMonitorGUI(root) root.mainloop()