Modbus TCP/RTU 从入门到实战：Python 采集引擎完整实现 | 工业IoT系列·第1章

Q: 连接返回 "No response"**

先看设备 IP 通不通（`ping`），再看从机地址对不对（设备手册查默认地址），最后检查串口 A/B 线有没有接反。

Q: 读出来的值是实际值的 10 倍**

设备手册上写了寄存器值是"实际值 × 10"，配 `scale: 0.1` 就行。

Q: 一个 RS485 总线上挂多个设备怎么配？**

同一个串口配置下配不同的 `slave_id`。引擎会自动轮询。

> 工业IoT系列开篇。不讲概念，直接写一个能在生产环境跑的 Modbus 数据采集引擎，支持 TCP 和 RTU 两种协议。

> 全文约 5000 字，代码可复制运行。文末有本文配套资源和延伸阅读。

---

1. Modbus 极简入门（够用就行） {#1}

Modbus 是工业领域最通用的通信协议，没有之一。PLC、电表、温湿度传感器、UPS 基本都支持。

两种物理层：

| 类型 | 接线 | 传输距离 | 速率 | 适用场景 |

|------|------|---------|------|---------|

核心概念：

主机(Master) ←→ 从机(Slave)
主动轮询        被动响应

从机地址 (Slave ID)：1-247，每个设备一个
功能码 (Function Code)：03 读保持寄存器（只用这个就够搞定 90% 场景）
寄存器地址：从 0 开始的偏移量
寄存器值：16 位整数（可组合成 32 位浮点/整数）

> 绝大多数场景只用到 03（读保持寄存器），配合设备手册的数据地址表就够了。不用把整个协议栈背下来。

---

2. 环境搭建 {#2}

python3 -m venv venv
source venv/bin/activate
pip install pymodbus==3.6.8 pyserial loguru

三行命令搞定 Python 环境。如果你有 USB 转 RS485（FTDI 芯片最稳，CH340 也行），一个转换器十几块钱。

没有物理设备也能验证：本文第5节提供了模拟器代码，在本地跑一个虚拟 Modbus 设备，数据会定时变化，效果和真实设备一样。

---

3. 引擎架构设计 {#3}

┌─────────────┐   ┌─────────────┐
│  TCP轮询器  │   │  RTU轮询器  │
├──────┬──────┤   ├──────┬──────┤
│      │           │      │
└──────▼───────────▼──────┘
          │
    ┌─────▼──────┐
    │  数据管道   │
    │ Console/   │
    │ JSON/      │
    │  MQTT      │
    └─────┬──────┘
          │
    输出到文件/数据库/消息队列

三个核心抽象：轮询器（与设备通信）→ 管道（数据处理）→ 输出（数据去向）。换输出方式不改采集逻辑，加设备只改配置不改代码。

---

4. 核心代码实现 {#4}

4.1 Modbus 通信客户端

# client.py
from pymodbus.client import ModbusTcpClient, ModbusSerialClient
from loguru import logger

class ModbusClientPool:
    """Modbus 连接池，支持 TCP 和 RTU"""

    def __init__(self, config: dict):
        self.clients = {}
        for device_id, cfg in config.get("devices", {}).items():
            mode = cfg.get("mode", "tcp")
            if mode == "tcp":
                client = ModbusTcpClient(
                    host=cfg["host"], port=cfg.get("port", 502), timeout=cfg.get("timeout", 5)
                )
            elif mode == "rtu":
                client = ModbusSerialClient(
                    port=cfg["port"], baudrate=cfg.get("baudrate", 9600),
                    bytesize=cfg.get("bytesize", 8), parity=cfg.get("parity", "N"),
                    stopbits=cfg.get("stopbits", 1), timeout=cfg.get("timeout", 5)
                )
            self.clients[device_id] = client

    def read_holding_registers(self, device_id: str, slave_id: int, address: int, count: int):
        """读取保持寄存器（功能码03）"""
        client = self.clients.get(device_id)
        if not client:
            return None
        try:
            result = client.read_holding_registers(address=address, count=count, slave=slave_id)
            return list(result.registers) if result and not result.isError() else None
        except Exception as e:
            logger.error(f"读取失败 [{device_id}]: {e}")
            return None

核心逻辑就是这段。无论你用 TCP 还是 RTU，接口都一样。写一次，两种协议都能用。

4.2 寄存器解析（易错点）

Modbus 寄存器存的是原始 16 位整数，怎么转成真实值？这是初学者掉坑最多的地方。

# parser.py
import struct

def parse_registers(raw: list[int], data_type: str) -> float:
    """Modbus 寄存器原始值转实际数值"""
    if data_type == "uint16":
        return float(raw[0])
    elif data_type == "int16":
        val = raw[0]
        return float(val - 65536) if val > 32767 else float(val)
    elif data_type == "float32":
        packed = struct.pack(">HH", raw[0], raw[1])
        return struct.unpack(">f", packed)[0]
    return 0.0

常见坑：读出来的值比实际大 10 倍或 0.1 倍 → 看设备手册的寄存器定义，配比例系数即可。比如电压值是"实际值 × 10"存储的，你就除以 10。

全套解析代码（含 uint32、int32、float32、缩放系数）和完整的引擎框架代码，我在 agent.bycims5g.cn 上放了可直接下载的版本，附真实项目的配置文件示例。

4.3 采集调度

# collector.py
import time, threading
from loguru import logger

class CollectorEngine:
    def __init__(self, pool, points: list):
        self.pool = pool
        self.points = points
        self.callbacks = []
        self.running = False

    def on_data(self, callback):
        self.callbacks.append(callback)

    def collect_once(self):
        """执行一次完整采集"""
        results = []
        for pt in self.points:
            try:
                device_id = pt.get("device_id", "default")
                raw = self.pool.read_holding_registers(device_id, pt["slave_id"], pt["address"], pt.get("count", 1))
                value = parse_registers(raw, pt.get("data_type", "uint16")) if raw else None
                scaled = round(value * pt.get("scale", 1.0) + pt.get("offset", 0.0), 3) if value else None
                results.append({"name": pt["name"], "value": scaled, "unit": pt.get("unit", ""), 
                                "quality": "good" if scaled is not None else "bad"})
            except Exception as e:
                results.append({"name": pt["name"], "value": None, "quality": "bad"})
        for cb in self.callbacks:
            cb(results)
        return results

    def start(self, interval: float = 5.0):
        self.running = True
        def _loop():
            while self.running:
                t0 = time.time()
                collected = self.collect_once()
                elapsed = time.time() - t0
                ok = sum(1 for v in collected if v["quality"] == "good")
                logger.info(f"采集: {ok}/{len(collected)} OK, {elapsed:.2f}s")
                time.sleep(max(0.1, interval - elapsed))
        threading.Thread(target=_loop, daemon=True).start()

大约 80 行代码，核心框架就跑起来了。

---

5. 启动与验证 {#5}

5.1 配置

# config.yaml
devices:
  iot_gateway:
    mode: tcp
    host: 192.168.1.100
    port: 502
    timeout: 3

points:
  - name: "A相电压"
    device_id: iot_gateway
    slave_id: 1
    address: 0
    count: 2
    data_type: float32
    scale: 0.1
    unit: "V"
  - name: "温度"
    slave_id: 2
    address: 0
    count: 1
    data_type: int16
    scale: 0.1
    unit: "°C"

你只需要改 IP 和采集点定义，不需要改代码。

5.2 模拟器模式（无设备也能跑）

想先试试但手头没有 Modbus 设备？加一个模拟模式：

# demo_mode.py
from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext
import threading, random, time

store = ModbusSlaveContext(zero_mode=True)
context = ModbusServerContext(slaves=store, single=True)

def update_data():
    while True:
        context[0].setValues(3, 0, [
            int(2200 + random.randint(-50, 50)), 0,
            int(50 + random.randint(-10, 10)), 0,
        ])
        context[0].setValues(3, 10, [int(255 + random.randint(-5, 5))])
        time.sleep(3)

threading.Thread(target=update_data, daemon=True).start()
StartTcpServer(context=context, address=("0.0.0.0", 5020))

跑起来后你本地就有一个虚拟电表和温度传感器，每 3 秒数据自动变化，效果跟接了一台真实设备一样。

5.3 主程序入口

# main.py
import yaml
from client import ModbusClientPool
from collector import CollectorEngine

with open("config.yaml") as f:
    config = yaml.safe_load(f)

pool = ModbusClientPool(config)
engine = CollectorEngine(pool, config["points"])

# 输出到控制台
def print_data(values):
    for v in values:
        q = "✓" if v["quality"] == "good" else "✗"
        print(f"{q} {v['name']}: {v['value']} {v['unit']}")

engine.on_data(print_data)
engine.start(interval=5)

input("按 Enter 停止\n")

5.4 运行效果

[10:30:01] 采集: 2/2 OK, 0.01s
  ✓ A相电压: 220.3 V
  ✓ 温度: 25.7 °C

从搭建到跑通，半小时以内。

---

6. 生产部署踩坑记录 {#6}

说几个实际踩过的坑。

6.1 TCP 连接保活

Modbus TCP 连接空闲 30-60 秒会被设备断开。不加保活的话，第二天来发现数据停在了昨晚某个时间点。

解法：每隔 30 秒发一次空读查询。

6.2 串口设备名乱跳

USB 转 RS485 插拔后 /dev/ttyUSB0 可能变成 /dev/ttyUSB1，采集程序就连不上了。

解法：用 udev 规则绑定串口序列号，让系统固定生成一个别名设备名。这样不管插哪个 USB 口，设备名不变。

6.3 批量读取提升 10 倍性能

50 个采集点逐点读取要 15 秒。把地址连续的寄存器合并成一次批量读取，1.5 秒搞定。

6.4 日志轮转

采集引擎跑一天能产生几百 MB 日志。不加轮转的话磁盘很快就满了。

from loguru import logger
logger.add("logs/modbus_{time:YYYY-MM-DD}.log", rotation="1 day", retention="30 days", compression="gz")

> 更完整的踩坑清单、本文完整代码、以及一个 150+ 测点的实际项目配置案例，我都整理在 agent.bycims5g.cn 上了，需要的可以去看。

---

7. 常见问题 {#7}

Q: 连接返回 "No response"

A: 先看设备 IP 通不通（ping），再看从机地址对不对（设备手册查默认地址），最后检查串口 A/B 线有没有接反。

Q: RTU 读出来全是 0

A: 大概率是波特率或校验位不对。换着试试 9600/4800，N/E 校验。

Q: 读出来的值是实际值的 10 倍

A: 设备手册上写了寄存器值是"实际值 × 10"，配 scale: 0.1 就行。

Q: 一个 RS485 总线上挂多个设备怎么配？

A: 同一个串口配置下配不同的 slave_id。引擎会自动轮询。

---

下章预告

系列二·第2章：边缘端跑 AI——RK3588 部署 7B 模型的完整踩坑记录。讲的是采集上来的数据怎么在设备本地做推理，不等云端。

如果这些内容对你有帮助，可以关注这个系列，每周更新 1-2 篇。遇到落地问题也欢迎交流，我们的案例和更多实战内容在 agent.bycims5g.cn。

---

本文代码完整版和实际项目案例可在 agent.bycims5g.cn 获取。有问题欢迎评论区讨论。

Modbus TCP/RTU 从入门到实战：Python 采集引擎完整实现 | 工业IoT系列·第1章

Modbus TCP/RTU 从入门到实战：Python 采集引擎完整实现 | 工业IoT系列·第1章

目录

1. Modbus 极简入门（够用就行） {#1}

2. 环境搭建 {#2}

3. 引擎架构设计 {#3}

4. 核心代码实现 {#4}

4.1 Modbus 通信客户端

4.2 寄存器解析（易错点）

4.3 采集调度

5. 启动与验证 {#5}

5.1 配置

5.2 模拟器模式（无设备也能跑）

5.3 主程序入口

5.4 运行效果

6. 生产部署踩坑记录 {#6}

6.1 TCP 连接保活

6.2 串口设备名乱跳

6.3 批量读取提升 10 倍性能

6.4 日志轮转

7. 常见问题 {#7}

下章预告