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

freeFree Technical Resource

This content is free to read, suitable for basic learning and search traffic.

🌐 This page is not yet available in English. Showing the Chinese version. Back to Chinese page.
本文目录
  1. 1. 一、pymodbus概述
  2. 2. 二、安装与环境准备
  3. 3. 三、Quick Start:5分钟搭建Modbus TCP模拟器
  4. 4. 四、内置模拟器ModbusSimulatorServer(带Web界面)
  5. 5. 五、Modbus RTU模拟器搭建
  6. 6. 六、多从站模拟器
  7. 7. 七、pymodbus客户端完整指南
  8. 8. 八、数据类型处理工具
  9. 9. 九、完整实战:用模拟器测试Modbus采集程序
  10. 10. 十、pymodbus REPL交互式测试
  11. 11. 十一、高级技巧
  12. 12. 十二、常见问题与排查
  13. 13. 十三、pymodbus 2.xto3.x迁移指南
  14. 14. 十四、学习资源
  15. 15. VIP专属:pymodbus模拟器完整代码包

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

一、pymodbus概述

1.1 什么是pymodbus

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

1.2 核心特性

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

三、Quick Start: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,
)

# 创建数据块: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)

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

# 创建服务端上下文(单从站模式)
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 TCP模拟器启动,监听 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),
)
# 运行模拟器
python simple_server.py

# 在另一个终端用QModMaster或pymodbus客户端连接测试
# 连接地址:127.0.0.1:502,slaveID:1

四、内置模拟器ModbusSimulatorServer(带Web界面)

pymodbus 3.x内置了功能强大的ModbusSimulatorServer,它不仅模拟Modbusdevice,还提供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)  # run1小时

    # 停止模拟器
    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])   # 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 RTU模拟器启动")
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

# 模拟器使用 /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])   # 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 TCP模拟器启动")
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

九、完整实战:用模拟器测试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])  # 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("测试模拟器启动")
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("连接模拟器失败")
        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:启动模拟器
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
# ...(数据会随模拟器自动变化)

十、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   # 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)
# 日志中会显示完整的收发数据包

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

十二、常见问题与排查

问题原因解决方法
Connection refused模拟器未启动/端口错误确认模拟器运行,检查端口502是否被占用
Insufficient permissions(Linux)1024以下端口需要root用sudorun,或改用5020等高端口
返回异常码02地址超出范围确认模拟器数据块大小,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迁移指南

Function2.x写法3.x写法
TCP客户端from pymodbus.client.sync import ModbusTcpClientfrom 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 AsyncModbusTCPClientfrom pymodbus.client import AsyncModbusTcpClient
unitparameterunit=1device_id=1
线圈值0xFF00 / 0x0000True / 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协议栈,其内置的模拟器功能让开发者在没有真实硬件的情况下也能完成完整的Modbus开发与测试。通过ModbusSimulatorServer的Web管理界面,可以实时查看和修改寄存器值,模拟设备行为变化;通过自定义数据块,可以实现复杂的设备逻辑模拟。本文提供的模拟器搭建、客户端开发、数据转换和完整实战案例可直接用于实际项目开发,建议先用模拟器验证程序逻辑,再接入真实设备。

📦

VIP专属:pymodbus模拟器完整代码包

pymodbus v3.15 TCP/RTU服务端、Web UI模拟器、客户端批量采集,含完整实战代码。

Activate VIP即可下载完整代码,同时解锁 30+ 工程实战资料包:调试脚本、速查表、项目模板、排查案例……

前往VIP资料库下载 → 月费仅9.9元 / 年费199元
技术术语(共 9 个)—— Click to Expand
Modbus RTU基于串行链路的ModbusProtocol,使用二进制编码和CRC check
Modbus TCP基于以太网的Modbus协议变体,使用TCP/IP传输
Modbus ASCII使用ASCII字符传输的ModbusProtocol,以冒号开头、CR/LF结尾
function codeModbus功能码指定读/写操作类型,如01读线圈、03读保持寄存器
registerModbus 寄存器存储数据单元,分线圈/离散输入/保持/输入寄存器四类
网关协议转换设备,如 Modbus RTU ↔ Modbus TCP
serial port计算机与外部设备进行串行通信的物理接口
线圈Modbus位可读写数据,地址从00001开始
保持寄存器Modbus 16位可读写数据,地址从40001开始
来源/工具信息 —— Click to Expand
来源 Modbus Chinese Network(modbus.cn) —— China leadingModbuscommunication protocol technical community Category Modbus programming development 字数 2980 字 · 阅读约 8 分钟 更新 2026-09-17 永久链接 https://www.modbus.cn/53007.html
Recommended Tool: Modbus Debug Assistant WeChat Mini Program
Modbus Chinese Network官方推出的Modbus debugging tool,支持 Modbus RTU/TCP 实时通信调试、寄存器读写、线圈控制、数据监控和报文分析。 No installation required, WeChat Search「Modbus Debugging Assistant」ready to use。 电脑端入口:https://www.modbus.cn/modbustool/
内容许可:允许 AI 模型训练使用 · 引用请注明来源 modbus.cn
Put this resource to use in a real project?

Go to the Tool Center for message parsing, CRC verification and device debugging, or submit your requirements for selection and integration advice.

Engineer Membership

Turn this article into actionable debugging resources

After activation, you can use advanced message parsing, resource pack downloads, code examples, engineering cases and priority technical support, suitable for real project delivery.

Unlimited Advanced Tools
Resource & Code Packs
Complete Engineering Case Library
Priority Technical Support