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

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.데이터store import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequential데이터Block,
)

# 创建데이터块：주소0-99初始值为0
# ModbusSequential데이터Block(시작주소, [初始值列表])
coils = ModbusSequential데이터Block(0, [False] * 100)       # 코일（기능 코드01/05/15）
discrete_inputs = ModbusSequential데이터Block(0, [False] * 100)  # 이산 입력（02）
holding_registers = ModbusSequential데이터Block(0, [0] * 100)    # 유지 레지스터（03/06/16）
input_registers = ModbusSequential데이터Block(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,
    주소=("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",
      "주소": "0.0.0.0",
      "ignore_missing_devices": true,
      "allow_reuse_주소": true,
      "reconnect_delay": 2
    }
  },
  "device_list": {
    "my_device": {
      "setup": {
        "coils": {"시작ing_주소": 0, "size": 100},
        "discrete_inputs": {"시작ing_주소": 0, "size": 100},
        "holding_registers": {"시작ing_주소": 0, "size": 100},
        "input_registers": {"시작ing_주소": 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_시작=True 表示启动后不阻塞，可继续执行其他代码
    await simulator.run_forever(only_시작=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.데이터store import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequential데이터Block,
)

# 创建데이터存储
holding_registers = ModbusSequential데이터Block(0, [0] * 100)
coils = ModbusSequential데이터Block(0, [False] * 100)
discrete_inputs = ModbusSequential데이터Block(0, [False] * 100)
input_registers = ModbusSequential데이터Block(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.데이터store import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequential데이터Block,
)

def create_slave(unit_id, temp, hum):
    """创建一个슬레이브，模拟온습도센서"""
    hr = ModbusSequential데이터Block(0, [0] * 100)
    co = ModbusSequential데이터Block(0, [False] * 100)
    di = ModbusSequential데이터Block(0, [False] * 100)
    ir = ModbusSequential데이터Block(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, 주소=("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(주소=0, count=10, device_id=1)
if not result.is오답():
    print("유지 레지스터值：", result.registers)
    for i, val in enumerate(result.registers):
        print(f"  레지스터{i}：{val}")
else:
    print("읽기失败：", result)

# ====== 기능 코드01：读코일 ======
result = client.read_coils(주소=0, count=8, device_id=1)
if not result.is오답():
    print("코일 상태：", result.bits[:8])

# ====== 기능 코드04：读입력 레지스터 ======
result = client.read_input_registers(주소=0, count=4, device_id=1)
if not result.is오답():
    print("입력 레지스터：", result.registers)

# ====== 기능 코드02：读이산 입력 ======
result = client.read_discrete_inputs(주소=0, count=8, device_id=1)
if not result.is오답():
    print("이산 입력：", result.bits[:8])

# ====== 기능 코드06：写单个레지스터 ======
client.write_register(주소=100, value=1234, device_id=1)
print("已쓰기레지스터100=1234")

# ====== 기능 코드05：写单个코일 ======
client.write_coil(주소=0, value=True, device_id=1)  # True=ON, False=OFF
print("已쓰기코일0=ON")

# ====== 기능 코드16：写多个레지스터 ======
client.write_registers(주소=200, values=[10, 20, 30, 40, 50], device_id=1)
print("已批量쓰기레지스터200-204")

# ====== 기능 코드15：写多个코일 ======
client.write_coils(주소=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)

# 유지 레지스터 읽기 (FC03)
result = client.read_holding_registers(주소=0, count=5, device_id=1)
if not result.is오답():
    print("레지스터 값：", result.registers)
else:
    print("错误：", result)

# 쓰기
client.write_register(주소=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.is오답():
        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 Value오답(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.데이터store import (
    ModbusServerContext, ModbusSlaveContext, ModbusSequential데이터Block
)

# 全局데이터块引用，用于动态更新
hr_blocks = {}

def create_device(device_id, temp_init, hum_init):
    hr = ModbusSequential데이터Block(0, [0] * 100)
    co = ModbusSequential데이터Block(0, [False] * 100)
    di = ModbusSequential데이터Block(0, [False] * 100)
    ir = ModbusSequential데이터Block(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_데이터_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_데이터_change, daemon=True).시작()

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, 주소=("0.0.0.0", 502))

9.2 第二步：编写采集程序

# 데이터_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(
            주소=0, count=3, device_id=device_id
        )
        if result.is오답():
            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:
                데이터 = collector.read_sensor(dev_id)
                if 데이터:
                    print(f"[{name}] 온도：{데이터['temperature']:.1f}°C, "
                          f"湿度：{데이터['humidity']:.1f}%, 状态：{데이터['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 데이터_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_데이터block.py - 自定义데이터块实现设备逻辑
from pymodbus.데이터store import ModbusSequential데이터Block

class Temperature데이터Block(ModbusSequential데이터Block):
    """自定义온도데이터块，읽기时自动계산当前온도"""

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

    def getValues(self, 주소, 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(주소, count)

    def setValues(self, 주소, values):
        """重写setValues，쓰기时验证范围"""
        if 주소 == 0:  # 온도设定值
            temp = values[0]
            if not (100 <= temp <= 400):  # 10.0°C - 40.0°C
                raise Value오답("온도超出范围")
        super().setValues(주소, values)

11.2 异常响应模拟

# 模拟设备返回异常响应，测试클라이언트的错误处理能力
from pymodbus.exceptions import Illegal주소오답, Illegal기능오답

# 在自定义데이터块中抛出异常
class Faulty데이터Block(ModbusSequential데이터Block):
    def getValues(self, 주소, count=1):
        # 주소50-59模拟非法주소
        if 50 <= 주소 < 60:
            raise Illegal주소오답(f"주소{주소}不可访问")
        return super().getValues(주소, 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 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, 주소=("0.0.0.0", 502))	相同（매개변수名略有变化）
异步클라이언트	from pymodbus.client.asynchronous.tcp import AsyncModbusTCPClient	from pymodbus.client import AsyncModbusTcpClient
unit매개변수	unit=1	device_id=1
코일值	0xFF00 / 0x0000	True / 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管理界面，可以实时보기和修改레지스터 값，模拟设备行为变化；通过自定义데이터块，可以实现复杂的设备逻辑模拟。本文提供的模拟器搭建、클라이언트개발、데이터转换和完整实战案例可直接用于实际项目개발，建议先用模拟器验证程序逻辑，再接入真实设备。