pymodbusSimulator（シミュレータ）搭建完全チュートリアル：零硬件测试Modbus通讯 TCP/RTUサーバー側。与Web管理界面実践

pymodbus是Python生态中最流行的开源Modbusプロトコル栈，不仅提供功能完善的クライアント側は，还内置了强大的Simulator（シミュレータ）（Simulator）Function，可以在没実際の硬件的情况下完全模拟Modbus TCP/RTUdevice，サポートWeb管理界面和自定义データ模型。本記事详细介绍pymodbus 3.x版本的インストール設定、TCP/RTUSimulator（シミュレータ）搭建、自定义データストア、クライアント側は全機能コード使用、异步プログラミング以及完全的测试実践，帮助開発者たちは在零硬件条件下完成Modbus通讯开发与测试。

一、pymodbus概述

1.1 什么是pymodbus

pymodbus是一个功能完全的Python Modbusプロトコル実装，サポートModbus TCP、Modbus RTU、Modbus ASCII和Modbus TLS四种协议，同时提供クライアント側は（master）和サーバー側。（slave/Simulator（シミュレータ））実装。它是工业自動化领域最流行的Python Modbus库，被广泛用于デバイスデバッグ、协议测试、データの収集和ゲートウェイ开发。

1.2 核心特性

• 全协议サポート：TCP、RTU、ASCII、TLS四种传输协议全カバー
• クライアント側は+サーバー側。：既可以做マスター読み書きデバイス，也可以做ステーションからのシミュレーションデバイス
• 内置Simulator（シミュレータ）：ModbusSimulatorServer带Web管理界面，可経由HTTP API修改レジスタの値
• 同步+asynchronous：サポート同步阻塞API和asyncioasynchronousAPI
• REPL交互：内置Interactive（インタラクティブ）命令行，可直接入力Modbus命令测试
• 纯Python実装：无需编译，跨プラットフォーム，pip一键インストール
• 活跃维护：2026年最新版本3.15.x，継続的な更新
• 完全無料：BSD开源协议，商业使用无限制

1.3 版本説明

pymodbus 3.x是当前主流版本，与2.x相比API有较大变化。本記事すべての代码基于pymodbus 3.10+版本。もし你还在使用2.x版本，建议アップグレード，因为3.x修复了大量bug并提供了更好的异步サポート。

二、インストール与环境准备

# インストールpymodbus（推奨される使用虚拟环境）
python -m venv venv
source venv/bin/activate  # Linux/macOS
# venv\Scripts\activate   # Windows

# インストール最新版本
pip install pymodbus

# インストール带シリアルポートサポート（RTU需要pyserial）
pip install pymodbus[serial]

# インストール带Simulator（シミュレータ）Web界面サポート
pip install pymodbus[simulator]

# 完全インストール（すべての可选依赖）
pip install pymodbus[all]

# 验证インストール
python -c "import pymodbus; print(pymodbus.__version__)"
# 出力：3.10.0 或更高版本

# 表示命令行工具
pymodbus --help
pymodbus.simulator --help  # Simulator（シミュレータ）命令行

三、Quick Start：5分搭建Modbus TCPSimulator（シミュレータ）

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

# simple_server.py - 最简Modbus TCPSimulator（シミュレータ）
from pymodbus.server import StartTcpServer
from pymodbus.datastore import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequentialDataBlock,
)

# 作成データ块：address0-99初期値は为0
# ModbusSequentialDataBlock(starting address, [初期値はリスト])
coils = ModbusSequentialDataBlock(0, [False] * 100)       # コイル（function code01/05/15）
discrete_inputs = ModbusSequentialDataBlock(0, [False] * 100)  # ショップ型入力（02）
holding_registers = ModbusSequentialDataBlock(0, [0] * 100)    # レジスタを保持する。（03/06/16）
input_registers = ModbusSequentialDataBlock(0, [0] * 100)      # 入力レジスタ。（04）

# 作成スレーブContextより，将四种データ块绑定駅から。ID=1
slave_context = ModbusSlaveContext(
    di=discrete_inputs,    # discrete inputs
    co=coils,              # coils
    hr=holding_registers,  # holding registers
    ir=input_registers,    # input registers
    unit_id=1,             # slave address
)

# 作成サーバー側。Contextより（单駅からのモデル）
server_context = ModbusServerContext(slaves=slave_context, single=True)

# 预设一些初期値は
holding_registers.setValues(0, [100, 200, 300, 400, 500])  # register0-4
coils.setValues(0, [True, False, True, False])              # コイル0-3

print("Modbus TCPSimulator（シミュレータ）启动，监听 0.0.0.0:502")
print("slaveID=1，レジスタを保持する。0-4初期値は：100,200,300,400,500")
print("按 Ctrl+C Stop")

# 启动サーバー側。（阻塞运行）
StartTcpServer(
    context=server_context,
    address=("0.0.0.0", 502),
)

# 运行Simulator（シミュレータ）
python simple_server.py

# 在另一个终端用QModMaster或pymodbusクライアント側は接続テスト
# 连接アドレス：127.0.0.1:502，slaveID：1

四、内置Simulator（シミュレータ）ModbusSimulatorServer（带Web界面）

pymodbus 3.x内置了功能强大的ModbusSimulatorServer，它不仅模拟Modbusdevice，还提供Web管理界面和REST API，可以在浏览器中实时表示和修改レジスタの値，模拟デバイス行为。

4.1 命令行启动Simulator（シミュレータ）

# 最简单的启动方式（使用デフォルト設定）
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 Simulator（シミュレータ）設定文件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（シミュレータ）

# 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)  # run1小时

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

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

五、Modbus RTUSimulator（シミュレータ）搭建

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

# rtu_server.py - Modbus RTUSimulator（シミュレータ）
from pymodbus.server import StartSerialServer
from pymodbus.datastore import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequentialDataBlock,
)

# 作成データストア
holding_registers = ModbusSequentialDataBlock(0, [0] * 100)
coils = ModbusSequentialDataBlock(0, [False] * 100)
discrete_inputs = ModbusSequentialDataBlock(0, [False] * 100)
input_registers = ModbusSequentialDataBlock(0, [0] * 100)

# 预设模拟データ（模拟温湿度センサー）
holding_registers.setValues(0, [250])   # temperature 25.0°C（分辨率0.1）
holding_registers.setValues(1, [652])   # 湿度 65.2%RH
input_registers.setValues(0, [2200])    # 电压 220.0V
input_registers.setValues(1, [135])     # 电流 13.5A

slave_context = ModbusSlaveContext(
    di=discrete_inputs,
    co=coils,
    hr=holding_registers,
    ir=input_registers,
    unit_id=1,
)

server_context = ModbusServerContext(slaves=slave_context, single=True)

print("Modbus RTUSimulator（シミュレータ）启动")
print("serial port: /tmp/pts0, Baud rate: 9600, 8N1, slaveID: 1")

StartSerialServer(
    context=server_context,
    port="/tmp/pts0",       # Serial Port Device
    baudrate=9600,          # Baud rate
    bytesize=8,             # data bit
    parity="N",             # check digit N/E/O
    stopbits=1,             # stop bit
    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=/tmp/pts0 pty,raw,echo=0,link=/tmp/pts1

# Simulator（シミュレータ）使用 /tmp/pts0
# クライアント側は使用 /tmp/pts1
# 2つバーチャルシリアルポート接続された，シミュレート可能真实RS485通讯

# Windows用户使用com0com作成バーチャルシリアルポート对
# 下载アドレス：https://com0com.sourceforge.net/
# 作成COM3<->COM4对，Simulator（シミュレータ）用COM3，クライアント側は用COM4

六、多ステーションからのシミュレーション器

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

# multi_slave_server.py - 多ステーションからのシミュレーション器
from pymodbus.server import StartTcpServer
from pymodbus.datastore import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequentialDataBlock,
)

def create_slave(unit_id, temp, hum):
    """作成一个スレーブ，模拟温湿度センサー"""
    hr = ModbusSequentialDataBlock(0, [0] * 100)
    co = ModbusSequentialDataBlock(0, [False] * 100)
    di = ModbusSequentialDataBlock(0, [False] * 100)
    ir = ModbusSequentialDataBlock(0, [0] * 100)

    hr.setValues(0, [temp])   # temperature
    hr.setValues(1, [hum])    # 湿度

    return ModbusSlaveContext(di=di, co=co, hr=hr, ir=ir, unit_id=unit_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("多スレーブModbus TCPSimulator（シミュレータ）启动")
print("slave1（车间A）：temperature25.0°C, 湿度65.2%")
print("slave2（车间B）：temperature28.3°C, 湿度72.1%")
print("slave3（车间C）：temperature22.1°C, 湿度58.3%")
print("监听 0.0.0.0:502")

StartTcpServer(context=server_context, address=("0.0.0.0", 502))

七、pymodbusクライアント側は完全ガイド

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

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

# 作成クライアント側は实例
client = ModbusTcpClient(
    host="127.0.0.1",    # ServerIP
    port=502,             # ポート
    timeout=3,            # タイムアウト時間（秒）
)

# connect
connection = client.connect()
if not connection:
    print("接続の失敗！")
    exit(1)
print("接続しましたModbus TCPServer")

# ====== function code03：read holding registers ======
# parameter：starting address, Quantity, device_id=slaveID
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"  register{i}：{val}")
else:
    print("Read failed：", result)

# ====== function code01：リードコイルを読む ======
result = client.read_coils(address=0, count=8, device_id=1)
if not result.isError():
    print("Coil status：", result.bits[:8])

# ====== function code04：read input registers ======
result = client.read_input_registers(address=0, count=4, device_id=1)
if not result.isError():
    print("入力レジスタ。：", result.registers)

# ====== function code02：read discrete inputs ======
result = client.read_discrete_inputs(address=0, count=8, device_id=1)
if not result.isError():
    print("ショップ型入力：", result.bits[:8])

# ====== function code06：write single register ======
client.write_register(address=100, value=1234, device_id=1)
print("書き込み済み。レジスタ100=1234")

# ====== function code05：write single coil ======
client.write_coil(address=0, value=True, device_id=1)  # True=ON, False=OFF
print("書き込み済み。コイル0=ON")

# ====== function code16：write multiple registers ======
client.write_registers(address=200, values=[10, 20, 30, 40, 50], device_id=1)
print("已一括書き込みき込み入レジスタ200-204")

# ====== function code15：write multiple coils ======
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クライアント側は
from pymodbus.client import ModbusSerialClient

# 作成RTUクライアント側は
client = ModbusSerialClient(
    port="/tmp/pts1",     # Serial Port Device
    baudrate=9600,        # Baud rate
    bytesize=8,           # data bit
    parity="N",           # check digit
    stopbits=1,           # stop bit
    timeout=3,            # タイムアウト
)

# connect
if not client.connect():
    print("シリアル接続に失敗しました")
    exit(1)

# Read and hold register
result = client.read_holding_registers(address=0, count=5, device_id=1)
if not result.isError():
    print("register value：", result.registers)
else:
    print("Error：", result)

# 写入
client.write_register(address=10, value=999, device_id=1)

client.close()

7.3 异步クライアント側は

# async_client.py - asynchronousModbusクライアント側は
import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def read_device(host, device_id, name):
    """読み取り单个設備のデータ"""
    client = AsyncModbusTcpClient(host=host, port=502, timeout=3)
    await client.connect()

    result = await client.read_holding_registers(0, 10, device_id=device_id)
    if not result.isError():
        print(f"[{name}] register：{result.registers}")
    else:
        print(f"[{name}] Read failed")

    client.close()

async def main():
    # 并发読み取り多个デバイス
    tasks = [
        read_device("192.168.1.100", 1, "deviceA"),
        read_device("192.168.1.101", 1, "deviceB"),
        read_device("192.168.1.102", 2, "deviceC"),
    ]
    await asyncio.gather(*tasks)

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

八、データ型処理工具

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

# modbus_utils.py - Modbusデータ型変換ツール
import struct

def regs_to_float32(regs, byte_order="ABCD"):
    """
    将2个16位レジスタ转换为32ビット浮動小数点数
    byte_order: ABCD(ビッグエンディアン), CDAB(Word exchange), DCBA(リトルエンディアン), BADC(バイトスワップ)
    """
    if byte_order == "ABCD":
        raw = (regs[0] << 16) | regs[1]
    elif byte_order == "CDAB":
        raw = (regs[1] << 16) | regs[0]
    elif byte_order == "DCBA":
        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:
        raise ValueError(f"サポートなし。的バイトオーダー：{byte_order}")

    return struct.unpack(">f", struct.pack(">I", raw))[0]

def float32_to_regs(value, byte_order="ABCD"):
    """将32ビット浮動小数点数转换为2个16位レジスタ"""
    raw = struct.unpack(">I", struct.pack(">f", value))[0]
    regs = [(raw >> 16) & 0xFFFF, raw & 0xFFFF]
    if byte_order == "CDAB":
        regs = [regs[1], regs[0]]
    return regs

def regs_to_int32(regs, signed=True, byte_order="ABCD"):
    """将2个16位レジスタ转换为32ビット整数"""
    if byte_order == "ABCD":
        raw = (regs[0] << 16) | regs[1]
    else:
        raw = (regs[1] << 16) | regs[0]
    if signed and raw >= 0x80000000:
        raw -= 0x100000000
    return raw

def regs_to_int64(regs):
    """将4个16位レジスタ转换为64ビット整数"""
    raw = 0
    for r in regs[:4]:
        raw = (raw << 16) | r
    return raw

def int16_to_signed(value):
    """将16ビット符号なし整数変換为記号付き。整数"""
    if value >= 0x8000:
        return value - 0x10000
    return value

# ====== 使用例の例 ======
if __name__ == "__main__":
    # 模拟从デバイス読み取り到的レジスタの値
    regs = [0x4148, 0x0000]  # 浮動小数点数12.5的ABCD表示

    temp = regs_to_float32(regs, "ABCD")
    print(f"temperature：{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_signed(0xFFF6)}")  # 出力：-10

九、完全実践：用Simulator（シミュレータ）测试Modbus采集程序

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

9.1 最初のステップ：启动Simulator（シミュレータ）

# 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])  # temperature（分辨率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("device1（车间A）: 127.0.0.1:502, slaveID=1")
print("device2（车间B）: 127.0.0.1:502, slaveID=2")
print("device3（车间C）: 127.0.0.1:502, slaveID=3")
print("データ每2秒自动变化")
StartTcpServer(context=context, address=("0.0.0.0", 502))

9.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+CStop）")
    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}] temperature：{data['temperature']:.1f}°C, "
                          f"湿度：{data['humidity']:.1f}%, status：{data['status']}")
                else:
                    print(f"[{name}] Read failed")
            time.sleep(5)
    except KeyboardInterrupt:
        print("\n采集停止")
    finally:
        collector.close()

if __name__ == "__main__":
    main()

9.3 第三のステップ：运行测试

# 终端1：启动Simulator（シミュレータ）
python test_simulator.py

# 终端2：运行采集程序
python data_collector.py

# 预期出力：
# 开始采集データ（每5秒一次，按Ctrl+CStop）
# === 2026-09-17 12:00:00 ===
# [车间A] temperature：25.0°C, 湿度：65.2%, status：0
# [车间B] temperature：28.3°C, 湿度：72.1%, status：0
# [车间C] temperature：22.1°C, 湿度：58.3%, status：0
# === 2026-09-17 12:00:05 ===
# [车间A] temperature：24.8°C, 湿度：65.5%, status：0
# ...（データ会随Simulator（シミュレータ）自动变化）

十、pymodbus REPLInteractive（インタラクティブ）测试

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   # connectTCPdevice
# client.read_holding_registers 0 10  # read holding registers
# client.read_coils 0 8               # リードコイルを読む
# client.write_register 100 1234      # 写レジスタ
# client.write_coil 0 true            # 写コイル
# client.disconnect                   # 断开
# help                                # 表示すべての命令
# exit                                # 退出

# 指定スレーブID：
# client.read_holding_registers 0 10 --unit 2

十一、エキスパート技巧

11.1 自定义データ块（模拟デバイス行为）

# custom_datablock.py - 自定义データ块実装デバイス逻辑
from pymodbus.datastore import ModbusSequentialDataBlock

class TemperatureDataBlock(ModbusSequentialDataBlock):
    """自定义温度データ块，読み取り时自动計算当前温度"""

    def __init__(self):
        super().__init__(0, [0] * 100)
        self.base_temp = 250  # 基础温度25.0°C

    def getValues(self, address, count=1):
        """重写getValues，読み取り时动态生成データ"""
        import time
        # 模拟温度随时间正弦波动
        import math
        fluctuation = int(50 * math.sin(time.time() / 60))  # ±5°C波动
        current_temp = self.base_temp + fluctuation

        # 更新内部存储
        self.setValues(0, [current_temp])

        # 调用父类方法戻る
        return super().getValues(address, count)

    def setValues(self, address, values):
        """重写setValues，写入时验证範囲"""
        if address == 0:  # 温度の設定值
            temp = values[0]
            if not (100 <= temp <= 400):  # 10.0°C - 40.0°C
                raise ValueError("温度超出範囲")
        super().setValues(address, values)

11.2 異常な応答模拟

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

# 在自定义データ块中抛出異常
class FaultyDataBlock(ModbusSequentialDataBlock):
    def getValues(self, address, count=1):
        # address50-59模拟違法な住所。
        if 50 <= address < 60:
            raise IllegalAddressError(f"address{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 表示通讯日志

十二、よくある質問与トラブルシューティング

問題	原因	解決方法
Connection refused	Simulator（シミュレータ）未启动/ポートエラー	確認Simulator（シミュレータ）运行，チェックポート502是否被占用
Insufficient permissions（Linux）	1024以下ポート需要root	用sudorun，或改用5020等高ポート
例外コードを返す02	アドレス超出範囲	確認Simulator（シミュレータ）データ块大小，addressStart from 0
例外コードを返す01	機能コードサポートなし。	確認データ块タイプ与機能コードマッチング
RTU応答なし。	シリアルパラメータの不一致	核对ボーレート、data bit、check digit、stop bit
浮動小数点数表示異常	バイトオーダー不マッチング	尝试ABCD/CDAB/DCBA三种バイトオーダー
pymodbus 2.x代码报错	API版本不兼容	アップグレード到3.x，修改import路径和API调用
多スレーブ読み取り失敗	single参数エラー	多スレーブ时ModbusServerContext(single=False)
异步クライアント側は卡住	未正确await	すべてのクライアント側は方法都需要await
シリアルポート打不开	被其他程序占用	关闭其他シリアルポート软件，チェックデバイス名

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

Function	2.x写法	3.x写法
TCPクライアント側は	from pymodbus.client.sync import ModbusTcpClient	from pymodbus.client import ModbusTcpClient
读レジスタ	client.read_holding_registers(0, 10, unit=1)	client.read_holding_registers(0, 10, device_id=1)
写コイル	client.write_coil(0, 0xFF00, unit=1)	client.write_coil(0, True, device_id=1)
启动サーバー側。	StartTcpServer(context, address=("0.0.0.0", 502))	相同（参数名略有变化）
异步クライアント側は	from pymodbus.client.asynchronous.tcp import AsyncModbusTCPClient	from pymodbus.client import AsyncModbusTcpClient
unitparameter	unit=1	device_id=1
コイル値	0xFF00 / 0x0000	True / False

十四、学习リソース

• 官方ドキュメント：https://pymodbus.readthedocs.io/
• GitHubsource code：https://github.com/pymodbus-dev/pymodbus
• PyPIPages：https://pypi.org/project/pymodbus/
• サンプルコード：https://github.com/pymodbus-dev/pymodbus/tree/dev/examples
• Modbusプロトコル规范：https://modbus.org/specs.php
• pymodbus REPLDocument：https://pymodbus.readthedocs.io/en/stable/source/repl.html

pymodbus作为Python生态最成熟的Modbusプロトコル栈，其内置的Simulator（シミュレータ）功能让開発者たちは在没実際の硬件的情况下也能完成完全的Modbus开发与测试。経由ModbusSimulatorServer的Web管理界面，可以实时表示和修改レジスタの値，模拟デバイス行为变化；経由自定义データ块，可以実装复杂的デバイス逻辑模拟。本記事提供的Simulator（シミュレータ）搭建、クライアント側は开发、データ変換和完全実践案例可直接使用する。于实际项目开发，建议先用Simulator（シミュレータ）验证程序逻辑，再接入真实デバイス。