工位终端上线之后，第一件要做的活通常是把下面几台设备的数采上来。现场最常见的是 Modbus：电表、温控器、PLC、变频器、称重仪表，很多都支持 Modbus RTU 或者 Modbus TCP。这篇给一套能直接用的 Python 实现，跑在工业一体机的 Linux 系统上，做多设备轮询采集，并且带断线重连和超时保护。文中示例基于 pymodbus 3.x，实际版本以你环境里的接口为准。
## 一、环境准备
工业一体机上装的是 Linux（Ubuntu 或者 Debian 都行），先建虚拟环境，别直接往系统 Python 里装。
```bash
sudo apt update
sudo apt install -y python3-venv python3-pip
python3 -m venv /opt/daq/venv
/opt/daq/venv/bin/pip install pymodbus pyserial
```
如果是串口采集（RTU），确认设备节点存在，并给当前用户串口权限：
```bash
ls -l /dev/ttyS0 /dev/ttyUSB0
sudo usermod -aG dialout $USER
```
改完用户组要重新登录才生效。
## 二、配置文件：把点位表抽出来
不要把从站地址和寄存器地址写死在代码里。现场改点位是常事，抽成 YAML 或者 JSON，改完重启服务就行。
`devices.json`：
```json
{
"poll_interval": 2.0,
"devices": [
{
"name": "meter_1",
"mode": "tcp",
"host": "192.168.1.11",
"port": 502,
"unit": 1,
"points": [
{"key": "voltage", "func": "input", "addr": 0, "count": 1, "scale": 0.1},
{"key": "current", "func": "input", "addr": 1, "count": 1, "scale": 0.01}
]
},
{
"name": "temp_ctrl_1",
"mode": "rtu",
"port": "/dev/ttyS0",
"baudrate": 9600,
"parity": "N",
"stopbits": 1,
"bytesize": 8,
"unit": 2,
"points": [
{"key": "pv", "func": "holding", "addr": 100, "count": 1, "scale": 0.1}
]
}
]
}
```
`scale` 是工程量换算系数，很多仪表返回的是整数，除以 10 或者 100 才是实际的电压、温度。
## 三、核心采集循环
关键点有三个：每个设备独立超时、失败不阻塞其他设备、失败后自动重建连接。
```python
import json
import time
import logging
from pymodbus.client import ModbusTcpClient, ModbusSerialClient
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger("daq")
FUNC_MAP = {
"holding": "read_holding_registers",
"input": "read_input_registers",
}
def build_client(dev):
if dev["mode"] == "tcp":
return ModbusTcpClient(dev["host"], port=dev["port"], timeout=1)
return ModbusSerialClient(
port=dev["port"],
baudrate=dev["baudrate"],
parity=dev.get("parity", "N"),
stopbits=dev.get("stopbits", 1),
bytesize=dev.get("bytesize", 8),
timeout=1,
)
def read_device(dev):
"""读一台设备的所有点位，失败返回 None"""
client = build_client(dev)
if not client.connect():
log.warning("connect failed: %s", dev["name"])
return None
values = {}
try:
for pt in dev["points"]:
method = getattr(client, FUNC_MAP[pt["func"]])
rr = method(pt["addr"], count=pt["count"], slave=dev["unit"])
if rr.isError():
log.warning("read error: %s.%s", dev["name"], pt["key"])
return None
raw = rr.registers[0]
values[pt["key"]] = round(raw * pt.get("scale", 1), 3)
except Exception as exc: # noqa: BLE001
log.warning("exception on %s: %s", dev["name"], exc)
return None
finally:
client.close()
return values
def main():
with open("devices.json", "r", encoding="utf-8") as fp:
cfg = json.load(fp)
interval = float(cfg.get("poll_interval", 2.0))
cycle = 0
while True:
cycle += 1
for dev in cfg["devices"]:
data = read_device(dev)
if data is None:
# 单台失败不影响其他设备，下一轮自动重连
continue
log.info("[%s] %s", dev["name"], data)
# 这里接你的上报逻辑：写本地库 / 发 MQTT / 写文件
time.sleep(interval)
if __name__ == "__main__":
main()
```
这套写法里，每次采集都新建客户端、用完就关。看着有点笨，好处是连接状态永远干净——上一轮坏掉的连接不会带到下一轮。轮询周期在秒级以上的场景，这点开销可以忽略。如果你需要更短的周期，再把连接做成常驻并加心跳检测。
## 四、几个现场很容易踩的坑
第一个坑是 unit 和 slave 参数名。pymodbus 3.x 里原来的 `slave` 改成过 `unit`，不同小版本的参数名和位置有差异，报 `TypeError` 的时候先看这个。建议在项目里锁版本，别让自动升级把现场跑崩。
第二个坑是把点位的起始地址按 0 还是按 1 算。很多仪表手册写的是 1-based 地址，代码里用的是 0-based，中间差 1。现象是能读到数，但跟你以为的那个量对不上。第一次接入新设备时，逐个点位和手册核对一遍，别嫌慢。
第三个坑是串口多设备共线。RS485 总线上一串设备，必须保证波特率、校验、停止位一致。总线上如果有一台设备参数不对，可能拖着整条总线一起超时。排查的时候先把其他设备断开，一台一台接。
第四个坑是没有看门狗。采集程序卡死，画面还在转，但数据已经半天没更新了，这种情况最坑。Linux 上可以配 systemd 的 `WatchdogSec`，让服务异常时自动重启；或者在主循环里打时间戳，超时主动退出。
还有一种更隐蔽的情况：程序没死，但读到的值是"卡住的旧值"。原因是异常处理写得太宽，读到错误以后没有把值标记为无效，上层系统还在拿上一轮的数显示。处理办法是每个点位带一个时间戳和质量位，超过若干个轮询周期没更新就标记为坏质量，让上层知道这条数据不可信。工业现场很怕"看起来有数据"，因为没人会去核对，等到发现的时候可能已经跑了很久。
再补充一点关于日志的。现场排查全靠日志，所以日志要能定位到具体设备和具体点位。上面示例里每行都带了设备名，这不够，建议把点位 key 也打出来，并且把原始寄存器值和换算后的值都记一份。电压这种量，如果显示 220 而实际是 22，只有对比原始值才能立刻判断是 scale 写错了还是设备真返回了异常值。日志文件别无限增长，用 logging 的 RotatingFileHandler 做轮转，或者交给系统日志管理。
```ini
# /etc/systemd/system/daq.service
[Unit]
Description=Modbus DAQ
After=network.target
[Service]
ExecStart=/opt/daq/venv/bin/python /opt/daq/main.py
Restart=always
RestartSec=3
WatchdogSec=30
Type=notify
[Install]
WantedBy=multi-user.target
```
## 五、数据怎么往上走
采集完本地聚合，再往上推。推荐在本地留一份缓存，断网的时候先存，网络恢复补传。上层对接方式按现场选：MQTT 适合推给物联网平台，HTTP 适合对接自有系统，OPC UA 适合跟既有的 SCADA 打通。
有一点要提醒：别让采集线程和执行机构控制逻辑挤在同一个进程里抢资源。工业一体机如果同时要跑 HMI 和采集，建议把采集做成独立服务，通过本地 socket 或者共享文件交给 HMI 显示。这样谁卡住都不会拖垮另一个。
本地缓存的实现方式很多，轻量场景可以先用 SQLite 或者按天切分的文件，把带时间戳的数据落盘；网络恢复后由一个补传任务按时间顺序推送，并去重。要注意缓存盘的空间，长期断网时得有上限和清理策略，别把系统盘写满导致整个设备出问题。
硬件层面，工位终端选型时把串口数量、网口数量、宽温范围这些确认好，能少很多返工。广州微嵌的工业一体机和嵌入式计算机在接口配置上可以按需组合，常见的一体化采集终端配置能直接对上，现场对接时省事一些。
## FAQ
**问：pymodbus 报参数错误，提示 slave/unit 不对怎么办？**
答：这是版本差异导致的。pymodbus 3.x 中部分接口的参数名经历调整，建议锁定一个版本（在 requirements 里写死），并按该版本文档核对参数名，不要依赖自动升级。
**问：一台设备掉线会不会影响其他设备的采集？**
答：按上面的写法不会。每台设备独立建立连接、独立超时，失败返回 None 后跳过，下一轮自动重连。需要注意的是串口总线场景，一台设备参数不一致可能拖累整条总线。
**问：读回来的数值和实际值差一个倍数，是什么原因？**
答：多数是工程量换算系数没设对，需要在点位表里配置 scale。另外要确认寄存器地址的起始基准（0-based 还是 1-based），这两种情况都会导致数值对不上。
**问：怎么防止采集程序卡死不重启？**
答：可以用 systemd 的 Restart=always 配合 WatchdogSec 做看门狗；也可以在主循环里记录时间戳，超过阈值主动退出让服务重启。不要依赖人工发现程序无响应。
**问：采集终端的硬件选型有什么要注意的？**
答：按现场确认串口和网口的数量、是否需要宽温、安装方式和散热形式。广州微嵌的工业一体机和工业平板电脑在接口和尺寸上可以按需组合，配套多系统支持，做这类采集终端时适配起来省事一些。
