# 心电皮电情绪生物信号脑波处理 **Repository Path**: fa223797/ECG_EDA_Data ## Basic Information - **Project Name**: 心电皮电情绪生物信号脑波处理 - **Description**: 心电皮电情绪生物信号脑波处理，处理这些方面的数据可用 - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2025-09-04 - **Last Updated**: 2025-09-04 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # 生理数据时间动态分析改造方案 ## 一、项目概述 联系方式17689866867 ## 二、现有代码分析 ### 2.1 现有功能特点 - **ECG_EDA_Data.py**：实现了完整的心电和皮电信号处理流程 - ECG处理：带通滤波、R峰检测、HRV指标计算 - EDA处理：低通滤波、cvxEDA分解、SCL/SCR特征提取 - 数据库操作：SQLite数据库读写、元数据管理 - 批量处理：支持多文件批量计算和CSV汇总 ### 2.2 现有数据结构 - **文件组织**：阶段0-4 → 实验组/对照组 → 个人.acq文件 - **数据库表**：acq_meta表存储文件元数据和计算指标 - **指标体系**：mean_rr、mean_hr、lf_hf、scl_mean、scl_slope、scr_freq ## 三、改造方案设计 ### 3.1 核心改造目标 1. **分钟级时间切分**：将连续数据按1分钟间隔切分 2. **时间动态分析**：计算每分钟的MHR、LF/HF、SCL指标 3. **群体统计分析**：计算群体均值和标准差 4. **新数据库设计**：创建专门的时间动态分析数据库 5. **CSV输出**：为每个阶段生成实验组和对照组的CSV文件 ### 3.2 数据库设计 #### 3.2.1 主表结构设计 ```sql CREATE TABLE time_dynamic_analysis ( id INTEGER PRIMARY KEY AUTOINCREMENT, stage_id INTEGER NOT NULL, -- 阶段ID (0-4) group_type TEXT NOT NULL, -- 分组类型 ('实验组'/'对照组') person_name TEXT NOT NULL, -- 人名 minute_index INTEGER NOT NULL, -- 分钟索引 (从1开始) timestamp_start REAL, -- 分钟开始时间戳 timestamp_end REAL, -- 分钟结束时间戳 -- ECG指标 mhr_bpm REAL, -- 平均心率 (bpm) rr_intervals_count INTEGER, -- RR间期数量 rr_intervals_mean REAL, -- RR间期均值 (ms) rr_intervals_std REAL, -- RR间期标准差 (ms) hrv_lf REAL, -- 低频功率 (ms²) hrv_hf REAL, -- 高频功率 (ms²) hrv_lf_hf REAL, -- LF/HF比值 -- EDA指标 scl_mean_uS REAL, -- SCL均值 (μS) scl_std_uS REAL, -- SCL标准差 (μS) scl_slope_uS_per_sec REAL, -- SCL斜率 (μS/s) scr_count INTEGER, -- SCR数量 scr_frequency_per_min REAL, -- SCR频率 (次/分钟) scr_mean_amplitude_uS REAL, -- SCR平均幅值 (μS) -- 数据质量标记 data_quality_score REAL, -- 数据质量评分 (0-1) signal_quality_ecg TEXT, -- ECG信号质量 ('good'/'fair'/'poor') signal_quality_eda TEXT, -- EDA信号质量 ('good'/'fair'/'poor') -- 元数据 file_path TEXT, -- 原始文件路径 sample_rate INTEGER, -- 采样率 processing_time TEXT, -- 处理时间 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); ``` #### 3.2.2 群体统计表设计 ```sql CREATE TABLE group_statistics ( id INTEGER PRIMARY KEY AUTOINCREMENT, stage_id INTEGER NOT NULL, -- 阶段ID group_type TEXT NOT NULL, -- 分组类型 minute_index INTEGER NOT NULL, -- 分钟索引 -- 群体统计量 mhr_bpm_mean REAL, -- 平均心率均值 mhr_bpm_std REAL, -- 平均心率标准差 hrv_lf_hf_mean REAL, -- LF/HF均值 hrv_lf_hf_std REAL, -- LF/HF标准差 scl_mean_uS_mean REAL, -- SCL均值 scl_mean_uS_std REAL, -- SCL标准差 scr_frequency_per_min_mean REAL, -- SCR频率均值 scr_frequency_per_min_std REAL, -- SCR频率标准差 -- 样本信息 participant_count INTEGER, -- 参与者数量 data_quality_mean REAL, -- 平均数据质量 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); ``` #### 3.2.3 元数据表设计 ```sql CREATE TABLE processing_metadata ( id INTEGER PRIMARY KEY AUTOINCREMENT, stage_id INTEGER NOT NULL, -- 阶段ID group_type TEXT NOT NULL, -- 分组类型 total_files INTEGER, -- 总文件数 processed_files INTEGER, -- 已处理文件数 failed_files INTEGER, -- 失败文件数 total_minutes INTEGER, -- 总分钟数 processing_start_time TEXT, -- 处理开始时间 processing_end_time TEXT, -- 处理结束时间 processing_status TEXT, -- 处理状态 error_log TEXT, -- 错误日志 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); ``` ### 3.3 代码实现架构 #### 3.3.1 核心处理流程 ```python class TimeDynamicAnalyzer: def __init__(self, base_path, db_path): self.base_path = base_path self.db_path = db_path self.setup_database() def setup_database(self): """初始化数据库表结构""" pass def process_all_stages(self): """处理所有阶段数据""" for stage_id in range(5): # 阶段0-4 self.process_stage(stage_id) def process_stage(self, stage_id): """处理单个阶段数据""" stage_path = os.path.join(self.base_path, f"阶段{stage_id}") for group_type in ["实验组", "对照组"]: self.process_group(stage_id, group_type, stage_path) def process_group(self, stage_id, group_type, stage_path): """处理单个组别数据""" group_path = os.path.join(stage_path, group_type) for file_name in os.listdir(group_path): if file_name.endswith('.acq'): self.process_file(stage_id, group_type, file_name, group_path) def process_file(self, stage_id, group_type, file_name, group_path): """处理单个文件，进行分钟级分析""" file_path = os.path.join(group_path, file_name) person_name = self.extract_person_name(file_name) # 加载ACQ文件 ecg_data, eda_data, fs = self.load_acq_file(file_path) # 按分钟切分数据 minute_segments = self.split_data_by_minute(ecg_data, eda_data, fs) # 处理每分钟数据 for minute_idx, (ecg_segment, eda_segment) in enumerate(minute_segments): metrics = self.analyze_minute_segment(ecg_segment, eda_segment, fs) self.save_minute_metrics(stage_id, group_type, person_name, minute_idx, metrics) def split_data_by_minute(self, ecg_data, eda_data, fs): """按分钟切分数据""" samples_per_minute = fs * 60 total_samples = len(ecg_data) segments = [] for i in range(0, total_samples, samples_per_minute): end_idx = min(i + samples_per_minute, total_samples) ecg_segment = ecg_data[i:end_idx] eda_segment = eda_data[i:end_idx] segments.append((ecg_segment, eda_segment)) return segments def analyze_minute_segment(self, ecg_segment, eda_segment, fs): """分析单分钟数据段""" # ECG分析 ecg_metrics = self.analyze_ecg_minute(ecg_segment, fs) # EDA分析 eda_metrics = self.analyze_eda_minute(eda_segment, fs) # 数据质量评估 quality_metrics = self.assess_data_quality(ecg_segment, eda_segment) return {**ecg_metrics, **eda_metrics, **quality_metrics} def analyze_ecg_minute(self, ecg_segment, fs): """分析单分钟ECG数据""" # 信号滤波 filtered_ecg = self.filter_ecg(ecg_segment, fs) # R峰检测 r_peaks = self.detect_r_peaks(filtered_ecg, fs) # RR间期计算 rr_intervals = self.calculate_rr_intervals(r_peaks, fs) # HRV分析 hrv_metrics = self.calculate_hrv_metrics(rr_intervals) # 心率计算 mhr_bpm = self.calculate_mhr(rr_intervals) return { 'mhr_bpm': mhr_bpm, 'rr_intervals_count': len(rr_intervals), 'rr_intervals_mean': np.mean(rr_intervals) if len(rr_intervals) > 0 else None, 'rr_intervals_std': np.std(rr_intervals) if len(rr_intervals) > 0 else None, **hrv_metrics } def analyze_eda_minute(self, eda_segment, fs): """分析单分钟EDA数据""" # 信号预处理 processed_eda = self.preprocess_eda(eda_segment, fs) # cvxEDA分解 phasic, tonic, _ = self.cvxeda_decomposition(processed_eda, fs) # SCL计算 scl_metrics = self.calculate_scl_metrics(tonic, fs) # SCR计算 scr_metrics = self.calculate_scr_metrics(phasic, fs) return {**scl_metrics, **scr_metrics} def save_minute_metrics(self, stage_id, group_type, person_name, minute_idx, metrics): """保存分钟指标到数据库""" pass def generate_group_statistics(self): """生成群体统计数据""" pass def export_csv_files(self): """导出CSV文件""" pass ``` #### 3.3.2 关键算法实现 **分钟级MHR计算：** ```python def calculate_mhr(self, rr_intervals): """计算平均心率 MHR = 60 / (1/n * Σ(RR_i)) """ if len(rr_intervals) == 0: return None mean_rr_seconds = np.mean(rr_intervals) / 1000.0 # 转换为秒 mhr_bpm = 60.0 / mean_rr_seconds return mhr_bpm ``` **LF/HF比值计算：** ```python def calculate_hrv_metrics(self, rr_intervals): """计算HRV频域指标 LF/HF = ∫(0.04-0.15Hz) P(f)df / ∫(0.15-0.4Hz) P(f)df """ if len(rr_intervals) < 30: # 数据不足 return {'hrv_lf': None, 'hrv_hf': None, 'hrv_lf_hf': None} # 重采样到等间隔 rr_ms = np.array(rr_intervals) time_axis = np.cumsum(rr_ms) / 1000.0 # 转换为秒 # 插值重采样到4Hz fs_resample = 4 new_time_axis = np.arange(0, time_axis[-1], 1.0/fs_resample) rr_interpolated = np.interp(new_time_axis, time_axis, rr_ms) # 计算功率谱密度 from scipy import signal freqs, psd = signal.welch(rr_interpolated, fs=fs_resample, nperseg=min(256, len(rr_interpolated))) # 计算LF和HF功率 lf_mask = (freqs >= 0.04) & (freqs <= 0.15) hf_mask = (freqs >= 0.15) & (freqs <= 0.4) lf_power = np.trapz(psd[lf_mask], freqs[lf_mask]) hf_power = np.trapz(psd[hf_mask], freqs[hf_mask]) lf_hf_ratio = lf_power / hf_power if hf_power > 0 else None return { 'hrv_lf': lf_power, 'hrv_hf': hf_power, 'hrv_lf_hf': lf_hf_ratio } ``` **SCL计算：** ```python def calculate_scl_metrics(self, tonic_signal, fs): """计算SCL指标 SCL = 1/m * Σ(EDA_j^slow) """ if len(tonic_signal) == 0: return {'scl_mean_uS': None, 'scl_std_uS': None, 'scl_slope_uS_per_sec': None} scl_mean = np.mean(tonic_signal) scl_std = np.std(tonic_signal) # 计算斜率 duration_seconds = len(tonic_signal) / fs if duration_seconds > 0: scl_slope = (tonic_signal[-1] - tonic_signal[0]) / duration_seconds else: scl_slope = 0 return { 'scl_mean_uS': scl_mean, 'scl_std_uS': scl_std, 'scl_slope_uS_per_sec': scl_slope } ``` ### 3.4 CSV输出格式 #### 3.4.1 分钟级详细数据CSV ```csv 阶段,分组,人名,分钟索引,平均心率_bpm,LF_HF比值,SCL均值_uS,SCR频率_次每分,数据质量评分,ECG信号质量,EDA信号质量 0,实验组,张三,1,72.5,1.2,3.5,0.8,0.95,good,good 0,实验组,张三,2,75.2,1.1,3.6,1.2,0.92,good,good ... ``` #### 3.4.2 群体统计CSV ```csv 阶段,分组,分钟索引,平均心率均值,平均心率标准差,LF_HF均值,LF_HF标准差,SCL均值,SCL标准差,参与者数量 0,实验组,1,72.3,5.2,1.15,0.23,3.45,0.56,30 0,实验组,2,74.8,4.8,1.12,0.19,3.52,0.61,30 ... ``` ### 3.5 可视化参数设计 虽然代码不直接生成图表，但会提供完整的可视化参数： ```python visualization_params = { 'time_series': { 'x_axis': 'minute_index', 'y_axis_metrics': ['mhr_bpm', 'hrv_lf_hf', 'scl_mean_uS'], 'group_by': ['stage_id', 'group_type'], 'error_bars': 'std', 'line_styles': {'实验组': '-', '对照组': '--'}, 'colors': {'实验组': '#FF6B6B', '对照组': '#4ECDC4'} }, 'statistical_plots': { 'box_plots': True, 'violin_plots': True, 'correlation_matrix': True, 'confidence_intervals': [0.95, 0.99] }, 'figure_parameters': { 'figsize': (12, 8), 'dpi': 300, 'font_size': 12, 'grid': True, 'legend_position': 'upper right' } } ``` ## 四、实现步骤 ### 4.1 第一阶段：基础架构搭建 1. 创建新的数据库结构 2. 实现文件遍历和数据加载 3. 实现分钟级数据切分 ### 4.2 第二阶段：核心算法实现 1. 改进ECG分析算法，支持短时分析 2. 改进EDA分析算法，支持分钟级特征提取 3. 实现数据质量评估 ### 4.3 第三阶段：数据存储和统计 1. 实现数据库存储逻辑 2. 实现群体统计分析 3. 生成CSV输出文件 ### 4.4 第四阶段：优化和测试 1. 性能优化和错误处理 2. 数据验证和质量检查 3. 文档完善和示例测试 ## 五、预期成果 ### 5.1 数据库成果 - **time_dynamic_analysis表**：包含所有分钟级详细数据 - **group_statistics表**：包含群体统计结果 - **processing_metadata表**：包含处理过程元数据 ### 5.2 文件成果 - **阶段0-4文件夹**：每个阶段包含实验组.csv和对照组.csv - **统计汇总文件**：包含跨阶段的统计分析结果 - **可视化参数文件**：包含完整的图表生成参数 ### 5.3 功能成果 - **分钟级时间动态分析**：支持1分钟精度的生理指标分析 - **群体统计分析**：支持实验组和对照组的对比分析 - **多阶段整合**：支持跨阶段的时间序列分析 - **数据质量保证**：包含完整的数据质量评估机制 ## 六、技术特点 ### 6.1 算法优势 - **高精度时间分析**：1分钟级的时间分辨率 - **多指标融合**：ECG和EDA指标的综合分析 - **统计完整性**：包含均值、标准差等统计量 - **质量可控**：内置数据质量评估机制 ### 6.2 系统优势 - **模块化设计**：清晰的代码结构和接口设计 - **可扩展性**：支持新增指标和分析方法 - **数据完整性**：完整的元数据管理和错误追踪 - **输出标准化**：统一的CSV和数据库输出格式 ## 七、应用前景 ### 7.1 科研应用 - **情绪研究**：生理指标的时间动态变化分析 - **压力评估**：自主神经系统活动的时序特征 - **认知负荷**：生理响应的动态模式识别 ### 7.2 临床应用 - **心理健康监测**：抑郁、焦虑等疾病的生理标志物 - **治疗效果评估**：治疗干预的生理响应变化 - **个性化医疗**：基于生理模式的个性化治疗方案 ### 7.3 技术融合 - **多模态数据融合**：与面部表情、语音等数据的联合分析 - **机器学习应用**：为AI模型提供高质量的时序特征 - **实时监测**：为实时生理监测系统提供算法支持 ## 八、总结 本改造方案基于现有的心电皮电数据预处理系统，设计了一个完整的生理数据时间动态分析解决方案。通过分钟级数据切分、多指标计算、群体统计分析等技术手段，实现了从原始生理信号到时间动态特征的完整转换。该方案不仅满足了当前的研究需求，还为后续的多模态数据融合和机器学习应用奠定了坚实的基础。