Python Modbus库完全ガイド：入門からマスターまで。

引言：No licensing fees, open standardPython进行Modbus开发？

在産業用IoT和自動化领域，Python凭借其简洁的语法、丰富的库生态系统和强大的社区サポート，已经成为開発者たちは的首选语言之一。对于Modbusプロトコル开发，Python提供了多个成熟的オープンソース·ライブラリ，能够满足从简单デバイス通信到复杂工业システム統合的各种需求。 本記事将包括的介绍Python中主流的Modbus库，包括pymodbus、minimalmodbus、uModbus等，経由实际コード例和ベストプラクティス，帮助您スピードマスター。Python Modbus开发的核心技能。

一、主流Python Modbus库对比

1.1 pymodbus：功能最包括的的選択

特点： - サポートModbus TCP、RTU、ASCII等多种协议 - 同时サポートクライアント側は（master）和サーバー·サーバー（slave）モード - asynchronous/同步两种プログラミングモード - 活跃的社区和継続的な更新 適用可能なシーン： - 复杂的工业控制系统 - 需要同时作为マスター和スレーブ的场景 - 高いパフォーマンスを要求的应用

1.2 minimalmodbus：轻量级ソリューション

特点： - 代码简洁，易于学习和使用 - 专注于Modbus RTUProtocol - 依赖少，部署简单 - 适合エンベデッド·イン系统和リソース受限环境 適用可能なシーン： - 简单的デバイス通信 - エンベデッド·イン系统和单板計算机 - 快速原型开发

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',  # deviceIPaddress
    port=502,              # Modbus TCPポート
    timeout=3,             # タイムアウト時間（秒）
    retries=3              # リトライ回数
)

try:
    # 连機器の接続
    if client.connect():
        print("成功连接到Modbusdevice")

        # Read and hold register（address0，Quantity10）
        result = client.read_holding_registers(address=0, count=10, slave=1)
        if not result.isError():
            print(f"読み取り到的Register value: {result.registers}")
        else:
            print(f"Read failed: {result}")

        # 写入单レジスタを保持する。（address5，值1234）
        result = client.write_register(address=5, value=1234, slave=1)
        if not result.isError():
            print("レジスタ写入成功")

        # Read coil status（address0，Quantity8）
        result = client.read_coils(address=0, count=8, slave=1)
        if not result.isError():
            print(f"Coil status: {result.bits}")

finally:
    # 接続を閉じる
    client.close()

2.3 Modbus RTUクライアント側はサンプル

from pymodbus.client import ModbusSerialClient

# 作成RTUクライアント側は
client = ModbusSerialClient(
    method='rtu',          # 协议タイプ（rtu或ascii）
    port='/dev/ttyUSB0',   # Serial Port Device
    baudrate=9600,         # Baud rate
    bytesize=8,            # data bit
    parity='N',            # check digit（N:无, E:偶, O:奇）
    stopbits=1,            # stop bit
    timeout=1              # タイムアウト時間
)

try:
    if client.connect():
        print("成功连接到シリアルデバイス")

        # 批量入力レジスタの読み取り
        result = client.read_input_registers(address=0, count=20, slave=1)
        if not result.isError():
            data = result.registers
            # 処理データ：假设前10个是温度值（需要除以10）
            temperatures = [value/10.0 for value in data[:10]]
            print(f"温度データ: {temperatures}")

finally:
    client.close()

2.4 异步クライアント側はサンプル

import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def read_device_data():
    # 作成异步クライアント側は
    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(
                address=0, 
                count=10, 
                slave=slave_id
            )
            tasks.append(task)

        # 并行执行すべての読み取り任务
        results = await asyncio.gather(*tasks)

        for i, result in enumerate(results):
            if not result.isError():
                print(f"device{i+1}data: {result.registers}")

    finally:
        client.close()

# 运行异步函数
asyncio.run(read_device_data())

三、minimalmodbus快速上手

3.1 基础使用

import minimalmodbus
import serial

# 作成仪器对象
instrument = minimalmodbus.Instrument(
    port='/dev/ttyUSB0',   # Serial Port Device
    slaveaddress=1,        # slave address
    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  # 秒

# Read and hold register
try:
    # 読み取り单1つのレジスタ（address30001）
    temperature = instrument.read_register(
        registeraddress=0,  # register address（0对应30001）
        functioncode=3,     # function code3：read holding registers
        signed=True         # 記号付き。整数
    )
    print(f"temperature: {temperature}°C")

    # 読み取り多1つのレジスタ
    values = instrument.read_registers(
        registeraddress=0,
        number_of_registers=5,
        functioncode=3
    )
    print(f"多个Register value: {values}")

    # Write to a single register
    instrument.write_register(
        registeraddress=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(addresses):
    """バッチ読み取り多1つのレジスタ，优化通信效率"""
    results = {}
    for addr in addresses:
        try:
            value = instrument.read_register(addr, 3)
            results[addr] = value
        except Exception as e:
            results[addr] = f"Error: {e}"
    return results

四、uModbus现代化开发

4.1 异步サーバー·サーバーサンプル

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

# 定义データストア
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):
    """処理ModbusRequest"""
    if function_code == 1:  # リードコイルを読む
        return [data_store['coils'].get(addr, False) 
                for addr in range(address, address + quantity)]

    elif function_code == 3:  # read holding registers
        return [data_store['holding_registers'].get(addr, 0)
                for addr in range(address, address + quantity)]

    elif function_code == 6:  # write single register
        # 在实际应用中，这里会更新データストア
        return True

    return None

async def main():
    # 作成TCPServer
    server = get_server(
        RequestHandler,
        data_store=data_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.data_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(
                        address=reg_config['address'],
                        count=reg_config.get('count', 1),
                        slave=device['slave_id']
                    )

                    if not result.isError():
                        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.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"温度过高: {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_data(self, filename='temperature_data.json'):
        """导出データ到JSON文件"""
        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"データ已导出To: {filename}")

# 使用例の例
if __name__ == "__main__":
    # デバイス設定
    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')}] 読み始めました取データ...")

            # Read switch status
            readings = monitor.read_all_temperatures()

            # 表示データ
            for reading in readings:
                print(f"\ndevice: {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秒読み取り一次
            time.sleep(5)

    except KeyboardInterrupt:
        print("\nモニタリング停止")
        # 导出データ
        monitor.export_data()

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

    # デバイスステータスエリア
    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

    # データ表示エリア
    data_frame = ttk.LabelFrame(main_frame, text="リアルタイムデータ", padding="10")
    data_frame.grid(row=1, column=0, sticky=(tk.W, tk.E, tk.N, tk.S), pady=(0, 10))

    # 作成Treeview表示データ
    columns = ('device', 'register', 'address', '值', 'time')
    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))

    # 控制ボタンエリア
    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_data).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)
    data_frame.columnconfigure(0, weight=1)
    data_frame.rowconfigure(0, weight=1)

def read_device_data(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(
                address=0,
                count=20,
                slave=device_config['slave_id']
            )

            if not result.isError():
                return {
                    'status': 'オンライン',
                    'data': result.registers,
                    'timestamp': datetime.now()
                }
            else:
                return {
                    'status': '読み取りエラー',
                    'data': None,
                    'timestamp': datetime.now()
                }
        else:
            return {
                'status': '接続の失敗',
                '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):
    """モニタリングデバイス线程"""
    while self.monitoring:
        for device in self.devices:
            result = self.read_device_data(device)

            # 更新データストア
            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):
    """更新用户界面"""
    # 更新ステータスタグ
    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 'Error' 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['data']:
        # クリア旧データ
        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['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):
    """手动リフレッシュデータ"""
    for device in self.devices:
        result = self.read_device_data(device)
        self.update_ui(device['name'], result)

def export_data(self):
    """导出データ到文件"""
    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)

    # 表示导出成功メッセージ
    from tkinter import messagebox
    messagebox.showinfo("导出成功", f"データ已导出To: {filename}")

def stop_monitoring(self):
    """停止モニタリング"""
    self.monitoring = False
    self.root.quit()

runGUI应用

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