60GHz毫米波雷达人体呼吸心率检测Python仿真代码(含指夹血氧仪心电实时对比数据)

# -*- coding: utf-8 -*-"""指夹血氧/心率仪 HID 数据采集 + 雷达信号处理 工具整合GUI绘图、串口、HID设备通信、信号滤波功能"""# ===================== 1. 标准库导入 =====================import osimport timeimport threadingfrom typing import List, Optional, Tuplefrom dataclasses import dataclass, fieldfrom collections import dequefrom datetime import datetimefrom queue import Empty, Queue# ===================== 2. 第三方库导入 =====================import numpy as npimport serialfrom serial.tools import list_portsfrom scipy.io import savematfrom scipy.signal import butter, find_peaks, sosfilt, sosfiltfilt# 可选依赖：HID设备通信try:    import hidexcept ImportError:    hid = None# ===================== 3. GUI/绘图库导入 =====================import tkinter as tkfrom tkinter import filedialog, messagebox, ttkfrom matplotlib.backends.backend_tkagg import FigureCanvasTkAggfrom matplotlib.figure import Figurefrom matplotlib import rcParams# ===================== 4. Matplotlib 全局配置 =====================rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'Noto Sans CJK SC', 'Arial Unicode MS']rcParams['axes.unicode_minus'] = False# ===================== 5. 全局常量 =====================# 指夹血氧仪 HID 设备标识VID = 0x28E9PID = 0x028A# 设备初始化指令序列SEQ_A = [    "7d81a78080808080807d81a2808080808080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "82020000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "80000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "831a011c121d3844056a000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "9f1f0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "8e031100000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "81010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",]SEQ_B = [    "80000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "81010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "82020000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "9f1f0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "8e071500000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "8e031100000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "9f1f0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "9b001b00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",    "9b011c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",]# 心跳包配置KEEPALIVE_INTERVAL = 5.0KEEPALIVE_PAYLOAD64 = bytes.fromhex("9a1a" + "00" * 62)# ===================== 6. 通用工具函数 =====================def get_timestamp() -> str:    """获取格式化时间戳"""    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")def safe_db(x: np.ndarray) -> np.ndarray:    """安全计算分贝值，避免log(0)报错"""    return 20.0 * np.log10(np.maximum(np.abs(x), np.finfo(float).eps))# ===================== 7. 指夹血氧/心率仪 HID 采集模块 =====================def hex_to_64_bytes(hexstr: str) -> bytes:    """将十六进制字符串转换为64字节固定长度数据"""    data = bytes.fromhex(hexstr)    return (data + b"\x00" * 64)[:64]def write_report65(dev: "hid.device", payload64: bytes) -> int:    """向HID设备写入65字节报告（1字节报告ID + 64字节数据）"""    return dev.write(b"\x00" + payload64)def parse_eb_frame(data: bytes) -> Tuple[Optional[Tuple[int, int]], List[int]]:    """解析EB帧数据，提取心率、血氧和波形数据"""    vitals = None    wave_data = []    index = 0    data_len = len(data)    while index < data_len:        if data[index] != 0xEB:            index += 1            continue        # 解析心率血氧数据        if index + 8 <= data_len and data[index+1] == 0x01 and data[index+2] == 0x05:            vitals = (data[index+3], data[index+4])            index += 8            continue        # 解析波形数据        if index + 6 <= data_len and data[index+1] == 0x00:            value = data[index+3] | (data[index+4] << 8)            wave_data.append(value)            index += 6            continue        index += 1    return vitals, wave_data@dataclassclass FingerState:    """血氧仪状态数据类（线程安全）"""    hr: Optional[float] = None    spo2: Optional[float] = None    source: str = ""    last_update_time: float = field(default_factory=time.time)    lock: threading.Lock = field(default_factory=threading.Lock)class FingerClipReader(threading.Thread):    """血氧仪数据读取线程（后台守护线程）"""    def __init__(self, state: FingerState, stop_event: threading.Event, status_queue: Queue):        super().__init__(daemon=True)        self.state = state        self.stop_event = stop_event        self.status_queue = status_queue        self.device: Optional["hid.device"] = None    def log(self, msg: str) -> None:        """发送日志到状态队列"""        self.status_queue.put(("status", f"[{get_timestamp()}] [心电仪] {msg}"))    def send_init_sequence(self, seq: List[str], label: str) -> None:        """发送设备初始化指令序列"""        self.log(f"初始化 {label} 指令...")        for cmd in seq:            payload = hex_to_64_bytes(cmd)            write_report65(self.device, payload)            time.sleep(0.01)    def _safe_close(self) -> None:        """安全关闭HID设备连接"""        try:            if self.device:                self.device.close()        except Exception:            pass        finally:            self.device = None    def _stream_loop(self) -> bool:        """设备数据循环读取"""        last_keepalive = time.time()        last_rx_time = time.time()        last_spo2 = None        while not self.stop_event.is_set():            current_time = time.time()            # 发送心跳包            if current_time - last_keepalive >= KEEPALIVE_INTERVAL:                try:                    write_report65(self.device, KEEPALIVE_PAYLOAD64)                except Exception:                    return False                last_keepalive = current_time            # 读取设备数据            try:                raw_data = self.device.read(64, timeout_ms=200)            except Exception:                return False            if raw_data:                last_rx_time = current_time                byte_data = bytes(raw_data)                head = byte_data[0]                # 解析EB帧                if head == 0xEB:                    vitals, _ = parse_eb_frame(byte_data)                    if vitals:                        hr, spo2 = vitals                        last_spo2 = spo2                        with self.state.lock:                            self.state.hr = float(hr)                            self.state.spo2 = float(spo2)                            self.state.source = "EB0105"                            self.state.last_update_time = time.time()                # 解析F3帧（心率）                elif head == 0xF3 and len(byte_data) >= 3:                    hr_val = int(byte_data[2])                    if 35 <= hr_val <= 220:                        with self.state.lock:                            self.state.hr = float(hr_val)                            if last_spo2 is not None:                                self.state.spo2 = float(last_spo2)                            self.state.source = "F3"                            self.state.last_update_time = time.time()                # 解析F0帧（血氧）                elif head == 0xF0 and len(byte_data) >= 2:                    spo2_val = int(byte_data[1])                    if 50 <= spo2_val <= 100:                        last_spo2 = spo2_val                        with self.state.lock:                            self.state.spo2 = float(spo2_val)                            self.state.source = "F0/F3"                            self.state.last_update_time = time.time()            # 超时断开            if current_time - last_rx_time > 3.0:                return False        return True    def _open_and_stream_once(self) -> bool:        """单次尝试连接并启动数据读取"""        if hid is None:            self.log("未安装hid库，无法连接设备")            return True        device_list = hid.enumerate(VID, PID)        if not device_list:            self.log(f"未找到设备 VID=0x{VID:04X} PID=0x{PID:04X}")            return False        for dev_info in device_list:            if self.stop_event.is_set():                return True            self._safe_close()            try:                self.device = hid.device()                self.device.open_path(dev_info["path"])                self.device.set_nonblocking(False)                # 发送初始化指令                self.send_init_sequence(SEQ_A, "序列A")                time.sleep(2.2)                self.send_init_sequence(SEQ_B, "序列B")                write_report65(self.device, KEEPALIVE_PAYLOAD64)                self.log("设备连接成功")                return self._stream_loop()            except Exception as e:                self.log(f"设备初始化失败: {str(e)}")                self._safe_close()                continue        return False    def run(self) -> None:        """线程主函数：循环重连设备"""        try:            while not self.stop_event.is_set():                if self._open_and_stream_once():                    break                if self.stop_event.is_set():                    break                time.sleep(2.0)        except Exception as e:            self.log(f"线程异常退出: {repr(e)}")        finally:            self._safe_close()# ===================== 8. 雷达信号处理模块 =====================def remove_impulse_noise(    data_prev2: float,    data_prev1: float,    data_curr: float,    threshold: float) -> Tuple[float, float, float]:    """    脉冲噪声滤波    返回：滤波后值, 后向差分, 前向差分    """    backward_diff = data_prev1 - data_prev2    forward_diff = data_prev1 - data_curr    # 检测脉冲噪声并插值    if ((forward_diff > threshold) and (backward_diff > threshold)) or \       ((forward_diff < -threshold) and (backward_diff < -threshold)):        filtered_val = data_prev2 + (data_curr - data_prev2) / 2    else:        filtered_val = data_prev1    return filtered_val, backward_diff, forward_diffdef suppress_phase_noise(data: np.ndarray) -> np.ndarray:    """相位噪声抑制：中值滤波+阈值去噪+卷积平滑"""    data = np.asarray(data, dtype=np.float64).copy()    data_len = data.size    if data_len <= 2:        return data    # 零均值化    data -= np.median(data)    abs_data = np.abs(data)    med_abs = np.median(abs_data)    # 计算噪声阈值    if med_abs < np.finfo(float).eps:        threshold = max(0.06, 3.0 * np.std(data))    else:        threshold = max(0.06, 5.0 * med_abs / 0.6745)    # 异常值插值修复    outlier_indices = np.where(abs_data > threshold)[0]    for idx in outlier_indices:        if 0 < idx < data_len - 1:            data[idx] = 0.5 * (data[idx-1] + data[idx+1])        elif idx == 0 and data_len > 1:            data[idx] = data[idx+1]        elif idx == data_len - 1 and data_len > 1:            data[idx] = data[idx-1]    # 加权卷积平滑    kernel = np.array([0.2, 0.6, 0.2], dtype=np.float64)    return np.convolve(data, kernel, mode='same')