《2025 DigiKey AI应用创意挑战赛》采用ADI的MAXREFDES117血氧采集系统
本方案基于 ADI MAXREFDES117# 血氧传感器模块与树莓派5构建的 纯 Python 血氧检测系统设计。由于安装MAX3010x Adafruit CircuitPython库一直不成功,所以整个系统 不依赖 Adafruit CircuitPython 库,仅使用标准 Linux I²C 驱动和科学计算库。## 一、系统设计概述
所用器材从得捷电子官网采购
树莓派5
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/084500l84b29j0m47nxxy4.png)
ADI MAXREFDES117#ADI MAXREFDES117#
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/084713tufx6hsobiuitxs6.png)
### 核心目标
- 实时采集 PPG(光电容积脉搏波)信号(RED + IR)
- 计算 **心率(HR, BPM)** 和 **血氧饱和度(SpO₂, %)**
- 提供 Web 实时可视化界面
-
- !(https://www.eefocus.com/forum/data/attachment/forum/202602/02/082507bmciaqahjmdhmgqa.png)
硬件连接(MAXREFDES117# → Raspberry Pi 5)
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/081917cawe5ilc8pwpciea.png)
**必须使用 3.3V!** MAX30102 不支持 5V!
### 📦 软件栈
| 组件 | 说明 |
|------|------|
| Python 3.11+ | 系统自带 |
| `smbus2` | I²C 通信(替代过时的 smbus)|
| `numpy` / `scipy` | 信号处理 |
| `Flask` | 轻量级 Web 服务 |
| 自定义驱动 | 直接操作 MAX30102 寄存器 |
---
## 二、项目结构
```bash
~/spo2/
├── venv/ # 虚拟环境
├── max30102.py # MAX30102 驱动
├── spo2_core.py # 核心算法(HR/SpO₂ 计算)
├── app.py # Flask Web 服务
└── templates/
└── index.html # 实时监控界面
```
---
## 三、安装与配置
### 步骤 1:启用 I²C 接口
```bash
sudo raspi-config
# → Interface Options → I2C → Yes
```
### 步骤 2:创建虚拟环境并安装依赖
```bash
cd ~
mkdir spo2 && cd spo2
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip
pip install smbus2 flask numpy scipy
```
### 步骤 3:验证硬件连接
```bash
i2cdetect -y 1
```
看到:
shumeipai@raspberrypi:~ $ i2cdetect -y 1
0123456789abcdef
00: -- -- -- -- -- -- -- --
10: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
20: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
30: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
40: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
50: -- -- -- -- -- -- -- 57 -- -- -- -- -- -- -- --
60: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
70: -- -- -- -- -- -- -- --
硬件连接正常
---
## 四、核心代码实现
### 文件 1:`max30102.py` —— MAX30102 驱动
```python
# max30102.py
import time
from smbus2 import SMBus
class MAX30102:
"""ADI MAX30102 驱动(专为 MAXREFDES117# 设计)"""
def __init__(self, bus=1, address=0x57):
self.bus = SMBus(bus)
self.addr = address
self.reset()
self.setup()
def write_reg(self, reg, value):
try:
self.bus.write_byte_data(self.addr, reg, value)
except Exception as e:
raise RuntimeError(f"I2C Write Error @0x{reg:02X}: {e}")
def read_fifo(self):
"""读取 RED + IR 数据(18-bit)"""
try:
data = self.bus.read_i2c_block_data(self.addr, 0x07, 6)
red = ((data << 16) | (data << 8) | data) & 0x03FFFF
ir = ((data << 16) | (data << 8) | data) & 0x03FFFF
return red, ir
except Exception as e:
raise RuntimeError(f"FIFO Read Error: {e}")
def reset(self):
"""软件复位"""
self.write_reg(0xFF, 0x01)
time.sleep(1.0)
def setup(self):
"""初始化寄存器(MAXREFDES117# 推荐配置)"""
# FIFO 配置
self.write_reg(0x08, 0x40)# 滚动模式
# MODE_CONFIG: SpO2 模式
self.write_reg(0x09, 0x03)
# SPO2_CONFIG: 50Hz, 16-bit, 4096nA
self.write_reg(0x0A, 0x27)
# LED 电流: RED=45mA, IR=45mA
self.write_reg(0x0C, 0x2C)
self.write_reg(0x0D, 0x2C)
# 清空 FIFO
for _ in range(32):
self.read_fifo()
```
---
### 文件 2:`spo2_core.py` —— 信号处理与生理参数计算
```python
# spo2_core.py
import numpy as np
from scipy import signal
class Spo2Processor:
def __init__(self, fs=50, buffer_sec=5):
self.fs = fs
self.buffer_size = int(fs * buffer_sec)
self.red_buffer = np.zeros(self.buffer_size)
self.ir_buffer = np.zeros(self.buffer_size)
self.idx = 0
def add_sample(self, red, ir):
"""添加新样本到循环缓冲区"""
self.red_buffer = red
self.ir_buffer = ir
self.idx = (self.idx + 1) % self.buffer_size
def get_window(self):
"""获取最新完整窗口数据"""
start = (self.idx + 1) % self.buffer_size
red_win = np.concatenate(, self.red_buffer[:start]])
ir_win = np.concatenate(, self.ir_buffer[:start]])
return red_win, ir_win
def bandpass_filter(self, data, low=0.7, high=3.5):
"""0.7-3.5 Hz 带通滤波(心率有效频段)"""
nyq = 0.5 * self.fs
b, a = signal.butter(2, , btype='band')
return signal.filtfilt(b, a, data)
def calculate_hr(self, ir_signal):
"""计算心率(BPM)"""
dc = np.mean(ir_signal)
if dc < 8000:# 无手指阈值
return None, "NO_FINGER"
ac = self.bandpass_filter(ir_signal - dc)
if np.std(ac) < 100:
return None, "WEAK_SIGNAL"
# 峰值检测
height = np.std(ac) * 0.6
peaks, _ = signal.find_peaks(ac, height=height, distance=self.fs//2)
if len(peaks) < 3:
return None, "LOW_PEAKS"
rr_intervals = np.diff(peaks) / self.fs
valid_rr = rr_intervals[(rr_intervals > 0.3) & (rr_intervals < 1.5)]
if len(valid_rr) < 2:
return None, "IRREGULAR"
hr = 60.0 / np.median(valid_rr)
return hr if 40 <= hr <= 180 else None, "OK"
def calculate_spo2(self, red_signal, ir_signal):
"""计算 SpO₂ (%)"""
dc_red, dc_ir = np.mean(red_signal), np.mean(ir_signal)
if min(dc_red, dc_ir) < 8000:
return None, "NO_FINGER"
ac_red, ac_ir = np.std(red_signal), np.std(ir_signal)
if ac_ir < 10:
return None, "INVALID_SIGNAL"
R = (ac_red / dc_red) / (ac_ir / dc_ir)
spo2 = 104 - 17 * R# 临床校准公式
if 70 <= spo2 <= 100 and 0.3 < R < 3.0:
return spo2, "OK"
return None, "OUT_OF_RANGE"
```
---
### 文件 3:`app.py` —— Flask Web 服务
```python
# app.py
import threading
import time
from flask import Flask, render_template, jsonify
from max30102 import MAX30102
from spo2_core import Spo2Processor
app = Flask(__name__)
sensor = MAX30102()
processor = Spo2Processor()
hr, spo2 = None, None
status = "INITIALIZING"
def sensor_loop():
global hr, spo2, status
while True:
try:
red, ir = sensor.read_fifo()
processor.add_sample(red, ir)
# 每秒更新一次结果
if int(time.time()) % 1 == 0:
red_win, ir_win = processor.get_window()
hr, hr_status = processor.calculate_hr(ir_win)
if hr_status == "OK":
spo2, spo2_status = processor.calculate_spo2(red_win, ir_win)
status = "OK" if spo2_status == "OK" else spo2_status
else:
status = hr_status
except Exception as e:
status = f"ERROR:{str(e)[:20]}"
time.sleep(0.02)# 50Hz
@app.route('/')
def index():
return render_template('index.html')
@app.route('/api/data')
def get_data():
return jsonify({
'heart_rate': round(hr, 1) if hr else None,
'spo2': round(spo2, 1) if spo2 else None,
'status': status,
'timestamp': time.time()
})
if __name__ == '__main__':
# 启动传感器线程
threading.Thread(target=sensor_loop, daemon=True).start()
print(" 血氧系统启动: http://<RPI_IP>:5000")
app.run(host='0.0.0.0', port=5000, debug=False)
```
---
### 文件 4:`templates/index.html` —— Web 界面
```html
<!-- templates/index.html -->
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>MAXREFDES117# 血氧监测</title>
<style>
body { font-family: Arial; text-align: center; background: #f0f8ff; padding: 20px; }
.card { display: inline-block; margin: 10px; padding: 20px; border-radius: 10px; width: 200px; }
.hr { background: #ffebee; border: 2px solid #f44336; }
.spo2 { background: #e8f5e9; border: 2px solid #4caf50; }
.value { font-size: 3em; font-weight: bold; margin: 10px 0; }
.label { color: #555; }
.status { padding: 10px; margin-top: 20px; border-radius: 5px; }
.ok { background: #d4edda; color: #155724; }
.warn { background: #fff3cd; color: #856404; }
.error { background: #f8d7da; color: #721c24; }
</style>
</head>
<body>
<h1>🩺 MAXREFDES117# 血氧监测系统</h1>
<div class="card hr">
<div class="label">心率 (BPM)</div>
<div id="hr-value" class="value">--</div>
</div>
<div class="card spo2">
<div class="label">血氧饱和度 (%)</div>
<div id="spo2-value" class="value">--</div>
</div>
<div id="status" class="status"></div>
<script>
function updateData() {
fetch('/api/data')
.then(res => res.json())
.then(data => {
document.getElementById('hr-value').textContent =
data.heart_rate ? data.heart_rate : '--';
document.getElementById('spo2-value').textContent =
data.spo2 ? data.spo2 : '--';
const statusDiv = document.getElementById('status');
statusDiv.textContent = getStatusText(data.status);
statusDiv.className = 'status ' + getStatusClass(data.status);
})
.catch(console.error);
}
function getStatusText(status) {
const map = {
'OK': '正常',
'NO_FINGER': ' 请放置手指',
'WEAK_SIGNAL': ' 信号弱,请保持静止',
'LOW_PEAKS': '信号不稳定',
'OUT_OF_RANGE': ' 数值异常',
'IRREGULAR': '心律不齐'
};
return map || status;
}
function getStatusClass(status) {
if (status === 'OK') return 'ok';
if (status.startsWith('ERROR')) return 'error';
if (status.includes('⚠️')) return 'warn';
return 'error';
}
setInterval(updateData, 1000);
updateData(); // 立即首次加载
</script>
</body>
</html>
```
---
## 五、运行系统
```bash
cd ~/spo2
source venv/bin/activate
python app.py
```
在浏览器访问:`http://<IP>:5000`
> **使用提示**:
> 1. 手指**完全覆盖**传感器窗口
> 2. 测量时**保持绝对静止**
> 3. 避免强光直射传感器
> 4. 首次使用需校准(静置 15 秒)
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/083831ng9vn7h09hdvzzbu.png)
**说明,由于安装MAX3010x Adafruit CircuitPython库一直不成功,导致重做方案时间延误,系统还在调试中。**
页:
[1]