使用pigpio读取MMA8452Q传感器时随机出现I2C读取失败问题排查

我用Tkinter GUI搭配MMA8452Q加速度传感器搭建了一套系统，设置传感器达到阈值时触发中断标志，随后采集指定时长的数据并通过GUI绘图。但在读取数据或调用I2C读取寄存器的过程中，会随机出现“I2C read failed”错误，比如在check_flags()方法中报错，具体报错信息如下：

interrupt 1 True
4 0 1810275105
Exception in thread Thread-2:
Traceback (most recent call last):
  File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.7/threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "/home/pi/.../tkinter/mix/mix_interrupt.py", line 307, in check_flags
    pi.i2c_read_byte_data(acc_sensor, FF_MT_SRC)
  File "/usr/lib/python3/dist-packages/pigpio.py", line 2874, in i2c_read_byte_data
    return _u2i(_pigpio_command(self.sl, _PI_CMD_I2CRB, handle, reg))
  File "/usr/lib/python3/dist-packages/pigpio.py", line 1011, in _u2i
    raise error(error_text(v))
pigpio.error: 'I2C read failed'

错误并非每次都会出现，而是随机发生。若错误出现在try-except代码块中，数据仍可绘图，但会存在数据缺失，导致绘图不完整。怀疑是代码问题或波特率设置问题，以下是关键代码及修改后的带线程锁代码，寻求问题原因及解决方法。

关键代码 #

线程创建 #

threading.Thread(target=self.check_flags).start() #This is in the init method

中断标志检查函数 #

def check_flags(self):
        while(True):
            #print("check the flags")
            if (self.interrupt_flag == True):
                pi.i2c_read_byte_data(acc_sensor, FF_MT_SRC)
                global scale_variable
                self.x_graph = []
                self.y1_graph = []
                self.y2_graph = []
                self.y3_graph = []
                
                a = datetime.datetime.now()
                one_time = True 
                start_time = time.time()*1000
                while(time.time()*1000-start_time) < self.timeframe:
                    try:
                        b = datetime.datetime.now()
                        delta = b-a
                        y1 = self.readACCx()
                        y2 = self.readACCy()
                        y3 = self.readACCz()
                        self.y1_graph.append(y1)
                        self.y2_graph.append(y2)
                        self.y3_graph.append(y3)
                        self.x_graph.append(delta.total_seconds()*1000)
                        if (y1 > self.threshold or y2 > self.threshold): # this is for catching the whole movement if its longer than the set timeframe
                            self.timeframe += 5          
                    except pigpio.error as e:
                        print("error:", e)
                self.save_data()
                self.plot_data()
                self.interrupt_flag = False

X轴数据读取函数 #

def readACCx(self): #reads x axis value 12 bit
        global scale_variable
        comp_acc_x2 = pi.i2c_read_byte_data(acc_sensor, OUT_X_MSB)
        comp_acc_x1 = pi.i2c_read_byte_data(acc_sensor, OUT_X_LSB)
        acc_combined = ((comp_acc_x2 << 8) | comp_acc_x1) >>4
        if acc_combined < 2048:
            acc_combined = acc_combined*scale_variable
            return acc_combined
        else:
            acc_combined = acc_combined - 4096
            acc_combined = acc_combined*scale_variable
            return acc_combined

中断回调函数 #

def my_callback1(self, gpio, level, tick): #first interrupt function
        self.interrupt_flag = True
        print("interrupt 1", self.interrupt_flag)
        
        print(gpio, level, tick)

修改后的带线程锁代码 #

protect = threading.Lock() # this is outside of any class at the top

def check_flags(self): # new data capture method
        while(True):
            #print("check the flags")
            if (self.interrupt_flag == True):
                protect.acquire()
                ff_flag = pi.i2c_read_byte_data(acc_sensor, FF_MT_SRC)
                print("interrupt cleared", ff_flag)
                global scale_variable
                self.x_graph = []
                self.y1_graph = []
                self.y2_graph = []
                self.y3_graph = []
                
                a = datetime.datetime.now()
                one_time = True 
                start_time = time.time()*1000
                while(time.time()*1000-start_time) < self.timeframe:
                    try:
                        b = datetime.datetime.now()
                        delta = b-a
                        y1 = self.readACCx()
                        y2 = self.readACCy()
                        y3 = self.readACCz()
                        self.y1_graph.append(y1)
                        self.y2_graph.append(y2)
                        self.y3_graph.append(y3)
                        self.x_graph.append(delta.total_seconds()*1000)
                        if (y1 > self.threshold or y2 > self.threshold):
                            self.timeframe += 5
                    except pigpio.error as e:
                        print("error:", e)
                self.save_data()
                self.plot_data()
                self.interrupt_flag = False
                self.timeframe = self.config.getint('setting', 'timeframe')
                print("scale_variable: ", scale_variable)
                protect.release()

问题原因及解决方法 #

1. 线程资源竞争未完全解决 #

你添加的线程锁protect只覆盖了check_flags开头的中断标志清除和后续采集流程，但readACCx、readACCy、readACCz这些函数里的I2C读取操作并没有被锁保护。这些函数在采集循环里被频繁调用，可能和其他I2C操作（比如中断标志读取）发生冲突，导致随机失败。

解决：把所有I2C读写操作都放到锁的保护范围内。比如修改readACCx：

def readACCx(self): #reads x axis value 12 bit
        global scale_variable
        protect.acquire()
        try:
            comp_acc_x2 = pi.i2c_read_byte_data(acc_sensor, OUT_X_MSB)
            comp_acc_x1 = pi.i2c_read_byte_data(acc_sensor, OUT_X_LSB)
        finally:
            protect.release()
        acc_combined = ((comp_acc_x2 << 8) | comp_acc_x1) >>4
        if acc_combined < 2048:
            acc_combined = acc_combined*scale_variable
            return acc_combined
        else:
            acc_combined = acc_combined - 4096
            acc_combined = acc_combined*scale_variable
            return acc_combined

同理修改readACCy和readACCz，确保所有I2C交互都被同步。

2. I2C波特率过高导致通信不稳定 #

树莓派默认I2C波特率是400kHz，部分传感器在这个速率下可能出现通信丢包，尤其是长距离或干扰大的环境。

解决：降低I2C波特率，编辑/boot/config.txt文件，添加或修改：

dtparam=i2c_arm_baudrate=100000

保存后重启树莓派，使用100kHz的标准速率通信。

3. 中断标志清除未做异常处理 #

当前代码中pi.i2c_read_byte_data(acc_sensor, FF_MT_SRC)这一步没有异常捕获，一旦这里出错会直接终止线程，导致后续无法响应中断。

解决：给这一步添加异常处理：

protect.acquire()
try:
    ff_flag = pi.i2c_read_byte_data(acc_sensor, FF_MT_SRC)
    print("interrupt cleared", ff_flag)
except pigpio.error as e:
    print("Failed to clear interrupt flag:", e)
    protect.release()
    self.interrupt_flag = False
    continue
# 后续采集代码...

4. 多次单字节读取增加通信开销 #

readACCx等函数每次读取两个寄存器都分两次调用i2c_read_byte_data，频繁的I2C启动/停止会增加出错概率。

解决：改用批量读取寄存器的方式，一次性读取所有加速度数据：

def read_all_acc(self):
    global scale_variable
    protect.acquire()
    try:
        # 从OUT_X_MSB开始，一次性读取6个字节（X/Y/Z各两个字节）
        data = pi.i2c_read_i2c_block_data(acc_sensor, OUT_X_MSB, 6)
    finally:
        protect.release()
    if not data:
        raise pigpio.error("Failed to read acceleration data")
    
    # 解析X轴
    acc_x = ((data[0] << 8) | data[1]) >> 4
    acc_x = acc_x if acc_x < 2048 else acc_x - 4096
    acc_x *= scale_variable
    
    # 解析Y轴
    acc_y = ((data[2] << 8) | data[3]) >> 4
    acc_y = acc_y if acc_y < 2048 else acc_y - 4096
    acc_y *= scale_variable
    
    # 解析Z轴
    acc_z = ((data[4] << 8) | data[5]) >> 4
    acc_z = acc_z if acc_z < 2048 else acc_z - 4096
    acc_z *= scale_variable
    
    return acc_x, acc_y, acc_z

然后在采集循环里替换为：

y1, y2, y3 = self.read_all_acc()

这样能大幅减少I2C通信次数，降低出错概率。

5. 中断回调与线程的同步问题 #

中断回调函数直接设置self.interrupt_flag = True，可能和check_flags线程中的self.interrupt_flag = False发生竞态条件。

解决：给interrupt_flag的访问也加上锁保护，或者使用threading.Event替代布尔变量，Event本身是线程安全的：

# 初始化时替换为
self.interrupt_event = threading.Event()

# 中断回调里改为
def my_callback1(self, gpio, level, tick):
    self.interrupt_event.set()
    print("interrupt 1", self.interrupt_event.is_set())
    print(gpio, level, tick)

# check_flags里改为
while(True):
    if self.interrupt_event.is_set():
        protect.acquire()
        try:
            # 清除中断标志等操作...
            # 最后重置event
            self.interrupt_event.clear()
        finally:
            protect.release()

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