PMS5003传感器反复无数据返回问题求助（代码排查）

问题现象 #

使用PMS5003传感器时，最初偶尔在读数间隙出现device reports readiness to read but returned no data(device disconnected or multiple access on port)错误，后续报错频率超过正常读数。目前故障呈现固定规律：每成功读取2次数据，就会出现3次无数据返回提示。已更换两次传感器及线缆，确认串口连接正常，怀疑代码读取逻辑存在问题。

代码中的核心问题 #

• 串口超时参数缺失：__init__方法未接收并传递timeout参数到串口初始化，导致serial.read(32)无超时限制，数据不同步时会触发阻塞或报错。
• 强制清空缓冲区破坏数据帧：每次读取前调用flushInput()会清空串口缓冲区，可能丢弃传感器正在发送的完整帧，导致后续读取到不完整数据，无法匹配帧头0x42 0x4d。
• MQTT客户端未初始化：__init__中未创建mqtt.Client()实例，直接调用self.client会引发错误；发布消息时使用未定义的mqtt_client变量。
• 无帧头校验逻辑：直接读取32字节，若缓冲区数据不是完整帧，会返回无效数据，没有循环查找有效帧头的机制，导致频繁读取失败。

修复后的代码 #

import json
import paho.mqtt.client as mqtt
import serial
import time

class PMSensor:
    def __init__(self, serial_port, baud_rate, bytesize, parity, stopbits, mqtt_broker, mqtt_port, mqtt_username, mqtt_password, mqtt_topic, timeout=2.0):
        # 串口配置
        self.serial_port = serial_port
        self.baud_rate = baud_rate
        self.bytesize = bytesize
        self.stopbits = stopbits
        self.parity = parity
        self.timeout = timeout
        
        # MQTT配置
        self.mqtt_broker = mqtt_broker
        self.mqtt_port = mqtt_port
        self.mqtt_username = mqtt_username
        self.mqtt_password = mqtt_password
        self.mqtt_topic = mqtt_topic
        self.ser = None
        self.client = mqtt.Client()  # 初始化MQTT客户端
        
    def on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            print('Connected to MQTT broker.')
            print("--------------------------------------")
            client.subscribe(self.mqtt_topic)
        else:
            print('Failed to connect to MQTT broker with error code:', rc)
            
    def on_disconnect(self, client, userdata, rc):
        print("Disconnected from MQTT broker with result code: " + str(rc))
        
    def connect_mqtt(self):
        self.client.username_pw_set(self.mqtt_username, self.mqtt_password)
        self.client.connect(self.mqtt_broker, self.mqtt_port)

    def establish_serial_connection(self):
        try:
            # 加入timeout参数
            self.ser = serial.Serial(
                self.serial_port, 
                self.baud_rate, 
                self.bytesize, 
                self.parity, 
                self.stopbits,
                timeout=self.timeout
            )
            if self.ser.is_open:
                print('Serial connection established:', self.ser.portstr)
                return True
            else:
                print('Failed to establish serial connection.')
                return False
        except Exception as e:
            print('Error while establishing serial connection:', e)
            return False
        
    def read_pm25(self):
        try:
            if not self.ser.is_open:
                return None
            
            # 循环查找有效帧头0x42 0x4d
            while True:
                byte = self.ser.read(1)
                if byte == b'\x42':
                    next_byte = self.ser.read(1)
                    if next_byte == b'\x4d':
                        # 读取剩余30字节组成完整帧
                        remaining_data = self.ser.read(30)
                        if len(remaining_data) == 30:
                            data = b'\x42\x4d' + remaining_data
                            break
                # 超时无数据则退出
                if not byte:
                    return None
            
            # 解析传感器数据
            apm1 = (data[4] << 8) + data[5]
            apm25 = (data[6] << 8) + data[7]
            apm10 = (data[8] << 8) + data[9]
            pm1 = (data[10] << 8) + data[11]
            pm25 = (data[12] << 8) + data[13]
            pm10 = (data[14] << 8) + data[15]
            gt03um = (data[16] << 8) + data[17]
            gt05um = (data[18] << 8) + data[19]
            gt10um = (data[20] << 8) + data[21]
            gt25um = (data[22] << 8) + data[23]
            gt50um = (data[24] << 8) + data[25]
            gt100um = (data[26] << 8) + data[27]
            return apm1, apm25, apm10, pm1, pm25, pm10, gt03um, gt05um, gt10um, gt25um, gt50um, gt100um
        except Exception as e:
            print('Error while reading PM sensor:', e)
            return None
        
    def display_in_shell(self, apm1, apm25, apm10, pm1, pm25, pm10, gt03um, gt05um, gt10um, gt25um, gt50um, gt100um):
        print("--------------------------------------")
        print("Concentration Units (Standard)")
        print(f'\n PM1.0 (ug/m3): {apm1} \t\t PM2.5 (ug/m3): {apm25} \t\t PM10 (ug/m3): {apm10}')
        print("--------------------------------------")
        print("Concentration Units (Environmental)")
        print(f'\nPM1.0 (ug/m3): {pm1} \t\t PM2.5 (ug/m3): {pm25} \t\t PM10 (ug/m3): {pm10}')
        print("--------------------------------------")
        print("Particle Count\n")
        print(f' >0.3um in 0.1L air: {gt03um}')
        print(f' >0.5um in 0.1L air: {gt05um}')
        print(f' >1.0um in 0.1L air: {gt10um}')
        print(f' >2.5um in 0.1L air: {gt25um}')
        print(f' >5.0um in 0.1L air: {gt50um}')
        print(f' >10um in 0.1L air: {gt100um}')
        print("--------------------------------------")

    def start(self):
        if self.establish_serial_connection():
            try:
                self.client.on_connect = self.on_connect
                self.client.on_disconnect = self.on_disconnect
                self.connect_mqtt()
                self.client.loop_start()
            except Exception as e:
                print("Could not establish connection with broker -->", e)

            while True:
                pm_data = self.read_pm25()
                if pm_data is not None:
                    apm1, apm25, apm10, pm1, pm25, pm10, gt03um, gt05um, gt10um, gt25um, gt50um, gt100um = pm_data
                    self.display_in_shell(apm1, apm25, apm10, pm1, pm25, pm10, gt03um, gt05um, gt10um, gt25um, gt50um, gt100um)
                    payload = json.dumps({
                        'apm1': apm1, 'apm25': apm25, 'apm10': apm10,
                        'pm1': pm1, 'pm25': pm25, 'pm10': pm10,
                        'gt03um': gt03um, 'gt05um': gt05um, 'gt10um': gt10um,
                        'gt25um': gt25um, 'gt50um': gt50um, 'gt100um': gt100um
                    })
                    self.client.publish(self.mqtt_topic, payload)
                else:
                    print('Failed to read PM sensor data.')
                time.sleep(1)
            
    def stop(self):
        self.client.loop_stop()
        self.client.disconnect()
        if self.ser and self.ser.is_open:
            self.ser.close()

# 实例化传感器
pms5003_sensor = PMSensor(
    serial_port='/dev/ttyUSB0',
    baud_rate=9600,
    bytesize=serial.EIGHTBITS,
    parity=serial.PARITY_NONE,
    stopbits=serial.STOPBITS_ONE,
    timeout=2.0,
    mqtt_broker="192.168.21.201",
    mqtt_port=1883,
    mqtt_username='WLED',
    mqtt_password='Toyota',
    mqtt_topic='TMNA/PEMC/FirstFloor/LivingWall/reTerminal/ParticulateSensor/Environmental'
)

pms5003_sensor.start()
# pms5003_sensor.stop()

修复说明 #

1. 补全超时配置：在串口初始化时加入timeout参数，避免读取操作无限阻塞。
2. 优化帧读取逻辑：移除强制清空缓冲区操作，循环查找有效帧头，确保每次读取完整的传感器数据帧。
3. 修复MQTT客户端问题：初始化mqtt.Client()实例，修正发布消息的变量引用错误。
4. 添加资源释放：在stop方法中关闭串口，确保资源正常回收。

内容的提问来源于stack exchange，提问作者DanieL Rodriguez