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

引言：Warum wählenPython进行Modbus开发？

在Das industrielle Internet der Dinge和自动化领域，Python凭借其简洁的语法、丰富的库生态Das System和强大的社区支持，已经成为Entwickler的首选语言之一。对于ModbusVereinbarung开发，Python提供了多个成熟的开源库，能够满足从简单AusrüstungKommunikationzu复杂工业Das System集成的各种需求。

本文将全面介绍Python中主流的Modbus库，包括pymodbus、minimalmodbus、uModbus等，通过实际Beispiel Code和最佳实践，帮助您快速掌握Python Modbus开发的核心技能。

一、主流Python Modbus库对比

1.1 pymodbus：Funktion最全面的Auswahl

特点：
– 支持Modbus TCP、RTU、ASCII等多种Vereinbarung
– 同时支持客户端（Hauptstation）和Der Server（Von der Station）模式
– Asynchron/同步两种编程模式
– 活跃的社区和持续更新

适用Szene：
– 复杂的工业控制Das System
– 需要同时作为Hauptstation和Von der Station的Szene
– 高性能要求的应用

1.2 minimalmodbus：轻Quantität级解决方案

特点：
– 代码简洁，易于学习和使用
– 专注于Modbus RTUVereinbarung
– 依赖少，部署简单
– 适合EingebettetDas System和资源受限环境

适用Szene：
– 简单Die AusrüstungKommunikation
– EingebettetDas System和单板Berechnung机
– 快速原型开发

1.3 uModbus：现代化设计

特点：
– 现代化的API设计
– 支持Asynchron编程
– 良好的Typus提示
– 模块化架构

适用Szene：
– 需要Asynchron处理的应用
– 大型项目开发
– 需要良好代码维护性的Szene

二、pymodbus深度Analyse

2.1 安装和基础配置

# 安装pymodbus
pip install pymodbus

# 或者安装包含Asynchron支持的版本
pip install pymodbus[asyncio]

2.2 Modbus TCP客户端Beispiel

from pymodbus.client import ModbusTcpClient

# 创建TCP客户端
client = ModbusTcpClient(
    host='192.168.1.100',  # AusrüstungIPAdresse
    port=502,              # Modbus TCP端口
    timeout=3,             # 超时Zeit（秒）
    retries=3              # 重试次数
)

try:
    # VerbindungAusrüstung
    if client.connect():
        print("成功VerbindungzuModbusAusrüstung")

        # Lesen und halten Register（Adresse0，Anzahl10）
        result = client.read_holding_registers(address=0, count=10, slave=1)
        if not result.isError():
            print(f"Lesenzu的Registerwert: {result.registers}")
        else:
            print(f"Lesen gescheitert: {result}")

        # 写入单个保持Register.（Adresse5，值1234）
        result = client.write_register(address=5, value=1234, slave=1)
        if not result.isError():
            print("Register.写入成功")

        # Lesen Sie den Zustand der Spule（Adresse0，Anzahl8）
        result = client.read_coils(address=0, count=8, slave=1)
        if not result.isError():
            print(f"Zustand der Coil: {result.bits}")

finally:
    # SchließenVerbindung
    client.close()

2.3 Modbus RTU客户端Beispiel

from pymodbus.client import ModbusSerialClient

# 创建RTU客户端
client = ModbusSerialClient(
    method='rtu',          # VereinbarungTypus（rtu或ascii）
    port='/dev/ttyUSB0',   # Serielle Ausrüstung
    baudrate=9600,         # Porter Rate
    bytesize=8,            # Datenplatz
    parity='N',            # Prüfstand.（N:无, E:偶, O:奇）
    stopbits=1,            # Stoppen Sie Platz
    timeout=1              # 超时Zeit
)

try:
    if client.connect():
        print("成功VerbindungzuSerielle Ausrüstung")

        # 批QuantitätEingangsregister lesen
        result = client.read_input_registers(address=0, count=20, slave=1)
        if not result.isError():
            data = result.registers
            # 处理Daten：假设前10个是Temperatur值（需要除以10）
            temperatures = [value/10.0 for value in data[:10]]
            print(f"TemperaturDaten: {temperatures}")

finally:
    client.close()

2.4 Asynchron客户端Beispiel

import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def read_device_data():
    # 创建Asynchron客户端
    client = AsyncModbusTcpClient(
        host='192.168.1.100',
        port=502
    )

    await client.connect()

    try:
        # 同时Lesen多Ein Gerät的Daten
        tasks = []
        for slave_id in [1, 2, 3]:
            task = client.read_holding_registers(
                address=0, 
                count=10, 
                slave=slave_id
            )
            tasks.append(task)

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

        for i, result in enumerate(results):
            if not result.isError():
                print(f"Ausrüstung{i+1}Daten: {result.registers}")

    finally:
        client.close()

# BetriebAsynchron函数
asyncio.run(read_device_data())

三、minimalmodbus快速上手

3.1 基础使用

import minimalmodbus
import serial

# 创建仪器对象
instrument = minimalmodbus.Instrument(
    port='/dev/ttyUSB0',   # Serielle Ausrüstung
    slaveaddress=1,        # von der Station Adresse
    mode='rtu'             # Vereinbarung模式
)

# 配置Serienparameter
instrument.serial.baudrate = 9600
instrument.serial.bytesize = 8
instrument.serial.parity = serial.PARITY_NONE
instrument.serial.stopbits = 1
instrument.serial.timeout = 1.0  # 秒

# Lesen und halten Register
try:
    # Lesen单ein Register.（Adresse30001）
    temperature = instrument.read_register(
        registeraddress=0,  # Register-Adresse（0对应30001）
        functioncode=3,     # Funktionscode3：Lesen Sie das Register
        signed=True         # 有符Nr.整数
    )
    print(f"Temperatur: {temperature}°C")

    # Lesen多ein Register.
    values = instrument.read_registers(
        registeraddress=0,
        number_of_registers=5,
        functioncode=3
    )
    print(f"mehrere Registerwerte: {values}")

    # Schreiben in ein einzelnes Register
    instrument.write_register(
        registeraddress=10,
        value=100,
        functioncode=6
    )

except Exception as e:
    print(f"KommunikationFehler.: {e}")

3.2 高级Funktion

# Custom definiertDie Reihenfolge（大端/小端）
instrument.byteorder = minimalmodbus.BYTEORDER_BIG

# 设置Tests模式（显示原始KommunikationDaten）
instrument.debug = True

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

# 批QuantitätLesen优化
def batch_read_registers(addresses):
    """批QuantitätLesen多ein Register.，优化Kommunikation效率"""
    results = {}
    for addr in addresses:
        try:
            value = instrument.read_register(addr, 3)
            results[addr] = value
        except Exception as e:
            results[addr] = f"Fehler.: {e}"
    return results

四、uModbus现代化开发

4.1 AsynchronDer ServerBeispiel

import asyncio
from umodbus.server.tcp import RequestHandler, get_server
from umodbus.conf import ModbusDataType

# 定义Daten存储
data_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, address, quantity):
    """处理ModbusBitte"""
    if function_code == 1:  # 读线圈
        return [data_store['coils'].get(addr, False) 
                for addr in range(address, address + quantity)]

    elif function_code == 3:  # Lesen Sie das Register
        return [data_store['holding_registers'].get(addr, 0)
                for addr in range(address, address + quantity)]

    elif function_code == 6:  # Schreiben eines einzelnen Registers
        # 在实际应用中，这里会更新Daten存储
        return True

    return None

async def main():
    # 创建TCPDer Server
    server = get_server(
        RequestHandler,
        data_store=data_store,
        request_handler=handle_request
    )

    # 启动Der Server
    await server.serve_forever()

# BetriebDer Server
asyncio.run(main())

五、实际应用案例

5.1 工业Temperatur监控Das System

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

class TemperatureMonitor:
    def __init__(self, device_configs):
        """
        初始化Temperatur监控Das System
        device_configs: Ausrüstung配置列表
        """
        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.data_history = []

    def read_all_temperatures(self):
        """Lesen所有Ausrüstung的TemperaturDaten"""
        readings = []

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

            if not client.connect():
                print(f"无法VerbindungzuAusrüstung: {device['name']}")
                continue

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

                    if not result.isError():
                        raw_value = result.registers[0]
                        # 应用转换公式（如除以10得zu实际Temperatur）
                        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"LesenAusrüstung{device['name']}时出错: {e}")
            finally:
                client.close()

        # 保存zu历史记录
        self.data_history.extend(readings)

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

        return readings

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

        for device_reading in readings:
            for sensor_name, sensor_data in device_reading['readings'].items():
                value = sensor_data['actual']

                # 简单报警逻辑
                if value > 80:  # 高温报警
                    alarms.append({
                        'device': device_reading['device'],
                        'sensor': sensor_name,
                        'value': value,
                        'type': 'high_temperature',
                        'message': f"Temperatur过高: {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"Temperatur过低: {value}°C",
                        'timestamp': device_reading['timestamp']
                    })

        return alarms

    def export_data(self, filename='temperature_data.json'):
        """ExportierenDatenzuJSON文件"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump({
                'history': self.data_history,
                'export_time': datetime.now().isoformat()
            }, f, ensure_ascii=False, indent=2)

        print(f"Daten已Exportierenzu: {filename}")

# 使用Beispiel
if __name__ == "__main__":
    # Ausrüstung配置
    device_configs = [
        {
            'name': '锅炉1',
            'ip': '192.168.1.101',
            'registers': {
                'temperature': {'address': 0, 'scale': 10.0, 'unit': '°C'},
                'pressure': {'address': 1, 'scale': 100.0, 'unit': 'kPa'}
            }
        },
        {
            'name': '反应釜2',
            'ip': '192.168.1.102',
            'registers': {
                'temperature': {'address': 0, 'scale': 10.0, 'unit': '°C'},
                'ph_value': {'address': 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')}] 开始Daten lesen...")

            # Daten lesen
            readings = monitor.read_all_temperatures()

            # 显示Daten
            for reading in readings:
                print(f"\nAusrüstung: {reading['device']}")
                for sensor, data in reading['readings'].items():
                    print(f"  {sensor}: {data['actual']} {data['unit']}")

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

            # 每5秒Lesen一次
            time.sleep(5)

    except KeyboardInterrupt:
        print("\n监控Stopp!")
        # ExportierenDaten
        monitor.export_data()

5.2 Daten可视化界面

“`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(“ModbusAusrüstung监控Das System”)
self.root.geometry(“800×600”)

    # Ausrüstung配置
    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}
    ]

    # Daten存储
    self.data = {}
    for device in self.devices:
        self.data[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.start()

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))

    # AusrüstungStatus区域
    status_frame = ttk.LabelFrame(main_frame, text="AusrüstungStatus", 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

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

    # 创建Treeview显示Daten
    columns = ('Ausrüstung', 'Register.', 'Adresse', '值', 'Zeit')
    self.tree = ttk.Treeview(data_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(data_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))

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

    ttk.Button(button_frame, text="ErfrischenDaten", command=self.manual_refresh).pack(side=tk.LEFT, padx=5)
    ttk.Button(button_frame, text="ExportierenDaten", command=self.export_data).pack(side=tk.LEFT, padx=5)
    ttk.Button(button_frame, text="Stopp!监控", command=self.stop_monitoring).pack(side=tk.LEFT, padx=5)

    # 配置网格权重
    main_frame.columnconfigure(0, weight=1)
    main_frame.rowconfigure(1, weight=1)
    data_frame.columnconfigure(0, weight=1)
    data_frame.rowconfigure(0, weight=1)

def read_device_data(self, device_config):
    """Lesen单Ein Gerät的Daten"""
    client = ModbusTcpClient(
        host=device_config['ip'],
        port=device_config['port'],
        timeout=3
    )

    try:
        if client.connect():
            # LesenRegister.0-19
            result = client.read_holding_registers(
                address=0,
                count=20,
                slave=device_config['slave_id']
            )

            if not result.isError():
                return {
                    'status': '在线',
                    'data': result.registers,
                    'timestamp': datetime.now()
                }
            else:
                return {
                    'status': 'LesenFehler.',
                    'data': None,
                    'timestamp': datetime.now()
                }
        else:
            return {
                'status': 'Verbindung失败',
                'data': None,
                'timestamp': datetime.now()
            }

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

def monitor_devices(self):
    """监控Ausrüstung线程"""
    while self.monitoring:
        for device in self.devices:
            result = self.read_device_data(device)

            # 更新Daten存储
            self.data[device['name']].update({
                'registers': result['data'] 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):
    """更新用户界面"""
    # 更新StatusKennzeichnung
    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 'Fehler.' in result['status'] or '失败' in result['status']:
        self.status_labels[device_name].config(foreground='red')
    else:
        self.status_labels[device_name].config(foreground='orange')

    # 更新Daten表格
    if result['data']:
        # Beseitigung旧Daten
        for item in self.tree.get_children():
            if self.tree.item(item)['values'][0] == device_name:
                self.tree.delete(item)

        # 添加新Daten
        for i, value in enumerate(result['data']):
            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):
    """手动ErfrischenDaten"""
    for device in self.devices:
        result = self.read_device_data(device)
        self.update_ui(device['name'], result)

def export_data(self):
    """ExportierenDatenzu文件"""
    import json
    from datetime import datetime

    export_data = {
        'export_time': datetime.now().isoformat(),
        'devices': {}
    }

    for device_name, device_data in self.data.items():
        export_data['devices'][device_name] = {
            'status': device_data['status'],
            'last_update': device_data['last_update'].isoformat() if device_data['last_update'] else None,
            'registers': device_data['registers']
        }

    filename = f"modbus_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(export_data, f, ensure_ascii=False, indent=2)

    # 显示Exportieren成功消息
    from tkinter import messagebox
    messagebox.showinfo("Exportieren成功", f"Daten已Exportierenzu: {filename}")

def stop_monitoring(self):
    """Stopp!监控"""
    self.monitoring = False
    self.root.quit()

BetriebGUI应用

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