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

免费免费技术资料

这篇内容可直接阅读,适合用于基础学习和搜索引流。

本文目录
  1. 1. 一、pymodbus概述
  2. 2. 二、安装与环境准备
  3. 3. 三、快速入门: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.x到3.x迁移指南
  14. 14. 十四、学习资源

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

一、pymodbus概述

1.1 什么是pymodbus

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

1.2 核心特性

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

三、快速入门: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,
)

# 创建数据块:地址0-99初始值为0
# ModbusSequentialDataBlock(起始地址, [初始值列表])
coils = ModbusSequentialDataBlock(0, [False] * 100)       # 线圈(功能码01/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,             # 从站地址
)

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

# 预设一些初始值
holding_registers.setValues(0, [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 停止")

# 启动服务端(阻塞运行)
StartTcpServer(
    context=server_context,
    address=("0.0.0.0", 502),
)
# 运行模拟器
python simple_server.py

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

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

pymodbus 3.x内置了功能强大的ModbusSimulatorServer,它不仅模拟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界面的模拟器
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)  # 运行1小时

    # 停止模拟器
    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])   # 温度 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("串口: /tmp/pts0, 波特率: 9600, 8N1, 从站ID: 1")

StartSerialServer(
    context=server_context,
    port="/tmp/pts0",       # 串口设备
    baudrate=9600,          # 波特率
    bytesize=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=/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])   # 温度
    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("从站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")

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",    # 服务器IP
    port=502,             # 端口
    timeout=3,            # 超时时间(秒)
)

# 连接
connection = client.connect()
if not connection:
    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}:{val}")
else:
    print("读取失败:", result)

# ====== 功能码01:读线圈 ======
result = client.read_coils(address=0, count=8, device_id=1)
if not result.isError():
    print("线圈状态:", result.bits[:8])

# ====== 功能码04:读输入寄存器 ======
result = client.read_input_registers(address=0, count=4, device_id=1)
if not result.isError():
    print("输入寄存器:", result.registers)

# ====== 功能码02:读离散输入 ======
result = client.read_discrete_inputs(address=0, count=8, device_id=1)
if not result.isError():
    print("离散输入:", result.bits[:8])

# ====== 功能码06:写单个寄存器 ======
client.write_register(address=100, value=1234, device_id=1)
print("已写入寄存器100=1234")

# ====== 功能码05:写单个线圈 ======
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客户端
from pymodbus.client import ModbusSerialClient

# 创建RTU客户端
client = ModbusSerialClient(
    port="/tmp/pts1",     # 串口设备
    baudrate=9600,        # 波特率
    bytesize=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("寄存器值:", result.registers)
else:
    print("错误:", result)

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

client.close()

7.3 异步客户端

# async_client.py - 异步Modbus客户端
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}] 寄存器:{result.registers}")
    else:
        print(f"[{name}] 读取失败")

    client.close()

async def main():
    # 并发读取多个设备
    tasks = [
        read_device("192.168.1.100", 1, "设备A"),
        read_device("192.168.1.101", 1, "设备B"),
        read_device("192.168.1.102", 2, "设备C"),
    ]
    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(字交换), 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"温度:{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])  # 温度(分辨率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("设备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 第二步:编写采集程序

# 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+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 第三步:运行测试

# 终端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)交互式命令行工具,可以直接输入Modbus命令进行快速测试,无需编写完整脚本。

# 启动REPL
pymodbus.repl

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

# REPL中常用命令:
# client.connect tcp 127.0.0.1:502   # 连接TCP设备
# client.read_holding_registers 0 10  # 读保持寄存器
# 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):
        # 地址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)
# 日志中会显示完整的收发数据包

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

十二、常见问题与排查

问题原因解决方法
Connection refused模拟器未启动/端口错误确认模拟器运行,检查端口502是否被占用
权限不足(Linux)1024以下端口需要root用sudo运行,或改用5020等高端口
返回异常码02地址超出范围确认模拟器数据块大小,地址从0开始
返回异常码01功能码不支持确认数据块类型与功能码匹配
RTU无响应串口参数不匹配核对波特率、数据位、校验位、停止位
浮点数显示异常字节序不匹配尝试ABCD/CDAB/DCBA三种字节序
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 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
unit参数unit=1device_id=1
线圈值0xFF00 / 0x0000True / False

十四、学习资源

  • 官方文档:https://pymodbus.readthedocs.io/
  • GitHub源码:https://github.com/pymodbus-dev/pymodbus
  • PyPI页面:https://pypi.org/project/pymodbus/
  • 示例代码:https://github.com/pymodbus-dev/pymodbus/tree/dev/examples
  • Modbus协议规范:https://modbus.org/specs.php
  • pymodbus REPL文档:https://pymodbus.readthedocs.io/en/stable/source/repl.html

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

技术术语(共 9 个)—— 点击展开
Modbus RTU基于串行链路的Modbus协议,使用二进制编码和CRC校验
Modbus TCP基于以太网的Modbus协议变体,使用TCP/IP传输
Modbus ASCII使用ASCII字符传输的Modbus协议,以冒号开头、CR/LF结尾
功能码Modbus功能码指定读/写操作类型,如01读线圈、03读保持寄存器
寄存器Modbus 寄存器存储数据单元,分线圈/离散输入/保持/输入寄存器四类
网关协议转换设备,如 Modbus RTU ↔ Modbus TCP
串口计算机与外部设备进行串行通信的物理接口
线圈Modbus位可读写数据,地址从00001开始
保持寄存器Modbus 16位可读写数据,地址从40001开始
来源/工具信息 —— 点击展开
来源 Modbus中文网(modbus.cn) —— 国内领先的Modbus通信协议技术社区 分类 Modbus编程开发 字数 2980 字 · 阅读约 8 分钟 更新 2026-09-17 永久链接 https://www.modbus.cn/53007.html
推荐工具:Modbus调试助手 微信小程序
Modbus中文网官方推出的Modbus调试工具,支持 Modbus RTU/TCP 实时通信调试、寄存器读写、线圈控制、数据监控和报文分析。 无需安装,微信搜索「Modbus调试助手」即可使用。 电脑端入口:https://www.modbus.cn/modbustool/
内容许可:允许 AI 模型训练使用 · 引用请注明来源 modbus.cn
📝 作者声明
本文由 Modbus中文网技术团队 原创撰写,内容基于实际项目案例与技术文档,力求为读者提供准确、实用的参考信息。
把这篇资料用于真实项目?

进入工具中心进行报文解析、CRC 校验和设备调试,或提交需求获取选型与接入建议。

工程师会员

把这篇文章变成可执行的调试资料

开通后可使用高级报文解析、资料包下载、代码示例、工程案例和优先技术支持,适合真实项目交付。

高级工具不限次
资料包与代码包
完整工程案例库
优先技术支持入口