当前位置：首页 > news >正文

工业现场设备监控：树莓派串口通信从零实现

news 2026/6/16 22:38:12

用树莓派打通工业设备的“神经末梢”：从串口通信到远程监控的实战之路

你有没有遇到过这样的场景？

工厂里一堆老式温控仪、电表、PLC还在稳定运行，功能完好，但就是“哑巴”——没有网口、不支持以太网，数据出不来。想做数字化改造？换整套系统成本太高，不动又眼睁睁看着数据孤岛越积越多。

别急，今天我们不用花大钱买工控网关，也不需要复杂的组态软件。一块几十元的树莓派 + 一根RS-485线，就能让这些“沉默的老兵”开口说话。

这不是理论推演，而是我们已经在水泵房、配电柜、生产线跑通的真实方案。接下来，我会带你一步步实现：如何用Python写一个稳定可靠的串口采集程序，把Modbus设备的数据读出来，再传上云，最终做到手机端实时告警。

全程无PPT式讲解，只有真实代码、踩过的坑和现场调试心得。

为什么是树莓派？它真能扛住工业环境吗？

先说结论：在边缘侧做协议转换和轻量级处理，树莓派是目前性价比最高的选择之一。

很多人一听“树莓派”，第一反应是“玩具”。但别忘了，它是一台完整的Linux计算机——有ARM处理器、512MB以上内存、支持Wi-Fi/以太网、能跑Docker、Flask、InfluxDB……关键是，GPIO还带硬件UART。

更重要的是，它的生态太成熟了。你要对接Modbus？pyserial和minimalmodbus几行代码搞定；要上传云端？MQTT库一装就行；要做本地HMI？前端随便搭个Vue页面丢上去就行。

当然，工业现场不是实验室。电压波动、电磁干扰、连续运行七天七夜……这些问题我们都遇到过。但我们发现，只要做好三点：

加隔离模块（必须！）
启用看门狗（watchdog）防死机
合理设计轮询逻辑避免总线冲突

这套系统完全可以长期稳定运行。

硬件怎么接？别让第一根线就烧了树莓派！

这是最关键的一步，也是新手最容易翻车的地方。

树莓派的串口在哪？

树莓派板子上有两个关键引脚：
-TXD（GPIO14，物理引脚8）：发送数据
-RXD（GPIO15，物理引脚10）：接收数据

它们输出的是TTL电平（3.3V），而工业RS-485是差分信号（A/B线），电压可达±12V。直接连？轻则通信失败，重则烧毁GPIO。

所以必须加一个电平转换模块，推荐使用带光耦隔离的SP3485 或 MAX485 隔离版模块。价格十几块，但能保命。

接线图（超简版）

[Modbus传感器] --- A/B线 ---> [SP3485隔离模块] | RO ←--------| RXD (TTL) DI →--------| TXD (TTL) /RE+DE ------| 控制收发方向（可接GPIO） | GPIO15 ----| RXD to Pi GPIO14 ----| TXD to Pi

💡 小技巧：如果你用的是标准Modbus RTU设备，通常只需要三根线——A、B、GND。注意A/B极性别接反，否则通信不通。

特别提醒：别被蓝牙“偷走”你的串口！

从树莓派3B+开始，系统默认把主UART（/dev/ttyAMA0）分配给了蓝牙模块，留给用户的只剩一个mini-UART（/dev/ttyS0），性能不稳定。

解决办法很简单，在/boot/config.txt最后加上这句：

dtoverlay=disable-bt

然后重启，主串口就会回到ttyAMA0，通信更稳。同时记得关闭串口登录终端：

sudo systemctl disable serial-getty@ttyS0.service

Python串口通信实战：不只是`read()`和`write()`

现在进入正题：怎么用Python真正可靠地跟设备对话。

我们常用的库是pyserial，但它只是一个底层驱动，真正的难点在于协议封装、错误处理和时序控制。

下面是我在线上项目中打磨出来的核心类，比网上那些“demo级”代码更能扛住实际考验。

✅ 稳定可用的串口通信类（含日志、重试、缓冲清理）

import serial import time import logging from typing import Optional, Union logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) class ModbusRTUClient: def __init__(self, port: str = '/dev/ttyAMA0', baudrate: int = 9600): self.ser = serial.Serial( port=port, baudrate=baudrate, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=1.0, # 读超时 write_timeout=1.0 # 写超时 ) self.device_addr = 1 # 默认设备地址 logging.info(f"串口已打开: {port}@{baudrate}") def _calculate_crc16(self, data: bytes) -> bytes: """计算Modbus CRC16校验码""" crc = 0xFFFF for byte in data: crc ^= byte for _ in range(8): if crc & 0x0001: crc = (crc >> 1) ^ 0xA001 else: crc >>= 1 return crc.to_bytes(2, 'little') def read_holding_register(self, reg_start: int, reg_count: int = 1) -> Optional[list]: """读保持寄存器（功能码0x03）""" request = bytes([ self.device_addr, 0x03, reg_start >> 8, reg_start & 0xFF, reg_count >> 8, reg_count & 0xFF ]) request += self._calculate_crc16(request) try: # 清空输入输出缓冲区 self.ser.flushInput() self.ser.flushOutput() self.ser.write(request) logging.debug(f"→ 发送: {request.hex()}") # 等待响应（根据波特率调整延时） time.sleep(0.1 + 0.01 * reg_count) response = self.ser.read(5 + reg_count * 2) if len(response) < 5: logging.warning("⚠️ 响应数据过短") return None # 校验CRC recv_crc = response[-2:] calc_crc = self._calculate_crc16(response[:-2]) if recv_crc != calc_crc: logging.error(f"❌ CRC校验失败: 收到={recv_crc.hex()}, 计算={calc_crc.hex()}") return None # 解析数据 byte_count = response[2] if byte_count != len(response) - 5: logging.warning("⚠️ 数据长度不匹配") return None values = [] for i in range(reg_count): start = 3 + i * 2 val = int.from_bytes(response[start:start+2], 'big') values.append(val) return values except serial.SerialException as e: logging.error(f"串口异常: {e}") return None except Exception as e: logging.error(f"未知错误: {e}") return None def close(self): if self.ser.is_open: self.ser.close() logging.info("串口已关闭")

🔍 关键点解析：
动态CRC计算：不再硬编码校验值，而是每次请求前实时生成；
flush双清缓存：防止旧数据残留导致粘包；
智能延时等待：根据读取寄存器数量动态调整等待时间；
完整错误捕获链：覆盖断线、超时、CRC失败等常见问题；
日志分级输出：方便后期排查问题。

实战案例：读取一台温控仪的实时温度

假设我们有一台Modbus温控仪，参数如下：

设备地址：1
波特率：9600
寄存器0x0000 存放当前温度（单位0.1°C，即500表示50.0°C）

调用代码非常简单：

if __name__ == "__main__": client = ModbusRTUClient(port='/dev/ttyAMA0', baudrate=9600) try: while True: temps = client.read_holding_register(0x0000, 1) if temps is not None: temperature = temps[0] / 10.0 logging.info(f"✅ 当前温度: {temperature:.1f} °C") else: logging.warning("❌ 读取失败，将重试...") time.sleep(5) # 每5秒采集一次 except KeyboardInterrupt: logging.info("⏹ 用户中断") finally: client.close()

跑起来后你会看到类似日志：

2025-04-05 10:23:01 [INFO] 串口已打开: /dev/ttyAMA0@9600 2025-04-05 10:23:01 [DEBUG] → 发送: 010300000001c40b 2025-04-05 10:23:01 [INFO] ✅ 当前温度: 48.5 °C 2025-04-05 10:23:06 [DEBUG] → 发送: 010300000001c40b 2025-04-05 10:23:06 [INFO] ✅ 当前温度: 48.7 °C

看到这个“✅”，你就知道数据真的拿回来了。

多设备轮询怎么做？别让总线“打架”

单设备没问题，但现场往往有十几个节点挂在同一根RS-485总线上。如果一股脑全发出去，总线会冲突。

正确做法是：顺序轮询 + 超时控制 + 地址切换

devices = [ {"addr": 1, "name": "入口温度"}, {"addr": 2, "name": "出口压力"}, {"addr": 3, "name": "电机状态"} ] for dev in devices: client.device_addr = dev["addr"] data = client.read_holding_register(0x0000, 1) if data: print(f"{dev['name']}: {data[0]}") time.sleep(0.3) # 每次查询间隔至少300ms，给设备留响应时间

⚠️ 经验之谈：RS-485是半双工，同一时刻只能一人说话。太快轮询会导致前一个设备还没回话，下一个命令就已经发出去了，结果全乱套。

工业级部署要考虑什么？不只是“能跑就行”

当你准备把这套东西放进配电箱、贴上标签、交付客户时，以下几点必须考虑：

1. 断线自动恢复机制

网络可以断，串口也可能突然没信号。我们要加一个“心跳检测 + 自动重连”：

def is_responsive(self) -> bool: """通过读ID寄存器判断设备是否在线""" result = self.read_holding_register(0x0001, 1) return result is not None

配合定时任务，连续三次失败就尝试重启串口甚至整个服务。

2. 启用系统看门狗（Watchdog）

Linux自带watchdog服务，可监测进程是否卡死。安装并启动：

sudo apt install watchdog sudo systemctl enable watchdog

然后在主循环里定期“喂狗”：

with open('/dev/watchdog', 'w') as f: while running: # ... your logic ... f.write('1') # 刷新看门狗 time.sleep(10)

一旦程序卡住超过阈值（默认60秒），树莓派会自动重启。

3. 使用`minimalmodbus`简化开发（强烈推荐）

上面的手动实现虽然透明，但日常开发建议直接上minimalmodbus，专为Modbus RTU设计，API简洁到爆：

import minimalmodbus instrument = minimalmodbus.Instrument('/dev/ttyAMA0', slaveaddress=1) instrument.serial.baudrate = 9600 instrument.mode = minimalmodbus.MODE_RTU temperature = instrument.read_register(0, functioncode=3) / 10.0

一行代码搞定读取，还能自动处理CRC、重试、延时，省心太多。

数据往哪去？下一步可以这样走

拿到数据只是起点。真正的价值在于利用它。

你可以轻松扩展以下能力：

本地存储：用SQLite或InfluxDB记录历史数据；
远程告警：温度超标时通过微信/钉钉机器人推送通知；
可视化面板：用Grafana展示趋势曲线；
对接MES：通过HTTP API 把数据送给企业管理系统；
边缘判断：当电压异常时自动切断继电器。

一个小例子：用MQTT上传到ThingsBoard平台：

import paho.mqtt.client as mqtt client_mqtt = mqtt.Client() client_mqtt.connect("your-mqtt-server.com", 1883) payload = { "temperature": 48.5, "status": "running", "timestamp": int(time.time()) } client_mqtt.publish("sensor/plc01", json.dumps(payload))

几分钟之内，你的手机App就能看到实时数据流。