树莓派4上基于Python使用MAX30101传感器实现准确血氧饱和度(SpO2)计算的方法咨询

最近我入手了一块MAX30101 breakout传感器，搭配树莓派4用Python开发。目前用SparkFun提供的sparkfun-qwiic-max3010x库和传感器通信、采集数据，这个Python库的示例里只有心率计算的代码，但没有血氧饱和度（SpO2）的实现；不过他们的Arduino库是同时包含心率和SpO2计算示例的。

我一直在琢磨，有没有合适的算法能在树莓派4的Python环境里准确计算SpO2？

我参考过MAX30101用户手册里的算法，自己写了一段代码试了试，但结果总是不准——有时候SpO2数值会变成负数，连续几次的读数差异也特别大。下面是我写的代码：

import time
import numpy as np
import qwiic_max3010x

# Initialize the MAX30101 sensor
sensor = qwiic_max3010x.QwiicMax3010x()

sensor.begin()
sensor.setup()

sensor.setPulseAmplitudeGreen(0)

# Function to read raw values from the sensor
def read_raw_values(num_samples=100):
    red_values = []
    ir_values = []

    for _ in range(num_samples):
            red = sensor.getRed()
            ir = sensor.getIR()
            red_values.append(red)
            ir_values.append(ir)
            time.sleep(0.01) # Adjust the delay as needed
    
    sensor.shutDown()
    
    return np.array(red_values), np.array(ir_values)


# Function to calculate AC and DC components
def calculate_ac_dc(raw_values):
    dc = np.mean(raw_values)  # DC component
    ac = raw_values - dc  # AC component
    return ac, dc


def calculate_spo2(red_ac, red_dc, ir_ac, ir_dc):
    r = (np.mean(red_ac) / red_dc) / (np.mean(ir_ac) / ir_dc)
    spo2 = 104 - (17 * r)
    return spo2


# Main loop
try:
    while True:
        # Read raw values
        red_raw, ir_raw = read_raw_values(num_samples=100)

        # Calculate AC and DC components
        red_ac, red_dc = calculate_ac_dc(red_raw)
        ir_ac, ir_dc = calculate_ac_dc(ir_raw)

        spo2 = calculate_spo2(red_ac, red_dc, ir_ac, ir_dc)

        print("Spo2: ", spo2)

        # Wait for a bit before the next reading
        time.sleep(1)  # Adjust the sleep time as needed

except KeyboardInterrupt:
    print("Program stopped.")

问题分析与改进方案 #

你的代码逻辑比较基础，直接用简单均值计算AC/DC比值，没有处理PPG信号里的噪声、基线漂移，也没有针对性提取心跳带来的AC峰值分量，这是结果不稳定的核心原因。SparkFun的Arduino库中SpO2计算是经过验证的，基于朗伯-比尔定律，结合了信号滤波、峰值检测和比值校准逻辑，我们可以把这部分逻辑移植到Python中，以下是改进后的代码：

import time
import numpy as np
import qwiic_max3010x

# 初始化传感器
sensor = qwiic_max3010x.QwiicMax3010x()
sensor.begin()
sensor.setup()
sensor.setPulseAmplitudeGreen(0)

# 配置参数（参考Arduino库逻辑）
SAMPLING_RATE = 100  # 采样率(Hz)
BUFFER_SIZE = 100    # 单次缓存样本数
SPO2_SAMPLE_COUNT = 4  # 用于稳定结果的样本组数

# 缓存与状态变量
red_buffer = []
ir_buffer = []
spo2_buffer = []

def moving_average(data, window_size=5):
    """滑动平均滤波：降低高频噪声"""
    return np.convolve(data, np.ones(window_size)/window_size, mode='valid')

def find_peaks(data):
    """PPG信号峰值检测：提取心跳引起的AC分量峰值"""
    peaks = []
    # 跳过首尾边缘，避免误判
    for i in range(1, len(data)-1):
        if data[i] > data[i-1] and data[i] > data[i+1]:
            peaks.append(data[i])
    return peaks

def calculate_spo2(red_peaks, ir_peaks, red_dc, ir_dc):
    """基于峰值比值计算SpO2（参考MAX30101官方校准公式）"""
    if len(red_peaks) < 2 or len(ir_peaks) < 2:
        return None
    
    # 计算多组峰值的R比值
    r_values = []
    for r_p, ir_p in zip(red_peaks[:-1], ir_peaks[:-1]):
        r = (r_p / red_dc) / (ir_p / ir_dc)
        r_values.append(r)
    
    avg_r = np.mean(r_values)
    # 官方校准公式，可根据实际佩戴情况微调系数
    spo2 = 100 - (25 * avg_r)
    # 限制结果在合理范围（0-100%）
    return max(0, min(100, spo2))

try:
    while True:
        # 采集样本至缓存满
        while len(red_buffer) < BUFFER_SIZE:
            red = sensor.getRed()
            ir = sensor.getIR()
            red_buffer.append(red)
            ir_buffer.append(ir)
            time.sleep(1/SAMPLING_RATE)
        
        # 滑动滤波处理原始信号
        red_filtered = moving_average(red_buffer)
        ir_filtered = moving_average(ir_buffer)
        
        # 提取DC分量（滤波后信号的均值）
        red_dc = np.mean(red_filtered)
        ir_dc = np.mean(ir_filtered)
        
        # 检测峰值
        red_peaks = find_peaks(red_filtered)
        ir_peaks = find_peaks(ir_filtered)
        
        # 计算SpO2并稳定结果
        spo2 = calculate_spo2(red_peaks, ir_peaks, red_dc, ir_dc)
        if spo2 is not None:
            spo2_buffer.append(spo2)
            # 取最近N次结果的平均值，提升稳定性
            if len(spo2_buffer) >= SPO2_SAMPLE_COUNT:
                avg_spo2 = np.mean(spo2_buffer[-SPO2_SAMPLE_COUNT:])
                print(f"SpO2: {avg_spo2:.1f}%")
        
        # 滑动窗口更新缓存，保留后半段数据，减少重复采集
        red_buffer = red_buffer[-int(BUFFER_SIZE/2):]
        ir_buffer = ir_buffer[-int(BUFFER_SIZE/2):]
        
        time.sleep(0.5)

except KeyboardInterrupt:
    sensor.shutDown()
    print("程序已停止。")

关键改进点说明 #

• 滑动平均滤波：有效降低PPG信号中的高频噪声，让波形更平滑；
• 峰值检测：精准提取心跳引起的AC分量峰值，比简单均值差更符合PPG信号的生理特征；
• 合理范围限制：避免出现负数或超出100%的异常结果；
• 滑动窗口缓存：减少重复采样，提升实时性；
• 多样本平均：多次结果取平均，进一步稳定读数。

额外实操建议 #

1. 传感器佩戴：确保传感器紧贴皮肤（不要过紧），避开毛发浓密区域，尽量在静止状态下测量，运动/晃动会严重干扰PPG信号；
2. 参数微调：不同佩戴位置、环境光下，可微调滤波窗口大小、峰值检测逻辑或SpO2校准公式的系数；
3. 完整移植官方逻辑：如果需要更高精度，可以对照Arduino库中的spo2.c文件，把其中的复杂滤波、阈值判断等逻辑完整移植过来。

备注：内容来源于stack exchange，提问作者Ish18