Python Modbusライブラリの完全ガイド：初心者からマスターまで

はじめに：Modbus開発にPythonを選ぶ理由

産業用モノのインターネットとオートメーションの世界では、Pythonはその簡潔な構文、豊富なライブラリエコシステム、強力なコミュニティサポートにより、開発者のための最初の言語の1つになっています。Modbusプロトコル開発のために、Pythonは単純なデバイスCommunicationから複雑な産業システム統合まで、あらゆるニーズに対応できる成熟したオープンソースライブラリを提供しています。 この記事では、pymodbus、minimalmodbus、uModbusなどのPythonの主要なModbusライブラリを包括的に紹介し、実際のコード例とベストプラクティスを使用して、Python Modbus開発のコアスキルをすばやく習得するのに役立ちます。

1.主流のPython Modbusライブラリの比較

1.1 pymodbus：最も包括的なオプション

特徴：- Modbus TCP、RTU、ASCIIなどの多種プロトコルをサポートする - クライアント（マスター）モードとサーバ（スレーブ）モードのサポート - 非同期/同期の2つのプログラミングモード - 活発なコミュニティと継続的な更新適用シーン：- 複雑な産業用制御システム - マスターとスレーブの両方が必要です。 - 高性能要件の適用

1.2 minimalmodbus：軽量ソリューション

特徴：- コードは簡潔で習得しやすく使いやすい - Modbus RTUプロトコル - 依存性が少なく、導入が簡単 - 组み込みシステムおよびリソースのに最适適用シーン：- 簡単なデバイスCommunication - 組込みシステムとシングルボードコンピュータ - ラピッドプロトタイピング

1.3 UModbus：モダンデザイン

特徴：- API設計の近代化 - 非同期プログラミングのサポート - 良いタイプのヒント。 - モジュラーアーキテクチャ適用シーン：- 非同期処理が必要なアプリケーション - 大規模プロジェクトの開発 - 良いコードメンテナンスが必要なシナリオ

2. pymodbusの詳細解析

2.1インストールと基本構成

Pymodbusのインストール
pip install pymodbus

#または非同期サポートを含むバージョンのインストール
pip install pymodbus[asyncio]

2.2 Modbus TCPクライアントの例

from pymodbus.client import modbusTcpClient

#TCPクライアントの作成
client = ModbusTcpクライアント
    host='192.16 8.1.100'，#デバイスのIPアドレス
    ポート= 50 2              # Modbus TCPポート
    timeout=3             #タイムアウト（秒）
    retries=3              #再試行回数
)

try
    #接続デバイス
    if client.connect。
        print“Modbusデバイスに正常に接続”

        #読み込みホールドレジスタ（アドレス0、数10）
        result = client.read_holding_registers (address=0, count=10, slave=1)
        if not result.isError
            print f"読み取られたレジスタ値{result.registers}"
        else nParticle
            print f"の読み取りに失敗しました{result}"

        #単一のホールドレジスタ（アドレス5、値1234）に書き込む
        result = client.write_register address=5、value=1234、slave=1
        if not result.isError
            print“レジスタ書き込みに成功しました”

        #コイルの読み込みアドレス0、Quantity8
        result = client.read_coils (address=0, count=8, slave=1)
        if not result.isError
            print f"コイル{.bits}"

Finally
    #接続を閉じる
    client.close

2.3 Modbus RTUクライアントの例

pymodbus.clientからmodbus SerialClientをインポート

RTUクライアントの作成
= ModbusSerialClient
    method ='rtu'、          プロトコルの種類（RTUまたはASCI I）
    port ='/dev/ttyUSB0'   #シリアルデバイス
    ボーレート= 96 00         #ボーレート
    besize =8            #データのビット
    parity ='N'            #チェックビットNなし、E偶、O奇
    stopbits=1、            #ストップ·ポイント
    タイムアウト=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 = [data [10]の値/10.0]
            print f"temperatureデータ{temperatures}"

Finally
    client.close

2.4非同期クライアントの例

asyncioのインポート
from pymodbus.client import AsyncModbusTcpClient

async defread_device_data
    #非同期クライアントの作成
    client = AsyncModbusTcpClient
        host ='192.16 8.1.100'，
        ポート= 50 2
    )

    await.

    try
        複数のデバイスから同時にデータを読み取る
        タグ= []
        for slave_id in [1 2 3]
            task = client.read_hing_registers
                address=0、 
                カウント=10。 
                slave=slave_id
            )
            tasks.append task

        すべての読み取りタスクを並列に実行
        rets = await asyncio.gather *tasks

        for i result in enumerate results
            if not result.isError
                print f"デバイス{i+1}データ{result.registers}"

    Finally
        client.close

#非同期関数の実行
asyncio.run__data

3.ミニマリズムバスの高速化

3.1基本的な利用

minimalmodbusのインポート
シリーズのインポート

#楽器の作成
instr = minimalmod.Instrument
    port ='/dev/ttyUSB0'   #シリアルデバイス
    slaveaddress=1        #駅の住所
    mode ='rtu'             #プロトコルモデル
)

#シリアルパラメータの設定
instrument.serial.baudrate = 96 00
instrument.serial.bytesize = 8
instrument.serial.parity = serial.PARITY_NONE
instrument.serial.stopbits = 1
instrument.serial.timeout = 1.0 # 秒

#リード保持レジスタ
try
    1つのレジスタを読み込む（アドレス300 01）
    temperature = instrument.read_register
        registeraddress=0 #レジスタ·アドレス0は30001に対応
        functioncode=3、     #3：リードホールドレジスタ
        signed=True         #符号付きの整数
    )
    print f"temperature{temperature}°C"

    #複数のレジスタを読み込む
    values = www.example.com_registers instrument.read
        register =0
        number_of_registers =5
        関数コード=3
    )
    print f"複数のレジスタ値{values}"

    #単一レジスタへの書き込み
    instrument.write_register
        registeraddress=10
        value=100
        関数コード=6
    )

exException as e
    print f"Communicationエラー {e}"

3.2高度な機能

#カスタムバイト順序（ビッグエンド/リトルエンド）
instrument.byteorder = minimalmodbus. BYTEORDER_BIG

#デバッグモードを設定する生のCommunicationデータを表示する
instr.= True

#異なる精度とスケール係数を使用する
read_temperature_with_precision
    #temperatureレジスタが実際の値を格納すると仮定 *10
    raw_value = instrument.read_register 0 3
    temperature = raw_value/ 10.0
    temperatureに戻る

#バッチ読み取り最適化
batch_read_registers
    “のレジスタを一括読み込み、通信効率を最する”
    結果= {}
    addr in addresses
        try
            value = instrument.read_register 3
            results[addr] = value
        exException as e
            rets [] = f"エラー {e}"
    Return Results

uModbusの近代化

4.1非同期サーバの例

asyncioのインポート
from umod.server.timport 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 - 30 0
    },
    'input_registers'{
        0：400です
        500から1
    }
}

async le_request slave_id function_code quantity
    “Modbusリクエストの処理”“”
    if function_code == 1 #リードコイル
        return [data_store [coils '].get 
                for addr in range address address + quantity]

    elif function_code == 3 #読み込みホールドレジスタ
        return [data_store ['holding_registers'].get addr 0
                for addr in range address address + quantity]

    elif function_code == 6 #1つのレジスタに书き込み
        #実際のアプリケーションでは、ここでデータストアを更新します。
        真のリターン

    リターンなし

async def main
    #TCPサーバの作成
    server = get_server
        RequestHandler
        data_store=data_store
        request_handler = handle_request
    )

    #サーバの起動
    await server.serve_forever

#サーバーの運用
asyncio.run main

V.実際の応用事例

5.1産業用温度監視システム

輸入時間の延長
datetimeからインポート
JSONをインポート
from pymodbus.client import modbusTcpClient

class Temperature
    ___
        """
        temperatureシステムの初期化
        device_configs：デバイス構成リスト
        """
        self.devices = []
        for config in device_configs
            client = ModbusTcpクライアント
                host=config ['ip']
                port= config.get 'port' 502
                timeout= config.get 'timeout' 3
            )
            self.devices.append {
                'client' client、
                'name' config ['name']
                'slave_idconfig.get' slave_id1
                'registers' config ['registers']
            })

        self.data_history = []

    read_all_temperatures
        “≪すべてのデバイスのtemperatureデータの読み取り| Read the temperature data for all device | emdw ≫”
        Read ing（）

        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_registers
                        address= reg_config ['address']
                        count= reg_config.get 'count' 1
                        slave=device ['slave_id']
                    )

                    if not result.isError
                        raw_value =.登録者[0]
                        変換式を適用する（例えば、10で割って実際のtemperatureを得る）。
                        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
                })

            exException as e
                print f"デバイス{device['name']}の読み取りエラー {e}"
            Finally
                client.close

        #履歴を保存する
        self.data_history.extend readings

        最新の1，000件のレコードのみを保存
        if self.data_history 1000
            self.data_history = self.data_history[-1000]

        リターン·リーディング

    check_alarms
        “≪アラーム条件の確認| Check Alerm Condition | emdw ≫”
        アラムズ= []

        for device_reading in readings
            for or_name or_data in_reading [readings '].items
                値= sensor_data ['actual']

                シンプルなアラームロジック
                if value 80 # 高温アラーム
                    alarms.append {
                        'device' device_reading ['device']
                        'sensor' sensor_name、
                        'value' value
                        'type''high_temperature'
                        'message' f"temperatureが高すぎる{value}°C"
                        'timestamp' device_reading ['timestamp']
                    })
                elif value10 # 低温アラーム
                    alarms.append {
                        'device' device_reading ['device']
                        'sensor' sensor_name、
                        'value' value
                        'type''low_temperature'
                        'message' f"temperatureが低すぎる{value}°C"
                        'timestamp' device_reading ['timestamp']
                    })

        リターン·アラムズ

    def export_data (self, filename ='temperature_data.json'):
        “JSONファイルへのデータのエクスポート”
        with open filename 'w' encoding ='utf-8' as f
            json.dump {
                'history' self.data_history
                '_time' datetime.now.isoformat
            } f ensure_ascii= indent=2

        print f"データがエクスポートされました{filename}"

#例を使う
if __name__== "___main__"
    #装備の配置
    device_configs = [
        {
            'name''ボイラ1'
            'ip''192.16 8.1.101'
            'registers'{
                'temperature'{'address' 0 'scale' 10.0 'unit''° C'}
                'pressure {''1' scale '100.0 'unit''kPa'}
            }
        },
        {
            'name''反応釜2'
            'ip''192.16 8.1.102'
            'registers'{
                'temperature'{'address' 0 'scale' 10.0 'unit''° C'}
                'ph_value'{'' 1 'scale' 100.0 'unit' pH '}
            }
        }
    ]

    モニターの作成
    monitor = Temperature_s

    #サイクルのモニタリング
    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 data in reading ['readings'].items
                    print f" {or}{data [actual ']} {data [']}"

            #アラームのチェック
            alarms = monitor.check_alarms readings
            if alarms
                print "\n️アラーム情報"
                Alarm in Alarmsの場合
                    print f" {al [']}"

            #5秒ごとに読む
            time.sleep 5

    exKeyboardIntert
        print "\nモニターStop"
        #データのエクスポート
        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: definit(self root self.root = root self.root.title“Modbusデバイス監視システム” self.root. ge "800x600"

#装備の配置
    self.devices = [
        {'name''PLC_1''ip''192.16 8.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._ui

    #監視の開始
    self.monitoring =真
    self.monitor_thread = threading. Thread =self.monitor_devices
    self. monitor_thread.daemon = True
    self. monitor_thread.start

def setup_ui (self):
    “≪ユーザー·インタフェースの設定| Setup User Interface | emdw ≫”
    #メインフレームワークの作成
    main_= ttk. 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_. grow = 0 n =0 st = 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']] =ラベル

    #データの表示エリア
    data_= ttk. Labelmain_text=“ライブデータ”、padding=“10”
    data_frame.grid row=1 column=0 sticky= tk.W tk.E tk.N tk.S pady = 0 10

    データを表示するツリービューの作成
    columns ='デバイス''レジスタ''アドレス''値''時刻'
    self.= ttk.Treeview data_ns = ns show = ings 'height=15

    columns内のcol
        self..hecol text=col
        self..n col width=100

    #スクロールバーを追加
    scrollbar = ttk.Scrollbar data_frame orient= tk.VERTICAL command=self. tree.yview
    self..ure ysccommand = scbar.set

    self.. grow = 0 n =0 st = tk.W tk.E tk.N tk.S
    scrollbar.grid row=0、column=1、sticky= tk.N、tk.S

    #コントロールボタン領域
    button_= ttk. main_
    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

read_
    “≪単一デバイスのデータの読み取り| Read Data | emdw ≫”
    client = ModbusTcpクライアント
        host= device_config ['ip']
        port= device_config ['port']
        タイムアウト=3
    )

    try
        if client.connect。
            #レジスタ0-19を読み込む
            result = client.read_registers
                address=0、
                カウント=20、
                slave= device_config ['slave_id']
            )

            結果がなければ、isError
                Return {
                    'status''オンライン'
                    'data' result.registers
                    'timestamp' datetime.now
                }
            else nParticle
                Return {
                    'status''読み取りエラー'
                    'data'
                    'timestamp' datetime.now
                }
        else nParticle
            Return {
                'status''接続が失敗しました'
                'data' None
                'timestamp' datetime.now
            }

    exException as e
        Return {
            'status' f'例外{str e}'
            'data'
            'timestamp' datetime.now
        }
    Finally
        client.close

def monitor_devices self
    “≪デバイス·スレッドの監視| Monitor Device Threads | emdw ≫”
    while self. moning
        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 [name ']

        #2秒ごとに更新
        time.sleep 2

update_ui
    “≪ユーザー·インタフェースの更新| Update User Interface | emdw ≫”
    #ステータスタグの更新
    status_text = f"{_name}{[status ']}"
    if [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[_name].config foreground ='green'
    elif 'エラー' in result['status'] or '失敗' in result['status']
        self.status_labels[_name].config foreground ='red'
    else nParticle
        self.status_labels[_name].config foreground ='orange'

    #データ表の更新
    if result ['data']
        古いデータの削除
        for in self..get_children
            if self..[val '][0] ==_name
                self..

        #新しいデータを追加
        for i value in enumerate result ['data']
            self..insert '' endval =
                device_name
                f '400 {i+1 02d}'
                私は、
                value、
                result ['timestamp'].strftime ' %H %M %S' if result ['timestamp'] else ''
            ))

def manual_refresh self
    “≪手動でデータをリフレッシュ| Refresh Data | emdw ≫”
    for device in self.devices
        result = self.read_device_data (device)
        self.update_ui [name ']

def export_data (self):
    “≪ファイルへのデータのエクスポート| Export Data to File | emdw ≫”
    JSONをインポート
    datetimeからインポート

    export_data = {
        '_time' datetime.now.isoformat
        'devices {}
    }

    for device_name device_data in self.data.items
        _data [
            'status' device_data ['status']
            'last_update' device_data ['last_update'].isoformat if device_data ['last_update'] else None
            'registers'_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

    #エクスポート成功のメッセージを表示
    Tキンターからのインポートメッセージボックス
    messagebox.showinfo "エクスポート成功" f"データはエクスポート先{filename}"

def stop_monitoring (self):
    “≪監視停止| Stop Monitoring | emdw ≫”
    self.monitoring = False
    self.root.quit

GUIアプリケーションの実行

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