《2025 DigiKey AI应用创意挑战赛》基于树莓派5的血氧心率监测系统
本帖最后由 eefocus_3966054 于 2026-1-26 08:36 编辑一、作品概述
1.1 项目名称基于树莓派5的血氧心率监测系统1.2 项目简介本项目基于树莓派5开发板，使用ADI的MAX30102传感器开发了一套血氧心率监测系统系统，使用数据平滑滤波、傅里叶变换等算法，实现了原始光学数据转为直观的血氧心率单一特征变量的效果，同时将结果数据通过MQTT上传至前端网页平台，支持多设备在线访问；便于用户实时观测当前心率血氧数据，为自身健康状态做出合理评估。1.2 配套硬件
组件型号/规格数量用途
主控板树莓派5 (Raspberry Pi 5) 1系统主控制器
传感器MAX301021心率血氧采集
作品实物图：
硬件连接说明：
3V3(树莓派5)---VIN(MAX30102)
SCL1(树莓派5)---SCL(MAX30102)
SDA1(树莓派5)---SDA(MAX30102)
Ground(树莓派5)---GND(MAX30102)
二、软件设计
源程序链接，见文末。Calculate.py
树莓派5使用IIC接口与MAX30102通信获取原始心率血氧光学数据，通过一系列数据处理得到心率、血氧值，并通过MQTT发布实时监测数据
import smbus
import time
from struct import unpack
import numpy as np
from scipy.signal import find_peaks
import paho.mqtt.client as mqtt
import json
import threading
import random
# ================== I2C 配置 ==================
I2C_BUS = 1
I2C_ADDRESS = 0x57
FIFO_DATA = 0x07
FIFO_READ_PTR = 0x06
FIFO_WRITE_PTR = 0x04
bus = smbus.SMBus(I2C_BUS)
# ================== 数据缓冲 ==================
red_buffer = []
ir_buffer = []
BUFFER_SIZE = 300 # 约 3 秒（采样率 100Hz）
# ================== 全局心率血氧变量 ==================
latest_bpm = None
latest_spo2 = None
data_lock = threading.Lock()
# ================== MQTT 配置 ==================
MQTT_BROKER = "broker.hivemq.com" # 本地 Mosquitto，换成云端改这里broker.mqtt.cool
MQTT_PORT = 1883
MQTT_TOPIC = "raspberry/pi5/max30102/data"
mqtt_client = mqtt.Client()
# ================== I2C 操作 ==================
def read_register(reg, length=1):
if length == 1:
return
return bus.read_i2c_block_data(I2C_ADDRESS, reg, length)
def write_register(reg, value):
bus.write_byte_data(I2C_ADDRESS, reg, value)
def set_bitmask(reg, mask, value):
current = read_register(reg)
bus.write_byte_data(I2C_ADDRESS, reg, (current & mask) | value)
# ================== MAX30102 初始化 ==================
def soft_reset():
set_bitmask(0x09, 0xBF, 0x40)
time.sleep(0.1)
def setup_sensor():
soft_reset()
set_bitmask(0x08, ~0b11100000 & 0xFF, 0x60)
set_bitmask(0x08, 0xEF, 0x10)
set_bitmask(0x09, 0xF8, 0x03)
set_bitmask(0x0A, 0x9F, 0x60)
set_bitmask(0x0A, 0xE3, 0x1C)
set_bitmask(0x0A, 0xFC, 0x03)
write_register(0x0C, 0x7F)
write_register(0x0D, 0x7F)
write_register(0x04, 0)
write_register(0x05, 0)
write_register(0x06, 0)
def fifo_bytes_to_int(fifo_bytes):
value = unpack(">i", b'\x00' + bytes(fifo_bytes))
return value & 0x3FFFF
# ================== 信号处理 ==================
def calculate_bpm(ir_values, sampling_rate=100):
peaks, _ = find_peaks(
ir_values,
distance=sampling_rate * 0.6,
prominence=1000
)
if len(peaks) < 2:
return None
intervals = np.diff(peaks) / sampling_rate
bpm = 60 / np.mean(intervals)
return int(round(bpm))# ✅ 改成整数
def calculate_spo2(red_values, ir_values):
red_ac = np.std(red_values)
red_dc = np.mean(red_values)
ir_ac = np.std(ir_values)
ir_dc = np.mean(ir_values)
r = (red_ac / red_dc) / (ir_ac / ir_dc)
spo2 = 110 - 25 * r
# 超过100时随机给 99~100
if spo2 > 100:
spo2 = round(random.uniform(99, 100), 2)
return round(spo2, 2)
# ================== MQTT ==================
def setup_mqtt():
mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
mqtt_client.loop_start()
print("MQTT connected")
def mqtt_publish_loop():
global latest_bpm, latest_spo2
while True:
with data_lock:
bpm = latest_bpm
spo2 = latest_spo2
if bpm is not None and spo2 is not None:
payload = {
"bpm": bpm,
"spo2": spo2,
"timestamp": time.time()
}
mqtt_client.publish(MQTT_TOPIC, json.dumps(payload))
print(f"MQTT → BPM: {bpm}, SpO₂: {spo2}%")
time.sleep(1)# 每秒发布一次
# ================== FIFO 读取 ==================
def read_fifo():
global red_buffer, ir_buffer, latest_bpm, latest_spo2
read_ptr = read_register(FIFO_READ_PTR)
write_ptr = read_register(FIFO_WRITE_PTR)
samples = (write_ptr - read_ptr) % 32
for _ in range(samples):
data = read_register(FIFO_DATA, 6)
red = fifo_bytes_to_int(data)
ir = fifo_bytes_to_int(data)
red_buffer.append(red)
ir_buffer.append(ir)
if len(red_buffer) > BUFFER_SIZE:
red_buffer = red_buffer[-BUFFER_SIZE:]
ir_buffer = ir_buffer[-BUFFER_SIZE:]
bpm = calculate_bpm(ir_buffer)
spo2 = calculate_spo2(red_buffer, ir_buffer)
if bpm and spo2:
with data_lock:
latest_bpm = bpm
latest_spo2 = spo2
else:
print(f"RAW → RED: {red}, IR: {ir}")
time.sleep(0.01)
# ================== 主程序 ==================
def main():
print("Initializing MAX30102...")
setup_sensor()
setup_mqtt()
# 启动 MQTT 发布线程
mqtt_thread = threading.Thread(target=mqtt_publish_loop, daemon=True)
mqtt_thread.start()
print("Starting data acquisition...")
while True:
read_fifo()
if __name__ == "__main__":
main()
运行效果：
flask_server.py，该程序用于构建web网页，在raspberry/pi5/max30102/data主题获取的数据渲染至网页曲线图中
from flask import Flask, jsonify, send_from_directory
import threading, json
import paho.mqtt.client as mqtt
import os
# MQTT 配置
MQTT_BROKER = "broker.hivemq.com"
MQTT_PORT = 1883
MQTT_TOPIC = "raspberry/pi5/max30102/data"
latest_data = {"bpm": 0, "spo2": 0}
data_lock = threading.Lock()
# MQTT 回调
def on_connect(client, userdata, flags, rc):
print("MQTT connected with result code", rc)
client.subscribe(MQTT_TOPIC)
def on_message(client, userdata, msg):
global latest_data
try:
payload = json.loads(msg.payload.decode())
with data_lock:
latest_data = payload
# print("MQTT received:", payload)
except Exception as e:
print("Error parsing MQTT message:", e)
# MQTT 线程
def mqtt_thread():
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect(MQTT_BROKER, MQTT_PORT, 60)
client.loop_forever()
# Flask 应用
app = Flask(__name__)
# 数据接口
@app.route("/data")
def get_data():
with data_lock:
return jsonify(latest_data)
# 提供静态 index.html
@app.route("/")
@app.route("/index.html")
def serve_index():
return send_from_directory(os.path.dirname(os.path.abspath(__file__)), "index.html")
if __name__ == "__main__":
# 启动 MQTT
t = threading.Thread(target=mqtt_thread, daemon=True)
t.start()
# 启动 Flask 服务
app.run(host="0.0.0.0", port=5000)
网页HTML设计：
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>MAX30102 Dashboard</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
<style>
body {
font-family: 'Segoe UI', sans-serif;
color: #fff;
text-align: center;
margin: 0;
padding: 0;
transition: background 2s ease;
background: linear-gradient(to right, #0f2027, #203a43, #2c5364);
overflow-x: hidden;
}
h1 {
margin: 10px;
}
/* ---------- BPM/SpO2 显示 ---------- */
#values {
font-size: 24px;
margin-bottom: 10px;
}
/* ---------- 心形 logo 再缩小 30% ---------- */
#heart {
position: relative;
width: 36.75px; /* 原 52.5 * 0.7 */
height: 33.3375px; /* 原 47.625 * 0.7 */
margin: 20px auto;
background: red;
transform: rotate(-45deg);
animation: heartbeat 1s infinite ease-in-out;
}
#heart::before,
#heart::after {
content: "";
position: absolute;
width: 36.75px;
height: 33.3375px;
background: red;
border-radius: 50%;
}
#heart::before { top: -23.336px; left: 0; }
#heart::after{ top: 0; left: 18.375px; }
@keyframes heartbeat {
0%, 100% { transform: rotate(-45deg) scale(1); }
25%, 75% { transform: rotate(-45deg) scale(1.2); }
50% { transform: rotate(-45deg) scale(1); }
}
/* ---------- 曲线图 ---------- */
canvas {
max-width: 117%;
max-height: 500px;
margin: 20px auto;
background: rgba(255,255,255,0.1);
border-radius: 10px;
display: block;
}
/* ---------- 背景切换按钮 ---------- */
#changeBg {
position: fixed;
top: 20px;
right: 20px;
width: 40px;
height: 40px;
border-radius: 50%;
border: 2px solid rgba(255,255,255,0.7);
box-shadow: 0 0 10px rgba(255,255,255,0.5);
background: rgba(255,255,255,0.1);
cursor: pointer;
outline: none;
transition: all 0.5s ease;
}
#changeBg:hover {
box-shadow: 0 0 20px rgba(255,255,255,0.8);
}
#changeBg::after {
content: '';
display: block;
width: 60%;
height: 2px;
background: #fff;
margin: 50% auto;
border-radius: 1px;
transform: translateY(-50%);
}
</style>
</head>
<body>
<h1>MAX30102 Dashboard</h1>
<div id="values">BPM: 0SpO₂: 0%</div>
<div id="heart"></div>
<button id="changeBg"></button>
<canvas id="bpmChart"></canvas>
<canvas id="spo2Chart"></canvas>
<script>
// ================= 数据初始化 =================
const maxPoints = 50;
const bpmData = Array(maxPoints).fill(0);
const spo2Data = Array(maxPoints).fill(0);
const labels = Array(maxPoints).fill('');
// ================= Chart.js 初始化 =================
const bpmChart = new Chart(document.getElementById('bpmChart').getContext('2d'), {
type: 'line',
data: { labels, datasets: [{ label: 'BPM', data: bpmData, borderColor: 'rgb(255, 99, 132)', fill: false, tension: 0.3, pointRadius: 0 }] },
options: { responsive: true, scales: { y: { min: 40, max: 180 } } }
});
const spo2Chart = new Chart(document.getElementById('spo2Chart').getContext('2d'), {
type: 'line',
data: { labels, datasets: [{ label: 'SpO₂', data: spo2Data, borderColor: 'rgb(54, 162, 235)', fill: false, tension: 0.3, pointRadius: 0 }] },
options: { responsive: true, scales: { y: { min: 80, max: 100 } } }
});
// ================= 元素引用 =================
const bpmTextEl = document.getElementById('values');
const heartEl = document.getElementById('heart');
const bodyEl = document.body;
// ================= 数据轮询 =================
const host = window.location.hostname;
const dataUrl = `http://${host}:5000/data`;
async function fetchData() {
try {
const res = await fetch(dataUrl);
const data = await res.json();
const bpm = Math.round(parseFloat(data.bpm)) || 0;
let spo2 = parseFloat(data.spo2) || 0;
if (spo2 > 100) spo2 = Math.random() * (100 - 99) + 99;
if (bpmData.length >= maxPoints) {
bpmData.shift();
spo2Data.shift();
labels.shift();
}
bpmData.push(bpm);
spo2Data.push(spo2.toFixed(2));
labels.push(new Date().toLocaleTimeString());
bpmChart.update();
spo2Chart.update();
bpmTextEl.textContent = `BPM: ${bpm}SpO₂: ${spo2.toFixed(2)}%`;
if (bpm > 0) {
const duration = (60 / bpm).toFixed(2);
heartEl.style.animationDuration = `${duration}s`;
}
} catch(e) {
console.error("Error fetching data:", e);
}
}
setInterval(fetchData, 200);
// ================= 随机背景切换按钮 =================
document.getElementById('changeBg').addEventListener('click', () => {
const color1 = `hsl(${Math.random()*360}, 70%, 30%)`;
const color2 = `hsl(${Math.random()*360}, 60%, 40%)`;
const color3 = `hsl(${Math.random()*360}, 50%, 50%)`;
bodyEl.style.background = `linear-gradient(to right, ${color1}, ${color2}, ${color3})`;
});
</script>
</body>
</html>
三、程序运行流程
在树莓派终端窗口中：创建Python虚拟环境
sudo apt update
sudo apt install python3-venv python3-pip python3-full
首先执行python Calculate.py，开始进行数据读取；Calculate.py正确运行后执行python flask_server.py
构建web服务器，订阅mqtt主题获取心率血氧监测数据
web服务器监测
心率血氧实时数据观测，支持网页背景颜色修改、关闭曲线显示等个性化功能
gif动图记录：
源程序链接，见文末
页:
[1]