pymodbusシミュレータ構築完全チュートリアル：Modbus通信TCP/RTUサーバとWeb管理インターフェイスのゼロハードウェアテスト

pymodbusはPythonエコシステムで最も人気のあるオープンソースのModbusプロトコルスタックであり、完全な機能を備えたクライアントを提供するだけでなく、強力なシミュレータ（Simulator）機能を内蔵しており、実際のハードウェアなしでModbus TCP/RTUデバイスを完全にエミュレートでき、Web管理インターフェイスとカスタムデータモデルをサポートしている。この記事では、pymodbus 3.xバージョンのインストールと構成、TCP/RTUシミュレータの構築、カスタムデータストレージ、クライアントのフルファンクションコードの使用、非同期プログラミング、完全なテスト実戦を詳細に説明し、開発者がハードウェアゼロでModbus通信の開発とテストを完了するのに役立ちます。

1. pymodbusの概要

1.1 什么是pymodbus

pymodbusはPython Modbusプロトコルの完全な実装であり、Modbus TCP、Modbus RTU、Modbus ASCII、Modbus TLSの4つのプロトコルをサポートし、クライアント（マスター）とサーバ（スレーブ/エミュレータ）の両方の実装を提供します。産業オートメーションで最も人気のあるPython Modbusライブラリであり、デバイスデバッグ、プロトコルテスト、データ収集、ゲートウェイ開発に広く使用されています。

1.2主な特徴

• 全协议サポート：TCP、RTU、ASCII、TLS四种传输协议全カバー
• クライアント+サーバーメインステーションの読み書き装置、スレーブステーションのアナログ装置の両方が可能です。
• 内蔵シミュレータ。ModbusSimulatorServerは、HTTP API経由でレジスタ値を変更できるWeb管理インターフェイスを備えています。
• 同期+非同期同期ブロッキングAPIとasyncio非同期APIのサポート
• REPL相互作用：Modbusコマンドテストに直接入力できるインタラクティブなコマンドラインを内蔵
• 純粋なPython実装：コンパイル不要、クロスプラットフォーム、pipワンクリックインストール
• アクティブ·メンテナンス：2026年最新版本3.15.x，継続的な更新
• 完全に無料。BSDオープンソース契約、商用利用制限なし

1.3バージョンの説明

pymodbus 3.xは現在の主流バージョンで、2.xからAPIが大幅に変更されています。この記事のコードはpymodbusバージョン3.10以降に基づいています。まだバージョン2.xを使用している場合は、3.xは多くのバグを修正し、非同期サポートを改善するため、アップグレードをお勧めします。

2.設置と環境準備

pymodbusのインストール（仮想環境推奨）
python-m venv venv
source venv/bin/activate # Linux/macOS
#vvScriptsactivate   # Windowsと

最新バージョンのインストール
pip install pymodbus

シリアルサポート付きのインストール（RTUにはpyserialが必要）
pip install pymodbus[シリアル]

#エミュレータWebインターフェイスのサポート付きインストール
pip install pymodbus[シミュレータ]

完全インストール（すべての依存関係）
pip install pymodbus[すべて]

#インストールの検証
python-c "import pymodbus; print pymodbus.__ version___"
#出力バージョン3.10.0以降

#コマンドラインツールを参照
Pymodbus -ヘルプ
pymodbus.simulator--help #シミュレータコマンド

クイックスタート：Modbus TCPシミュレータを5分でセットアップ

pymodbus最简单的Simulator（シミュレータ）只需几行代码，启动后即可用任何Modbusクライアント側は（如QModMaster、Modbus Poll）接続テスト。

#simple_server.py-最小限のModbus TCPエミュレータ
from pymodbus.server import StartTcpServer
from pymodbus.datastore import
    ModbusServerContext
    ModbusSlaveContext、
    ModbusSequentialDataBlock
)

#データブロックの作成：アドレス0-99の初期値は0です。
# ModbusSequentialDataBlock開始アドレス、[初期値リスト]
coils = ModbusalDataBlock 0 [] * 100       #コイルコード01/05/15
discrete_inputs = ModbusalDataBlock 0 [] * 100 #入力02
hing_registers = ModbusalDataBlock 0 [0] * 100    #レジスタを保持（03/06/16）
input_registers = ModbusalDataBlock 0 [0] * 100      #登録簿（0 4）

#4つのデータブロックをスレーブID=1にバインドするスレーブコンテキストを作成する
slave_context = ModbusSl
    di=discrete_inputs    #Imputsをダウンロード
    co=coils              # Coilsの
    = h
    ir= input_registers    #入力レジスタ
    unit_id =1             #駅の住所
)

#サーバコンテキストの作成（シングルスレーブモード）
server_context = ModbusServerContext slaves= slave_context single=True

初期値の設定
hing_registers.setVal0 [100 200 300 400 500]#レジスタ0-4
coils.setValues 0 [True False True False]              #コイル0-3

print "Modbus TCPシミュレータが起動し、0.0.0.0 502"をリッスンします。
print "スレーブID=1、レジスタ0-4の初期値を保持100、200、300、400、500"
print "Ctrl+Cで停止"

#サーバーを起動する（実行をブロックする）
StartTServer
    context=server_context
    address="0.0.0.0" 502
)

#シミュレータを実行する
Python simple_server.py

QModMasterまたはpymodbusクライアントを使用した別の端末での接続テスト
#接続アドレス：127.0.0.1 502、スレーブID：1

ModbusSimulator Server（Webインターフェイス付き）

pymodbus 3.xには強力なModbus Simulator Serverが組み込まれており、Modbusデバイスをエミュレートするだけでなく、Web管理インターフェイスとREST APIを提供し、ブラウザ内でリアルタイムでレジスタ値を表示および変更し、デバイスの動作をシミュレートします。

4.1コマンドライン起動エミュレータ{{こまんどらいんエミュレータのかいどう}}

# 最简单的启动方式（使用デフォルト設定）
pymodbus.simulator

# 指定設定文件ポートと港。
pymodbus.simulator --json_file setup.json --port 8080

# 完全参数
pymodbus.simulator 
  --modbus_server "my_server" 
  --modbus_device "my_device" 
  --http_host "0.0.0.0" 
  --http_port 8080 
  --json_file setup.json 
  --log_file server.log

# 启动后アクセス：
# Modbus TCPポート：502
# Web管理界面：http://localhost:8080
# REST API：
#   GET  /api/registers  - 表示レジスタ
#   POST /api/registers  - 修改レジスタ
#   GET  /api/log        - 表示通讯日志
#   GET  /api/calls      - 表示機能コード调用统计
#   GET  /api/server     - 表示サーバー·サーバーステータス

4.2エミュレータ設定ファイルsetup.json

{
  "server_list": {
    "my_server": {
      "comm": "tcp",
      "port": 502,
      "framer": "socket",
      "address": "0.0.0.0",
      "ignore_missing_devices": true,
      "allow_reuse_address": true,
      "reconnect_delay": 2
    }
  },
  "device_list": {
    "my_device": {
      "setup": {
        "coils": {"starting_address": 0, "size": 100},
        "discrete_inputs": {"starting_address": 0, "size": 100},
        "holding_registers": {"starting_address": 0, "size": 100},
        "input_registers": {"starting_address": 0, "size": 100}
      },
      "invalid": {
        "holding_registers": [35, 36, 37]
      },
      "device_id": 1
    }
  },
  "simulator": {
    "modbus_server": "my_server",
    "modbus_device": "my_device",
    "http_host": "0.0.0.0",
    "http_port": 8080,
    "log_file": "server.log",
    "json_file": "setup.json"
  }
}

4.3 Pythonコード起動シミュレータ

# simulator_server.py - 带Web界面的Simulator（シミュレータ）
import asyncio
from pymodbus import ModbusSimulatorServer

async def run_simulator():
    simulator = ModbusSimulatorServer(
        modbus_server="my_server",      # 对应setup.json中的server名
        modbus_device="my_device",      # 对应setup.json中的device名
        http_host="0.0.0.0",
        http_port=8080,
        json_file="setup.json",
        log_file="simulator.log",
    )

    print("Simulator（シミュレータ）启动中...")
    print("Modbus TCP: 0.0.0.0:502")
    print("Web界面:   http://localhost:8080")

    # only_start=True 表示启动后不阻塞，可继续执行其他代码
    await simulator.run_forever(only_start=True)

    # Simulator（シミュレータ）运行中...
    await asyncio.sleep(3600)  # 运行1小时

    # 停止Simulator（シミュレータ）
    await simulator.stop()
    print("Simulator（シミュレータ）停止した。")

if __name__ == "__main__":
    asyncio.run(run_simulator())

Modbus RTUシミュレータの構築

在没実際のRS485デバイス时，可以使用バーチャルシリアルポート对（如socat、com0com）配合pymodbus RTUSimulator（シミュレータ）进行测试。

#rtu_server.py-Modbus RTUシミュレータ
pymodbus.serverからStart Serial Serverをインポート
from pymodbus.datastore import
    ModbusServerContext
    ModbusSlaveContext、
    ModbusSequentialDataBlock
)

#データストアの作成
hing_registers = ModbusalDataBlock 0 [0] * 100
coils = ModbusalDataBlock 0 [] * 100
discrete_inputs = ModbusalDataBlock 0 [] * 100
input_registers = ModbusalDataBlock 0 [0] * 100

#プリセットアナログデータ（アナログ温湿度センサ）
hing_registers.setVal0 [250]   # 温度 25.0°C分解能0.1
hing_registers.setVal1 [652]   湿度65.2% RH
input_registers.setVal0 [2200]    電圧は220.0 V
input_registers.setVal1 [135]     13.5 Aの電流

slave_context = ModbusSl
    di=discrete_inputs
    co=coils
    = h
    ir= input_registers
    unit_id =1
)

server_context = ModbusServerContext slaves= slave_context single=True

print "Modbus RTUシミュレータ起動"
print "シリアルポート/tmp/pts0、ボーレート9600、8N1、スレーブID 1"

StartserialServer
    context=server_context
    port="//pts0"       #シリアルデバイス
    ボーレート= 96 00          #ボーレート
    besize =8             #データのビット
    parity="N"             #検証ビットN/E/O
    stopbits=1、             #ストップ·ポイント
    timeout=3              #タイムアウト
    framer="rtu"           # RTUモデル
)

5.1 Linux仮想シリアルポートペアの作成

#socatを使った仮想シリアルポートペアの作成
#Socatのインストール
sudo apt-get install socat # Ubuntu/Debian
brew install socat          # macOSの

#仮想シリアルポートペア/tmp/pts0と/tmp/pts1を作成する
socat -d -d pty raw echo=0 link=//pts0 pty raw echo=0 link=//pts1

#エミュレータでは/tmp/pts0を使用
#クライアントは//pts1を使用
2つの仮想シリアルポートが接続され、実際のRS 485通信をシミュレートできます。

# Windowsユーザーはcom0comを使用して仮想シリアルポートペアを作成する
#ダウンロード先https//com0com.sourceforge.net/
#COM3の作成<->COM4ペア、エミュレータ用COM3、クライアント用COM4

6.マルチステーションシミュレータ

实际项目中一条バス上往往有多个スレーブデバイス，pymodbus可以模拟多个スレーブ，每个スレーブ有独立的データストア。

#multi_slave_server.py-マルチスレーブシミュレータ
from pymodbus.server import StartTcpServer
from pymodbus.datastore import
    ModbusServerContext
    ModbusSlaveContext、
    ModbusSequentialDataBlock
)

def create_slave (unit_id, temp, hum):
    “”“温度湿度センサーをシミュレートするスレーブステーションを作成”“”
    = Modbus
    co = ModbusalDataBlock 0 [] * 100
    di = ModbusalDataBlock 0 [] * 100
    ir = ModbusSequentialDataBlock 0 [0] * 100

    . setVal   # 温度は
    . setVal    # 湿度は

    return ModbusSldi = di co=co = ir = ir_id =_id

#ステーションから3つのセンサを作成し、ワークショップA/B/Cをシミュレートします。
slaves = {
    1 create_slave 1，250，652   #ショップA 25.0°C、65.2 ℃
    2 create_slave 2，283，721   #ワークショップB 28.3°C、72.1 ℃
    3 create_slave 3 221 583   #C 22.1°C、58.3%
}

# single=Falseはマルチスレーブモードを示す
server_context = ModbusServerContext slaves= slaves single=False

print“Multi-station Modbus TCPエミュレータ起動”
print "ステーション1 A温度25.0°C、湿度65.2%"
print "スタート2（B）温度28.3°C、湿度72.1%"
print "スタート3（C）温度22.1°C、湿度58.3%"
print "リスニング0.0.0.0 502"

StartTServer context=server_context ="0.0.0.0" 502

pymodbusクライアントの完全ガイド

7.1 TCPクライアント側は基础

#tcp_client.py-Modbus TCPクライアント
from pymodbus.client import modbusTcpClient

#クライアントインスタンスの作成
client = ModbusTcpクライアント
    host="127.0.0.1"    #サーバー IP
    ポート= 50 2             #ポート
    timeout=3            #タイムアウト（秒）
)

#接続する
connection = client.connect
接続しない場合
    print "接続に失敗しました! ")
    exit 1
print "Modbus TCPサーバーに接続"

# ======関数コード03：リードホールドレジスタ======
#パラメータ開始アドレス、数、device_id =スレーブID
result = client.read_holding_registers (address=0, count=10, device_id =1)
if not result.isError
    print "レジスタ値を保持" result.registers
    for i val in enumerate result.registers
        print f"レジスタ{i}{}"
else nParticle
    print "読み取りに失敗しました" result

# ======関数コード01：リードコイル=======
result = client.read address=0、count=8、device_id =1
if not result.isError
    print "コイル".bits[8]

# ======関数コード0 4：入力レジスタの読み取り=====
result = client.read address=0、count=4、device_id =1
if not result.isError
    print "入力レジスタ" result.registers

# ======関数コード0 2：離散入力を読む=====
result = client.read address=0、count=8、device_id =1
if not result.isError
    print "離散入力" result.bits [8]

# ======関数コード0 6：単一のレジスタを書き込む=====
.
print "はレジスタ100=1234"

# =====関数コード0 5：単一コイルを書く=====
client.write_coil address=0 value=True device_id =1 # True=ON False=OFF
print "コイルに书き込まれた0=ON"

# =====関数コード16：複数のレジスタを書き込む=====
client.write_registers address=200、values=[10 20 30 40 50]、device_id =1
print "レジスタ200-204"に一括書き込みき込み済み

# ======関数コード15：複数のコイルを書く====
client.write_coils address=10、values=[True False True False]、device_id =1
print "コイル10-13"

#接続を閉じる
client.close
print“接続は闭じられました”

7.2 RTUクライアント

#rtu_client.py-Modbus RTUクライアント
pymodbus.clientからmodbus SerialClientをインポート

RTUクライアントの作成
= ModbusSerialClient
    port="/tmp/pts1"     #シリアルデバイス
    ボーレート= 96 00        #ボーレート
    besize =8           #データのビット
    parity="N"           #チェックポイント
    stopbits=1、           #ストップ·ポイント
    timeout=3            #タイムアウト
)

#接続する
if not client.connect。
    print“シリアル接続に失敗しました”
    exit 1

#リード保持レジスタ
result = client.read_holding_registers (address=0, count=5, device_id =1)
if not result.isError
    print "レジスタ値".registers
else nParticle
    print "エラー" result

#書き込み#
client.write_register address=10、value=999、device_id =1

client.close

7.3非同期クライアント

async_client.py：非同期Modbusクライアント
asyncioのインポート
from pymodbus.client import AsyncModbusTcpClient

async def read_device host device_id name
    “≪個々のデバイスデータの読み取り| Read Single Device Data | emdw ≫”
    = AsyncModbusT
    await.

    result = await client.read 0 10 device_id = device_id
    if not result.isError
        print f"[{name}]レジスタ：{.registers}"
    else nParticle
        print f"[{name}]読み取り"

    client.close

async def main
    #複数のデバイスを同時に読み込む
    TACK = [
        read_"192.168.1.100" 1 "デバイスA"
        read_"192.168.1.101" 1 "デバイスB"
        read_"192.168.1.102" 2 "デバイスC"
    ]
    await asyncio.gather *tasks

if __name__== "___main__"
    asyncio.run main

VIII.データ型処理ツール

pymodbus的read_holding_registers戻る的是整数リスト，但工业デバイス中经常使用浮動小数点数、32ビット整数等。以下提供完全的转换工具函数。

modbus_utils.py-Modbusデータ型変換ツール
構造体のインポート

regs_to_float32
    """
    2つの16ビットレジスタを32ビット浮動小数点数に変換する
    byte_order ABCDビッグエンディアン、CDABワードスワップ、DCBAリトルエンディアン、BADCバイトスワップ
    """
    if byte_order == "ABCD"
        raw = regs[0]<< 16| RGS〔1〕
    elif byte_order == "CDAB"
        raw = regs[1]<< 16|レグ〔0〕
    elif byte_order == "BA"
        raw = regs[1]& 0xFF<< 24|- regs[1]>> 8<< 16| 
              ((regs[0]& 0xFF)<< 8)| regs[0]>> 8
    elif byte_order == "BADC"
        raw = regs[0]& 0xFF<< 24|- regs[0]>> 8<< 16| 
              - regs[1]& 0xFF<< 8| regs[1]>> 8
    else nParticle
        raise ValueError f"サポートされていないエンディアン{byte_order}"

    return str.unpack "f" str.pack "I" raw [0]

float32_to_regs
    “”“32ビット浮動小数点数を2つの16ビットレジスタに変換”“”
    raw = str.unpack "I" str.pack "f" value [0]
    regs = [raw 16 0xFFFF raw 0xFFFF]
    if byte_order == "CDAB"
        regs = [regs[1] regs[0]]
    Regsへの復帰

regs_to_int32
    """2つの16ビットレジスタを32ビット整数に変换""""""
    if byte_order == "ABCD"
        raw = regs[0]<< 16| RGS〔1〕
    else nParticle
        raw = regs[1]<< 16|レグ〔0〕
    signedとraw = 0x8000000の場合
        raw = 0 x 10000000
    Rawに戻る

regs_to_int64
    “”“4つの16ビットレジスタを64ビット整数に変換”“”
    RAW = 0
    for r in regs[4]
        raw = raw<< 16| Rは
    Rawに戻る

int16_to_
    “”“16ビット符号なし整数を符号付き整数”に変换
    if value = 0x8000
        リターン値-0 x 10000
    リターン値は

# ===== 使用例=========
if __name__== "___main__"
    #デバイスから読み出したレジスタ値をシミュレートする
    regs = [0 x 414 8 0 x 0000] #浮動小数点数12.5のABCD表現

    = regs_to_float32
    print f"温度{temp. 1f}°C"#出力12.5°C

    # 32の合計
    int_regs = [0x0001 0x0000] # 65536
    print f"カウント値{regs_to_int32 int_regs}"#出力65536

    #16の符号付き
    print f"符号付き値{int16_to_ed 0xFFF6}"#出力-10

完全な戦闘：シミュレータでModbus取得手順をテストする

以下是一个完全の実践案例：先用pymodbus搭建一个模拟的多デバイス环境，然后编写采集程序読み取りデータ，最后データの検証正确性。整个过程必要なし。任何真实硬件。

9.1ステップ1：シミュレータの起動

# test_simulator.py - 测试用Simulator（シミュレータ）
import threading
import time
from pymodbus.server import StartTcpServer
from pymodbus.datastore import (
    ModbusServerContext, ModbusSlaveContext, ModbusSequentialDataBlock
)

# 全局データ块引用，用于动态更新
hr_blocks = {}

def create_device(device_id, temp_init, hum_init):
    hr = ModbusSequentialDataBlock(0, [0] * 100)
    co = ModbusSequentialDataBlock(0, [False] * 100)
    di = ModbusSequentialDataBlock(0, [False] * 100)
    ir = ModbusSequentialDataBlock(0, [0] * 100)

    hr.setValues(0, [temp_init])  # 温度（分辨率0.1）
    hr.setValues(1, [hum_init])   # 湿度（分辨率0.1）
    hr.setValues(2, [0])          # ステータス字
    hr_blocks[device_id] = hr

    return ModbusSlaveContext(di=di, co=co, hr=hr, ir=ir, unit_id=device_id)

# 作成3个模拟デバイス
slaves = {
    1: create_device(1, 250, 652),   # 车间A
    2: create_device(2, 283, 721),   # 车间B
    3: create_device(3, 221, 583),   # 车间C
}

context = ModbusServerContext(slaves=slaves, single=False)

def simulate_data_change():
    """后台线程：模拟センサーはデータ实时变化"""
    import random
    while True:
        for dev_id, hr in hr_blocks.items():
            # 温度在当前值±2範囲内波动
            current_temp = hr.getValues(0, 1)[0]
            new_temp = max(100, min(400, current_temp + random.randint(-2, 2)))
            hr.setValues(0, [new_temp])

            # 湿度波动
            current_hum = hr.getValues(1, 1)[0]
            new_hum = max(300, min(950, current_hum + random.randint(-3, 3)))
            hr.setValues(1, [new_hum])

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

# 启动データ模拟线程
threading.Thread(target=simulate_data_change, daemon=True).start()

print("测试Simulator（シミュレータ）启动")
print("デバイス1（车间A）: 127.0.0.1:502, スレーブID=1")
print("デバイス2（车间B）: 127.0.0.1:502, スレーブID=2")
print("デバイス3（车间C）: 127.0.0.1:502, スレーブID=3")
print("データ每2秒自动变化")
StartTcpServer(context=context, address=("0.0.0.0", 502))

9.2ステップ2：収集手順の作成

# data_collector.py - データの収集程序（待测试）
import time
from pymodbus.client import ModbusTcpClient

class ModbusCollector:
    def __init__(self, host, port=502):
        self.host = host
        self.port = port
        self.client = ModbusTcpClient(host=host, port=port, timeout=3)

    def connect(self):
        return self.client.connect()

    def read_sensor(self, device_id):
        """読み取り温湿度センサーデータ"""
        result = self.client.read_holding_registers(
            address=0, count=3, device_id=device_id
        )
        if result.isError():
            return None

        temp = result.registers[0] / 10.0   # 分辨率0.1°C
        hum = result.registers[1] / 10.0    # 分辨率0.1%
        status = result.registers[2]

        return {
            "device_id": device_id,
            "temperature": temp,
            "humidity": hum,
            "status": status,
        }

    def close(self):
        self.client.close()

def main():
    collector = ModbusCollector("127.0.0.1", 502)
    if not collector.connect():
        print("连接Simulator（シミュレータ）失敗")
        return

    devices = [
        (1, "车间A"),
        (2, "车间B"),
        (3, "车间C"),
    ]

    print("开始采集データ（每5秒一次，按Ctrl+C停止）")
    try:
        while True:
            print(f"n=== {time.strftime('%Y-%m-%d %H:%M:%S')} ===")
            for dev_id, name in devices:
                data = collector.read_sensor(dev_id)
                if data:
                    print(f"[{name}] 温度：{data['temperature']:.1f}°C, "
                          f"湿度：{data['humidity']:.1f}%, ステータス：{data['status']}")
                else:
                    print(f"[{name}] 読み取り失敗")
            time.sleep(5)
    except KeyboardInterrupt:
        print("n采集停止")
    finally:
        collector.close()

if __name__ == "__main__":
    main()

9.3ステップ3：テストの実行

#1：シミュレータの起動
Python test_simulator.py

#2：収集手順の実行
Python data_collector.py

#予想される輸出：
#データ収集を開始する（5秒ごとにCtrl+Cを押して停止）
# === 2026-09-17 12 00 00 ===
# [A] 温度25.0°C、湿度65.2%、0
# [B] 温度：28.3°C、湿度：72.1%、：0
# [C] 温度22.1°C、湿度58.3%、0
# === 2026-09-17 12 00 05 ====
# [A] 温度24.8°C、湿度65.5%、0
# ...（データはシミュレータにより自動的に変更されます）

pymodbus REPLインタラクティブテスト

pymodbus内置了REPL（Read-Eval-Print Loop）Interactive（インタラクティブ）命令行工具，可以直接入力Modbus命令进行快速测试，无需编写完全脚本。

#REPLの起動
pymodbus.repl

#またはPythonモジュールで起動
python-m pymodbus.repl

REPLでよく使われるコマンド：
#client.connect tcp 127.0.0.1:502   #TCPデバイスの接続
#client.read 0 10 #リードホールドレジスタ
#client.read_coils 0 8               #リード·コイル
#client.write_register 100 1234      #レジスタの書き込み
#client.write_coil 0 true            #コイルを書く
#client.disconnect                   #切断する
#ヘルプ#                                すべてのコマンドを見る
# exit                                #脱退する

#ステーションIDを指定する：
#client.read_hing_registers 0 10 ---2

11.高度なスキル

11.1カスタム·データ·ブロックデバイスの動作をシミュレート

#custom_datablock.py-カスタムデータブロック実装デバイス論理
from pymodbus.datastore import modbusSequentialDataBlock

class TemperatureDataBlock ModbusSequentialDataBlock
    “≪カスタム温度データ·ブロック| Custom Temperature Data Block | emdw ≫”、読取り時に現在の温度が自動的に計算される

    ___
        super.__ init__0 [0] * 100
        self.base_= 250 #温度25.0°C

    getVal
        “≪ getValuesを上書きし、読取り時にデータを動的に生成| Data Generation | emdw ≫”
        輸入時間の延長
        #時間の経過とともに温度変動をシミュレート
        数学のインポート。
        fluation = int 50 * math. time.time/ 60 # ±5°C変动
        _

        #内部ストレージの更新
        self. setVal0 [_]

        #親メソッドを呼び出す
        return super.getValues address count

    setVal
        “≪ setValues | emdw ≫”を上書きし、“≪書き込み時に範囲を検証| Validate Range on Write | emdw ≫”
        if address == 0 # 温度の設定値
            temp = values[0]
            if not 100 = temp = 400 # 10.0°C - 40.0°C
                raise ValueError“範囲外の温度”
        super. setValval

11.2異常応答シミュレーション

# 模拟デバイス戻る異常な応答，测试クライアント側は的エラー処理能力
from pymodbus.exceptions import IllegalAddressError, IllegalFunctionError

# 在自定义データ块中抛出異常
class FaultyDataBlock(ModbusSequentialDataBlock):
    def getValues(self, address, count=1):
        # アドレス50-59模拟違法な住所。
        if 50 <= address < 60:
            raise IllegalAddressError(f"アドレス{address}不可アクセス")
        return super().getValues(address, count)

# クライアント側は読み取り这些アドレス时会異常コードを受信02
# 用于测试クライアント側は是否正确処理異常な応答

11.3通信ログとパケット。

# 开启详细日志
import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("pymodbus")
logger.setLevel(logging.DEBUG)

# クライアント側は日志
from pymodbus.client import ModbusTcpClient
client = ModbusTcpClient("127.0.0.1", port=502)
client.connect()

# 开启通讯追踪
client.trace_packet = True  # 追踪生データ包
client.trace_pdu = True     # 追踪PDU

result = client.read_holding_registers(0, 5, device_id=1)
# 日志中会表示完全的收发データパケットは

# Simulator（シミュレータ）的Web界面也有日志功能
# アクセス http://localhost:8080/api/log 表示通讯日志

12.よくある質問とチェック

問題	理由は	解決策は
接続の取り消し	Simulator（シミュレータ）未启动/ポートエラー	確認Simulator（シミュレータ）运行，チェックポート502是否被占用
権限が不十分なLinux	1024以下のポートにrootが必要	sudoで実行するか、5020などのハイエンドポートに切り替える
例外コードを返す02	範囲外のアドレス	エミュレータのブロックサイズを確認します。アドレスは0から始まります。
背番号は0 1。	機能コード未対応	確認データ块タイプ与機能コードマッチング
RTU応答なし	シリアルパラメータの不一致	チェックボーレート、データビット、チェックビット、ストップビット
浮動小数点数表示異常	エンディアンの不一致	ABCD/CDAB/DCBAの3つのエンディアンを試す
pymodbus 2.xコードエラー	APIバージョンの互換性がない	アップグレード到3.x，修改import路径和API调用
ステーションからの読み取り失敗	singleパラメータエラー	マルチスレーブの場合のModbusServerContext single=False
异步クライアント側は卡住	正しくないawait	すべてのクライアントメソッドにawaitが必要
シリアルポートが開けない	他の手続きによる占有	他のシリアルポートソフトウェアを閉じて、デバイス名を確認します。

十三、pymodbus 2.x到3.x迁移ガイド

機能性は	2. Xの書き込み	3.x写法
TCPクライアント	from pymodbus.client.sync import ModbusTcpClient	from pymodbus.client import modbusTcpClient
读レジスタ	client.read_hing_registers 0 10 =1	client.read_hing_registers 0 10_id=1
コイルを書く	client.write_coil 0 0xFF00 unit=1	client.write_coil 0 True device_id =1
サーバーの起動	StartTServer context ="0.0.0.0" 502	同一（パラメータ名が若干変更）
非同期クライアント	from pymodbus.client.asynchronous.tcp import AsyncModbusTCPClient	from pymodbus.client import AsyncModbusTcpClient
unitパラメータ	unit=1	device_id =1
コイル値	0 x FF 00/0 x 0000	True / False

14.学習資源

• 公式ドキュメントpymodbus.readthedocs.io/
• GitHubのソースコードgithub.com/pymodbus-dev/pymodbus
• PyPIページpypi.org/project/pymodbus/
• サンプルコード。github.com/pymodbus-dev/pymodbus/tree/dev/examples
• Modbusプロトコル仕様modbus.org/specs.php
• pymodbus REPLドキュメントpymodbus.readthedocs.io/en/stable/source/repl.html

pymodbusはPythonエコシステムで最も成熟したModbusプロトコルスタックであり、内蔵のエミュレータ機能により、開発者は実際のハードウェアなしで完全なModbus開発とテストを完了できます。ModbusSimulatorServerのWeb管理インターフェイスを通して、リアルタイムにレジスタ値を見て修正し、設備挙動変化をシミュレーションすることができる;カスタムデータブロックを通して、複雑な設備論理シミュレーションを実現することができる。本稿で提供するシミュレータ構築、クライアント開発、データ変換、完全な実戦事例は、実際のプロジェクト開発に直接使用でき、シミュレータでプログラムロジックを検証し、実際の機器に接続することを推奨します。