pymodbus模拟器搭建完整教程：零硬件测试Modbus通讯 TCP/RTU服务端与Web管理界面实战

pymodbus是Python生态中最流行的开源Modbus协议栈，不仅提供功能完善的客户端，还内置了强大的模拟器（Simulator）Funktion，可以在没有真实硬件的情况下完整模拟Modbus TCP/RTUAusrüstung，支持Web管理界面和自定义数据模型。本文详细介绍pymodbus 3.x版本的安装配置、TCP/RTU模拟器搭建、自定义数据存储、客户端全功能码使用、异步编程以及完整的测试实战，帮助开发者在零硬件条件下完成Modbus通讯开发与测试。

一、pymodbus概述

1.1 什么是pymodbus

pymodbus是一个功能完整的Python Modbus协议实现，支持Modbus TCP、Modbus RTU、Modbus ASCII和Modbus TLS四种协议，同时提供客户端（Hauptstation）和服务端（Von der Station/模拟器）实现。它是工业自动化领域最流行的Python Modbus库，被广泛用于设备调试、协议测试、数据采集和网关开发。

1.2 核心特性

• 全协议支持：TCP、RTU、ASCII、TLS四种传输协议全覆盖
• 客户端+服务端：既可以做主站读写设备，也可以做从站模拟设备
• 内置模拟器：ModbusSimulatorServer带Web管理界面，可通过HTTP API修改寄存器值
• 同步+Asynchron：支持同步阻塞API和asyncioAsynchronAPI
• REPL交互：内置交互式命令行，可直接输入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]

# 安装带模拟器Web界面支持
pip install pymodbus[simulator]

# 完整安装（所有可选依赖）
pip install pymodbus[all]

# 验证安装
python -c "import pymodbus; print(pymodbus.__version__)"
# 输出：3.10.0 或更高版本

# 查看命令行工具
pymodbus --help
pymodbus.simulator --help  # 模拟器命令行

三、Schneller Eintritt：5分钟搭建Modbus TCP模拟器

pymodbus最简单的模拟器只需几行代码，启动后即可用任何Modbus客户端（如QModMaster、Modbus Poll）连接测试。

# simple_server.py - 最简Modbus TCP模拟器
from pymodbus.server import StartTcpServer
from pymodbus.datastore import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequentialDataBlock,
)

# 创建数据块：Adresse0-99初始值为0
# ModbusSequentialDataBlock(Ausgang Adresse, [初始值列表])
coils = ModbusSequentialDataBlock(0, [False] * 100)       # 线圈（Funktionscode01/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）

# 创建从站上下文，将四种数据块绑定到从站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,             # von der Station Adresse
)

# 创建服务端上下文（单从站模式）
server_context = ModbusServerContext(slaves=slave_context, single=True)

# 预设一些初始值
holding_registers.setValues(0, [100, 200, 300, 400, 500])  # Register.0-4
coils.setValues(0, [True, False, True, False])              # 线圈0-3

print("Modbus TCP模拟器启动，监听 0.0.0.0:502")
print("Von der StationID=1，保持寄存器0-4初始值：100,200,300,400,500")
print("按 Ctrl+C Stopp!")

# 启动服务端（阻塞运行）
StartTcpServer(
    context=server_context,
    address=("0.0.0.0", 502),
)

# 运行模拟器
python simple_server.py

# 在另一个终端用QModMaster或pymodbus客户端连接测试
# 连接地址：127.0.0.1:502，Von der StationID：1

四、内置模拟器ModbusSimulatorServer（带Web界面）

pymodbus 3.x内置了功能强大的ModbusSimulatorServer，它不仅模拟ModbusAusrüstung，还提供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界面的模拟器
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("模拟器启动中...")
    print("Modbus TCP: 0.0.0.0:502")
    print("Web界面:   http://localhost:8080")

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

    # 模拟器运行中...
    await asyncio.sleep(3600)  # Betrieb1小时

    # 停止模拟器
    await simulator.stop()
    print("模拟器已停止")

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

五、Modbus RTU模拟器搭建

在没有真实RS485设备时，可以使用虚拟串口对（如socat、com0com）配合pymodbus RTU模拟器进行测试。

# rtu_server.py - Modbus RTU模拟器
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])   # Temperatur 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 RTU模拟器启动")
print("Die Serie: /tmp/pts0, Porter Rate: 9600, 8N1, Von der StationID: 1")

StartSerialServer(
    context=server_context,
    port="/tmp/pts0",       # Serielle Ausrüstung
    baudrate=9600,          # Porter Rate
    bytesize=8,             # Datenplatz
    parity="N",             # Prüfstand. N/E/O
    stopbits=1,             # Stoppen Sie Platz
    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

# 模拟器使用 /tmp/pts0
# 客户端使用 /tmp/pts1
# 两个虚拟串口已连通，可模拟真实RS485通讯

# Windows用户使用com0com创建虚拟串口对
# 下载地址：https://com0com.sourceforge.net/
# 创建COM3<->COM4对，模拟器用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])   # Temperatur
    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 TCP模拟器启动")
print("Von der Station1（车间A）：Temperatur25.0°C, 湿度65.2%")
print("Von der Station2（车间B）：Temperatur28.3°C, 湿度72.1%")
print("Von der Station3（车间C）：Temperatur22.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",    # Der ServerIP
    port=502,             # 端口
    timeout=3,            # 超时时间（秒）
)

# Verbindung
connection = client.connect()
if not connection:
    print("连接失败！")
    exit(1)
print("已连接到Modbus TCPDer Server")

# ====== Funktionscode03：Lesen Sie das Register ======
# Parameter：Ausgang Adresse, Menge, device_id=Von der StationID
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("Lesen gescheitert：", result)

# ====== Funktionscode01：读线圈 ======
result = client.read_coils(address=0, count=8, device_id=1)
if not result.isError():
    print("Zustand der Coil：", result.bits[:8])

# ====== Funktionscode04：Eingangsregister lesen ======
result = client.read_input_registers(address=0, count=4, device_id=1)
if not result.isError():
    print("输入寄存器：", result.registers)

# ====== Funktionscode02：Lesen Sie getrennte Eingabe ======
result = client.read_discrete_inputs(address=0, count=8, device_id=1)
if not result.isError():
    print("离散输入：", result.bits[:8])

# ====== Funktionscode06：Schreiben eines einzelnen Registers ======
client.write_register(address=100, value=1234, device_id=1)
print("已写入寄存器100=1234")

# ====== Funktionscode05：Schreiben Sie eine einzelne Schleife. ======
client.write_coil(address=0, value=True, device_id=1)  # True=ON, False=OFF
print("已写入线圈0=ON")

# ====== Funktionscode16：Schreiben mehrerer Register ======
client.write_registers(address=200, values=[10, 20, 30, 40, 50], device_id=1)
print("已批量写入寄存器200-204")

# ====== Funktionscode15：Schreiben Sie mehrere Schleife. ======
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",     # Serielle Ausrüstung
    baudrate=9600,        # Porter Rate
    bytesize=8,           # Datenplatz
    parity="N",           # Prüfstand.
    stopbits=1,           # Stoppen Sie Platz
    timeout=3,            # 超时
)

# Verbindung
if not client.connect():
    print("串口连接失败")
    exit(1)

# Lesen und halten Register
result = client.read_holding_registers(address=0, count=5, device_id=1)
if not result.isError():
    print("Registerwert：", result.registers)
else:
    print("Fehler.：", result)

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

client.close()

7.3 异步客户端

# async_client.py - AsynchronModbus客户端
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}] Lesen gescheitert")

    client.close()

async def main():
    # 并发读取多个设备
    tasks = [
        read_device("192.168.1.100", 1, "AusrüstungA"),
        read_device("192.168.1.101", 1, "AusrüstungB"),
        read_device("192.168.1.102", 2, "AusrüstungC"),
    ]
    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(Wort ausgetauscht.), 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"Temperatur：{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

九、完整实战：用模拟器测试Modbus采集程序

以下是一个完整的实战案例：先用pymodbus搭建一个模拟的多设备环境，然后编写采集程序读取数据，最后验证数据正确性。整个过程不需要任何真实硬件。

9.1 第一步：启动模拟器

# test_simulator.py - 测试用模拟器
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])  # Temperatur（分辨率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("测试模拟器启动")
print("Ausrüstung1（车间A）: 127.0.0.1:502, Von der StationID=1")
print("Ausrüstung2（车间B）: 127.0.0.1:502, Von der StationID=2")
print("Ausrüstung3（车间C）: 127.0.0.1:502, Von der StationID=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("连接模拟器失败")
        return

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

    print("开始采集数据（每5秒一次，按Ctrl+CStopp!）")
    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}] Temperatur：{data['temperature']:.1f}°C, "
                          f"湿度：{data['humidity']:.1f}%, Status：{data['status']}")
                else:
                    print(f"[{name}] Lesen gescheitert")
            time.sleep(5)
    except KeyboardInterrupt:
        print("\n采集停止")
    finally:
        collector.close()

if __name__ == "__main__":
    main()

9.3 第三步：运行测试

# 终端1：启动模拟器
python test_simulator.py

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

# 预期输出：
# 开始采集数据（每5秒一次，按Ctrl+CStopp!）
# === 2026-09-17 12:00:00 ===
# [车间A] Temperatur：25.0°C, 湿度：65.2%, Status：0
# [车间B] Temperatur：28.3°C, 湿度：72.1%, Status：0
# [车间C] Temperatur：22.1°C, 湿度：58.3%, Status：0
# === 2026-09-17 12:00:05 ===
# [车间A] Temperatur：24.8°C, 湿度：65.5%, Status：0
# ...（数据会随模拟器自动变化）

十、pymodbus REPL交互式测试

pymodbus内置了REPL（Read-Eval-Print Loop）交互式命令行工具，可以直接输入Modbus命令进行快速测试，无需编写完整脚本。

# 启动REPL
pymodbus.repl

# 或通过Python模块启动
python -m pymodbus.repl

# REPL中常用命令：
# client.connect tcp 127.0.0.1:502   # VerbindungTCPAusrüstung
# client.read_holding_registers 0 10  # Lesen Sie das Register
# 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):
        # Adresse50-59模拟非法地址
        if 50 <= address < 60:
            raise IllegalAddressError(f"Adresse{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)
# 日志中会显示完整的收发数据包

# 模拟器的Web界面也有日志功能
# 访问 http://localhost:8080/api/log 查看通讯日志

十二、常见问题与排查

十三、pymodbus 2.xzu3.x迁移指南

十四、学习资源

• 官方文档：https://pymodbus.readthedocs.io/
• GitHubQuellencode：https://github.com/pymodbus-dev/pymodbus
• PyPISeiten：https://pypi.org/project/pymodbus/
• 示例代码：https://github.com/pymodbus-dev/pymodbus/tree/dev/examples
• Modbus协议规范：https://modbus.org/specs.php
• pymodbus REPLDokumentation：https://pymodbus.readthedocs.io/en/stable/source/repl.html

pymodbus作为Python生态最成熟的Modbus协议栈，其内置的模拟器功能让开发者在没有真实硬件的情况下也能完成完整的Modbus开发与测试。通过ModbusSimulatorServer的Web管理界面，可以实时查看和修改寄存器值，模拟设备行为变化；通过自定义数据块，可以实现复杂的设备逻辑模拟。本文提供的模拟器搭建、客户端开发、数据转换和完整实战案例可直接用于实际项目开发，建议先用模拟器验证程序逻辑，再接入真实设备。