🏆本文收录于专栏 《YOLOv11实战：从入门到深度优化》。
本专栏围绕 YOLOv11 的改进、训练、部署与工程优化 展开，系统梳理并复现当前主流的 YOLOv11 实战案例与优化方案，内容目前已覆盖 分类、检测、分割、追踪、关键点、OBB 检测 等多个方向。
整体坚持 持续更新 + 深度解析 + 工程导向 的写作思路，不仅关注模型结构本身，也关注训练策略、损失函数设计、推理加速、部署适配以及真实项目中的问题排查。部分章节还会结合国内外前沿论文与 AIGC 大模型技术，对主流改进方案进行重构与再设计。

🎯当前专栏限时优惠中：一次订阅，终身有效，后续更新内容均可免费解锁 👉 点此查看专栏详情 👈️
 
🎉本专栏还不够过瘾？别急，好戏才刚刚开始！我已经为你准备了一整套 YOLO 进阶实战大礼包🎁：

👉《YOLOv8实战》
👉《YOLOv9实战》
👉《YOLOv10实战》
👉《YOLOv11实战》
👉《YOLOv12实战》
👉以及最新上线的 《YOLOv26实战》

想一次搞定所有版本？直接冲 《YOLO全栈实战合集》，一站式涵盖 YOLO 各版本实战教学！

🚀想学哪个版本？直接找 bug 菌“许愿”，安排！必须安排！🚀

🎯 本文定位：目标检测 × YOLOv11 脑机接口与生物计算篇
📅 预计阅读时间：约50～60 分钟
⭐ 难度等级：⭐⭐⭐⭐☆（高级）
🔧 技术栈：Ultralytics YOLO11 | Python v3.9+ | PyTorch v2.0+ | torchvision v0.9+ | Ultralytics v8.x | CUDA v11.8+

📖 上期回顾

在上期《YOLOv11【第十四章：脑机接口与生物计算篇·第11节】BCI 数据集制作：OpenBCI + YOLOv11 标注规范》内容中，我们系统地讲解了如何利用 OpenBCI 硬件平台 构建一套完整的脑机接口数据集制作流程，并将其与 YOLOv11 的标注规范深度融合。核心知识点回顾如下：

上期核心知识点梳理

1. OpenBCI 硬件体系
我们介绍了 OpenBCI Cyton（8通道）与 Ganglion（4通道）两款主流开源脑电采集板的硬件参数、采样率配置（250Hz/500Hz）以及电极帽布局（基于国际10-20系统）。重点讲解了如何通过 brainflow Python 库实现稳定的数据流采集，并处理常见的信号伪迹（眼电、肌电干扰）。

2. 同步时间戳对齐机制
脑电信号与视觉刺激之间的毫秒级时间对齐是 BCI 数据集质量的核心。我们设计了基于 LSL（Lab Streaming Layer）协议的事件标记系统，确保每一帧视觉刺激都能与对应时间窗口的 EEG 信号精确绑定，时间偏差控制在 ±2ms 以内。

3. YOLOv11 标注规范扩展
传统 YOLO 标注仅包含 [class_id, x_center, y_center, width, height]，我们在此基础上扩展了 BCI 元数据字段，包括：受试者注意力状态标签（focused/distracted/fatigued）、P300 事件触发标记、认知负荷等级（1-5级）等，形成了一套面向脑机融合的多模态标注规范。

4. 数据质量控制流水线
构建了包含 阻抗检测 → 基线校正 → 频带滤波 → 伪迹去除 → 特征提取 的完整质控流水线，确保入库数据的信噪比（SNR）达到临床研究标准（>10 dB）。

5. 数据集发布与版本管理
介绍了使用 DVC（Data Version Control）对 BCI 数据集进行版本管理的最佳实践，以及如何将数据集发布到 HuggingFace Hub 供社区共享。

上期内容为我们夯实了 BCI 数据集的工程基础。本节我们将在此基础上，聚焦一个极具现实意义的临床应用场景——中风患者的视觉注意力康复检测，深入探讨如何将 YOLOv11 的视觉感知能力与脑电信号分析相结合，为中风康复提供智能化的客观评估工具。

一、引言：中风康复的视觉注意力挑战

1.1 中风后视觉注意力障碍的临床现状

中风（Stroke）是全球范围内导致成人残疾的首要原因。据世界卫生组织统计，全球每年新增中风患者超过 1500 万例，其中约三分之一会遗留不同程度的永久性残疾。在中风后遗症中，视觉注意力障碍（Visual Attention Disorders）是最常见且最影响生活质量的认知障碍之一，主要表现为：

单侧空间忽略（Hemispatial Neglect / Spatial Neglect）
这是中风后最典型的视觉注意力障碍，患者无法感知或响应对侧空间的刺激。右半球损伤的患者（占单侧忽略患者的约 75-85%）会忽略左侧视野中的物体、人物甚至自己身体的左侧部分。患者在进食时只吃右侧餐盘的食物，阅读时只读右半页的文字，严重影响日常生活自理能力。

视觉扫描障碍（Visual Scanning Deficits）
正常人在浏览复杂场景时，眼球运动遵循系统化的扫描模式（通常从左到右、从上到下）。中风患者的扫描路径往往出现随机化、碎片化的特征，无法有效覆盖整个视野，导致重要目标的检测遗漏率大幅上升。

持续性注意障碍（Sustained Attention Deficits）
中风后的神经损伤会影响大脑的警觉网络和执行控制网络，导致患者维持持续注意力的能力下降。在进行持续 5 分钟以上的视觉任务时，患者的检测性能会显著衰退，而这种衰退模式与健康对照组有着显著差异。

注意力资源竞争失衡
当视野中同时出现多个目标时，中风患者的注意力资源分配能力严重受损，往往只能关注单一突出目标，对其他目标形成"竞争性抑制"，这在临床上称为消退现象（Extinction Phenomenon）。

1.2 传统评估方法的局限性

目前临床上评估中风患者视觉注意力的传统方法包括：

评估方法	优点	局限性
线段等分测试（Line Bisection Test）	操作简单，成本低	主观性强，无法量化空间分布
字母删除测试（Letter Cancellation Test）	标准化程度高	静态评估，缺乏动态场景
眼动追踪（Eye Tracking）	客观量化眼球运动	设备昂贵，患者配合度要求高
神经心理评估量表（NIHSS/MoCA）	综合评估认知状态	耗时长，对评估者专业要求高
功能性磁共振成像（fMRI）	揭示神经机制	无法实时，成本极高

传统方法的核心问题在于：无法在真实生活场景中进行实时、连续、客观的评估。患者在医院的评估结果往往无法真实反映其在日常生活环境中的功能状态（即"医院-家庭落差"）。

1.3 YOLOv11 + BCI 的解决方案愿景

本节提出的解决方案，是将 YOLOv11 实时目标检测 与 脑电信号分析 有机结合，构建一个能够在自然视觉场景下对中风患者视觉注意力进行实时评估和辅助康复训练的闭环系统。

系统核心价值：

1. 客观性：不依赖患者主观报告，通过神经信号（EEG）和行为信号（视线方向、反应时间）客观量化注意力状态
2. 生态效度：在真实或模拟真实的视觉场景中进行评估，而非孤立的人工测试
3. 实时性：毫秒级的脑电特征提取 + 实时目标检测，支持即时反馈
4. 个性化：基于个体神经数据自适应调整训练难度和刺激策略
5. 可及性：相比 fMRI，EEG 设备成本更低，可在康复病房甚至家庭中部署

二、系统整体架构设计

2.1 系统架构总览

2.2 核心技术栈选型

相关示意图绘制如下，仅供参考：

2.3 数据流时序设计

相关示意图绘制如下，仅供参考：

三、中风视觉注意力的神经机制与检测原理

3.1 大脑注意力网络解析

理解中风患者视觉注意力检测的原理，必须首先了解正常大脑的注意力神经网络架构。根据现代认知神经科学的研究成果，视觉注意力主要由以下三个功能网络协作完成：

相关示意图绘制如下，仅供参考：

关键神经机制说明：

背侧注意网络（Dorsal Attention Network, DAN）：负责自上而下（goal-directed）的注意力控制，主要包括前额叶眼动区（FEF）和顶内沟（IPS）。当我们主动搜索特定目标时，这个网络被激活，引导眼球和注意力朝向目标位置。

腹侧注意网络（Ventral Attention Network, VAN）：负责自下而上（stimulus-driven）的注意力捕获，主要包括颞顶联合区（TPJ）和腹侧额叶皮层（VFC）。当突然出现意外或显著刺激时，这个网络响应，并将注意力重定向到新刺激。单侧忽略综合征主要源于右侧 VAN 的损伤，因为右侧腹侧注意网络负责处理双侧空间的刺激，损伤后左侧空间的突显事件将无法被有效捕获。

3.2 EEG 生物标记物与注意力状态映射

不同的注意力状态在 EEG 频谱上有不同的特征表现，这是我们进行客观注意力检测的神经电生理基础：

EEG 生物标记物	频率范围	与注意力的关系	康复价值
Alpha 波	8-12 Hz	幂值升高 → 注意力降低（大脑"空转"）	监控忽略区域的皮层激活状态
Beta 波	13-30 Hz	幂值升高 → 主动认知处理增强	评估主动搜索努力程度
Theta 波	4-7 Hz	幂值升高 → 认知负荷增大/疲劳	监控训练疲劳度
P300 成分	事件相关电位	出现 → 目标刺激被有效感知	判断特定位置目标是否被注意
N2pc 成分	事件相关电位	出现 → 注意力聚焦于目标位置	定位注意力偏向方向
Alpha 不对称性	8-12 Hz 左右差异	右侧升高 → 左侧空间被忽略	单侧忽略程度量化指标

3.3 P300 信号的目标感知检测原理

P300 是我们系统中最核心的 EEG 生物标记物。它是一种事件相关电位（ERP），在目标刺激出现后约 250-500ms 在头皮顶部（Pz、Cz电极）出现正向偏转峰值，幅值约 5-20 μV。

物理意义：P300 反映了大脑对"意外"或"目标"刺激的识别和处理过程。在奇异范式（Oddball Paradigm）中，当受试者注意到目标刺激时，P300 会出现；若目标刺激落在忽略区域，P300 幅值将显著降低甚至消失。

在中风评估中的应用：

• 若刺激出现在患者健侧空间 → P300 幅值正常（10-20μV）
• 若刺激出现在患者忽略侧空间 → P300 幅值显著降低（<5μV）或缺失
• 这种位置依赖性 P300 幅值差异可以客观量化忽略程度

四、EEG 注意力信号处理模块

4.1 环境配置与依赖安装

# 创建专用虚拟环境
conda create -n stroke_rehab python=3.10 -y
conda activate stroke_rehab

# 安装核心依赖
pip install ultralytics==8.3.0          # YOLOv11
pip install mne==1.7.0                  # EEG 处理
pip install brainflow==5.12.1           # OpenBCI 数据采集
pip install pylsl==1.16.2               # Lab Streaming Layer
pip install scipy==1.13.0               # 科学计算
pip install scikit-learn==1.5.0         # 机器学习
pip install torch==2.3.0 torchvision    # 深度学习框架
pip install pyqt6==6.7.0                # GUI 框架
pip install matplotlib==3.9.0           # 可视化
pip install pandas==2.2.0               # 数据处理
pip install pyxdf==1.16.5               # XDF 格式读取
pip install tobii-research==1.10.1      # Tobii 眼动追踪（可选）

4.2 EEG 预处理核心模块

# eeg_preprocessor.py
# 中风患者 EEG 信号预处理模块
# 作者：BCI-YOLOv11 系列
# 功能：原始 EEG 信号 → 可用于注意力分类的特征向量

import numpy as np
import mne
from mne.preprocessing import ICA
from scipy import signal
from scipy.stats import zscore
import warnings
warnings.filterwarnings('ignore')


class StrokeEEGPreprocessor:
    """
    中风患者 EEG 预处理器
    
    设计原则：
    1. 保守的伪迹去除策略（避免过度处理导致P300失真）
    2. 支持在线（实时）和离线两种处理模式
    3. 针对中风患者特殊性：低振幅信号、高伪迹率、配合度差
    """
    
    def __init__(self, sfreq=250, n_channels=8, online_mode=False):
        """
        初始化预处理器
        
        参数：
            sfreq: 采样频率（Hz），OpenBCI Cyton 默认 250Hz
            n_channels: EEG 通道数
            online_mode: 是否为实时处理模式
        """
        self.sfreq = sfreq          # 采样频率
        self.n_channels = n_channels  # 通道数
        self.online_mode = online_mode  # 在线/离线模式标志
        
        # 国际10-20系统标准通道名称（8通道配置）
        self.channel_names = ['Fp1', 'Fp2', 'C3', 'C4', 'P3', 'P4', 'O1', 'O2']
        
        # 注意力相关频段定义（单位：Hz）
        self.freq_bands = {
            'delta': (1, 4),    # Delta波：睡眠/深度疲劳
            'theta': (4, 8),    # Theta波：认知负荷/工作记忆
            'alpha': (8, 13),   # Alpha波：安静放松/注意抑制
            'beta': (13, 30),   # Beta波：主动思维/注意激活
            'gamma': (30, 45)   # Gamma波：高级认知整合
        }
        
        # 在线处理的滑动窗口缓冲区（2秒，用于P300检测）
        self.window_size = int(2.0 * sfreq)  # 500 个采样点
        self.buffer = np.zeros((n_channels, self.window_size * 3))  # 6秒缓冲
        self.buffer_ptr = 0  # 缓冲区指针
        
        # 用于在线基线校正的历史统计量
        self.baseline_mean = None
        self.baseline_std = None
        
        print(f"[EEGPreprocessor] 初始化完成 | 采样率:{sfreq}Hz | "
              f"通道数:{n_channels} | 模式:{'在线' if online_mode else '离线'}")
    
    def bandpass_filter(self, data, low_freq, high_freq, order=4):
        """
        巴特沃斯带通滤波器
        
        参数：
            data: 输入数据，shape=(n_channels, n_times)
            low_freq: 低截止频率（Hz）
            high_freq: 高截止频率（Hz）
            order: 滤波器阶数（建议4阶，避免相位失真过大）
        
        返回：
            filtered_data: 滤波后数据，shape=(n_channels, n_times)
        
        注意：使用零相位滤波（filtfilt）避免ERP成分相位偏移
        """
        # 计算奈奎斯特频率
        nyquist = self.sfreq / 2.0
        
        # 归一化截止频率（避免超出奈奎斯特频率）
        low = max(low_freq / nyquist, 0.001)
        high = min(high_freq / nyquist, 0.999)
        
        # 设计巴特沃斯滤波器
        b, a = signal.butter(order, [low, high], btype='band')
        
        # 零相位双向滤波（消除相位失真，对ERP分析至关重要）
        filtered_data = signal.filtfilt(b, a, data, axis=1)
        
        return filtered_data
    
    def notch_filter(self, data, notch_freq=50.0, q_factor=30):
        """
        工频陷波滤波器（去除电源线噪声）
        
        参数：
            data: 输入数据
            notch_freq: 工频频率（中国/欧洲为50Hz，北美为60Hz）
            q_factor: 品质因数（越高陷波越窄，默认30）
        
        返回：
            filtered_data: 陷波后数据
        """
        # 设计陷波滤波器（IIR带阻）
        b, a = signal.iirnotch(notch_freq, q_factor, self.sfreq)
        filtered_data = signal.filtfilt(b, a, data, axis=1)
        
        # 同时去除工频谐波（100Hz, 150Hz...）
        harmonic_freq = notch_freq * 2
        if harmonic_freq < self.sfreq / 2:
            b2, a2 = signal.iirnotch(harmonic_freq, q_factor, self.sfreq)
            filtered_data = signal.filtfilt(b2, a2, filtered_data, axis=1)
        
        return filtered_data
    
    def remove_artifacts_threshold(self, data, threshold_uv=100):
        """
        基于振幅阈值的简单伪迹检测与标记
        
        中风患者常见伪迹来源：
        - 眼电（EOG）：振幅可达 ±500μV，主要影响前额叶电极（Fp1, Fp2）
        - 肌电（EMG）：高频宽带噪声，振幅 10-100μV
        - 电极脱落：突发大幅跳变
        
        参数：
            data: 输入数据，shape=(n_channels, n_times)，单位 μV
            threshold_uv: 幅值阈值（μV），超过此值标记为伪迹
        
        返回：
            clean_data: 清洁数据（伪迹段置零）
            artifact_mask: 伪迹掩码，shape=(n_times,)，True表示伪迹
        """
        n_channels, n_times = data.shape
        artifact_mask = np.zeros(n_times, dtype=bool)
        
        # 遍历每个通道检测超阈值段
        for ch in range(n_channels):
            # 检测超阈值点
            exceed_threshold = np.abs(data[ch]) > threshold_uv
            
            # 扩展伪迹窗口（前后各延伸25ms，避免伪迹边缘污染）
            extension = int(0.025 * self.sfreq)  # 25ms
            extended = np.convolve(exceed_threshold.astype(float), 
                                   np.ones(2 * extension + 1), 
                                   mode='same') > 0
            artifact_mask |= extended
        
        # 计算伪迹率（用于质量报告）
        artifact_rate = artifact_mask.sum() / n_times
        
        if artifact_rate > 0.5:
            # 超过50%被标记为伪迹，质量过差，发出警告
            print(f"[警告] 伪迹率过高: {artifact_rate:.1%}，请检查电极接触状态！")
        
        # 将伪迹段置为NaN（而非0，便于后续处理时识别）
        clean_data = data.copy()
        clean_data[:, artifact_mask] = np.nan
        
        # 线性插值填补短暂伪迹段（<0.5秒）
        for ch in range(n_channels):
            nan_mask = np.isnan(clean_data[ch])
            if nan_mask.any():
                # 找到非NaN的索引
                good_indices = np.where(~nan_mask)[0]
                if len(good_indices) > 2:
                    # 使用线性插值填补
                    clean_data[ch, nan_mask] = np.interp(
                        np.where(nan_mask)[0],
                        good_indices,
                        clean_data[ch, good_indices]
                    )
        
        return clean_data, artifact_mask, artifact_rate
    
    def compute_erp(self, raw_data, event_times, pre_stim=0.1, post_stim=0.8):
        """
        计算事件相关电位（ERP）
        
        ERP 是通过叠加平均（Averaging）方法，从噪声背景中提取与刺激
        锁时相关的神经电位。对于P300检测，通常需要20次以上叠加才能
        获得足够的信噪比。
        
        参数：
            raw_data: 原始EEG，shape=(n_channels, n_times)
            event_times: 刺激事件时间点列表（样本索引）
            pre_stim: 刺激前基线窗口（秒），默认100ms
            post_stim: 刺激后分析窗口（秒），默认800ms
        
        返回：
            erp_mean: 平均ERP波形，shape=(n_channels, n_erp_times)
            erp_epochs: 单试次ERP，shape=(n_epochs, n_channels, n_erp_times)
            times: 时间轴（秒）
        """
        pre_samples = int(pre_stim * self.sfreq)     # 基线采样点数
        post_samples = int(post_stim * self.sfreq)    # 分析窗采样点数
        epoch_length = pre_samples + post_samples     # 单个epoch总长度
        
        # 时间轴（以刺激时刻为0点）
        times = np.linspace(-pre_stim, post_stim, epoch_length)
        
        epochs = []
        valid_events = 0
        
        for event_idx in event_times:
            # 检查epoch边界是否有效
            start = event_idx - pre_samples
            end = event_idx + post_samples
            
            if start < 0 or end > raw_data.shape[1]:
                continue  # 跳过边界不完整的epoch
            
            epoch = raw_data[:, start:end]
            
            # 基线校正：减去刺激前100ms的平均值
            baseline_mean = np.mean(epoch[:, :pre_samples], axis=1, keepdims=True)
            epoch = epoch - baseline_mean
            
            # 排除包含NaN的epoch（伪迹段）
            if not np.any(np.isnan(epoch)):
                epochs.append(epoch)
                valid_events += 1
        
        if valid_events < 5:
            print(f"[警告] 有效epoch数量不足（{valid_events}），ERP可靠性低！")
        
        erp_epochs = np.array(epochs)  # shape: (n_epochs, n_channels, n_times)
        erp_mean = np.mean(erp_epochs, axis=0)  # 叠加平均
        
        print(f"[ERP计算] 有效epoch: {valid_events}/{len(event_times)} | "
              f"信噪比改善倍数: √{valid_events:.0f} ≈ {np.sqrt(valid_events):.1f}x")
        
        return erp_mean, erp_epochs, times
    
    def compute_band_power(self, data, band_name):
        """
        计算特定频段的相对功率谱密度
        
        使用 Welch 方法估计 PSD（比 FFT 更稳定，对短时数据更友好）
        
        参数：
            data: EEG数据，shape=(n_channels, n_times)
            band_name: 频段名称（'alpha', 'beta', 'theta', etc.）
        
        返回：
            band_power: 各通道在目标频段的相对功率，shape=(n_channels,)
        """
        if band_name not in self.freq_bands:
            raise ValueError(f"未知频段: {band_name}，可选: {list(self.freq_bands.keys())}")
        
        low_freq, high_freq = self.freq_bands[band_name]
        n_channels, n_times = data.shape
        band_power = np.zeros(n_channels)
        
        for ch in range(n_channels):
            # Welch 方法估计功率谱密度
            # nperseg=256 对应约1秒窗口（250Hz采样率）
            freqs, psd = signal.welch(
                data[ch], 
                fs=self.sfreq, 
                nperseg=min(256, n_times),
                noverlap=128,
                window='hann'
            )
            
            # 提取目标频段的功率
            freq_idx = np.where((freqs >= low_freq) & (freqs <= high_freq))[0]
            target_power = np.trapz(psd[freq_idx], freqs[freq_idx])
            
            # 计算全频段总功率（用于归一化）
            total_power = np.trapz(psd, freqs) + 1e-10  # 防止除零
            
            # 相对功率（%）
            band_power[ch] = (target_power / total_power) * 100
        
        return band_power
    
    def compute_alpha_asymmetry(self, data):
        """
        计算 Alpha 波半球不对称性指数
        
        Alpha 不对称性（Frontal Alpha Asymmetry）是单侧忽略的重要神经标记。
        忽略症患者通常表现为：损伤侧（右）Alpha 功率降低，健侧（左）Alpha 功率升高，
        导致 Alpha 不对称性指数异常。
        
        公式：
            Asymmetry Index = ln(Right_Alpha) - ln(Left_Alpha)
            正值 → 右侧激活更强（左侧忽略风险较低）
            负值 → 左侧激活更强（左侧忽略风险较高）
        
        参数：
            data: EEG数据，shape=(n_channels, n_times)
        
        返回：
            asymmetry_indices: 各电极对的不对称性指数字典
        """
        # 定义对称电极对（左-右）
        electrode_pairs = {
            'frontal': ('Fp1', 'Fp2'),     # 前额叶
            'central': ('C3', 'C4'),        # 中央区
            'parietal': ('P3', 'P4')        # 顶叶（对单侧忽略最敏感）
        }
        
        # 先提取所有通道的alpha功率
        alpha_power = self.compute_band_power(data, 'alpha')
        
        asymmetry_indices = {}
        
        for region, (left_ch, right_ch) in electrode_pairs.items():
            # 找到对应通道索引
            if left_ch in self.channel_names and right_ch in self.channel_names:
                left_idx = self.channel_names.index(left_ch)
                right_idx = self.channel_names.index(right_ch)
                
                left_alpha = alpha_power[left_idx] + 1e-10
                right_alpha = alpha_power[right_idx] + 1e-10
                
                # 对数不对称性指数（常用的计算方式）
                asymmetry = np.log(right_alpha) - np.log(left_alpha)
                asymmetry_indices[region] = asymmetry
        
        # 综合不对称性指数（加权平均，顶叶权重最高）
        weights = {'frontal': 0.2, 'central': 0.3, 'parietal': 0.5}
        weighted_asymmetry = sum(
            asymmetry_indices.get(region, 0) * weight 
            for region, weight in weights.items()
        )
        asymmetry_indices['composite'] = weighted_asymmetry
        
        return asymmetry_indices
    
    def extract_attention_features(self, data):
        """
        提取综合注意力特征向量
        
        整合多个频段功率、不对称性指数，构建用于注意力状态分类的特征向量
        
        参数：
            data: EEG数据，shape=(n_channels, n_times)
        
        返回：
            features: 特征向量，shape=(n_features,)
            feature_names: 特征名称列表
        """
        features = []
        feature_names = []
        
        # 1. 各频段相对功率（8通道 × 5频段 = 40维）
        for band_name in ['delta', 'theta', 'alpha', 'beta', 'gamma']:
            band_power = self.compute_band_power(data, band_name)
            features.extend(band_power)
            feature_names.extend([f'{ch}_{band_name}' for ch in self.channel_names])
        
        # 2. Alpha/Beta 功率比（认知状态关键指标，8维）
        alpha_power = self.compute_band_power(data, 'alpha')
        beta_power = self.compute_band_power(data, 'beta')
        alpha_beta_ratio = alpha_power / (beta_power + 1e-10)
        features.extend(alpha_beta_ratio)
        feature_names.extend([f'{ch}_alpha_beta_ratio' for ch in self.channel_names])
        
        # 3. Theta/Alpha 功率比（认知负荷指标，8维）
        theta_power = self.compute_band_power(data, 'theta')
        theta_alpha_ratio = theta_power / (alpha_power + 1e-10)
        features.extend(theta_alpha_ratio)
        feature_names.extend([f'{ch}_theta_alpha_ratio' for ch in self.channel_names])
        
        # 4. 半球不对称性指数（3维）
        asymmetry = self.compute_alpha_asymmetry(data)
        for region in ['frontal', 'central', 'parietal']:
            features.append(asymmetry.get(region, 0.0))
            feature_names.append(f'{region}_alpha_asymmetry')
        
        # 5. 综合不对称性指数（1维）
        features.append(asymmetry.get('composite', 0.0))
        feature_names.append('composite_alpha_asymmetry')
        
        # Z-score 归一化（使特征具有可比性）
        features = np.array(features)
        features = np.nan_to_num(features, nan=0.0, posinf=0.0, neginf=0.0)
        
        return features, feature_names


# ======================== 代码测试入口 ========================
if __name__ == '__main__':
    import matplotlib
    matplotlib.use('Agg')  # 非交互式后端
    import matplotlib.pyplot as plt
    
    print("=" * 60)
    print("EEG 预处理模块测试")
    print("=" * 60)
    
    # 模拟参数
    sfreq = 250         # 采样频率 250Hz
    n_channels = 8      # 通道数
    duration = 10       # 模拟10秒数据
    n_times = sfreq * duration  # 总采样点数
    
    # 生成模拟 EEG 信号（真实信号约为 1-50 μV 范围）
    np.random.seed(42)
    t = np.linspace(0, duration, n_times)
    
    # 模拟各频段成分叠加
    raw_data = np.zeros((n_channels, n_times))
    for ch in range(n_channels):
        # Alpha 成分（8-12Hz，主要在顶枕区）
        alpha_amp = 15 if ch >= 4 else 8  # 顶枕区alpha更强
        raw_data[ch] += alpha_amp * np.sin(2 * np.pi * 10 * t + np.random.rand() * 2 * np.pi)
        
        # Beta 成分（13-30Hz）
        raw_data[ch] += 5 * np.sin(2 * np.pi * 20 * t + np.random.rand() * 2 * np.pi)
        
        # 工频干扰（50Hz）
        raw_data[ch] += 20 * np.sin(2 * np.pi * 50 * t)  # 模拟电源线噪声
        
        # 白噪声背景
        raw_data[ch] += np.random.randn(n_times) * 3
    
    # 模拟眼电伪迹（大振幅，主要影响前两个通道）
    blink_times = [250, 750, 1500, 2100]  # 眨眼时刻（样本索引）
    for bt in blink_times:
        if bt + 30 < n_times:
            blink_shape = np.exp(-np.arange(30) / 8)  # 指数衰减形状
            raw_data[0, bt:bt+30] += 150 * blink_shape  # Fp1 眼电
            raw_data[1, bt:bt+30] += 120 * blink_shape  # Fp2 眼电
    
    print(f"\n原始数据统计:")
    print(f"  形状: {raw_data.shape}")
    print(f"  均值: {raw_data.mean():.2f} μV")
    print(f"  标准差: {raw_data.std():.2f} μV")
    print(f"  最大值: {raw_data.max():.2f} μV")
    
    # 初始化预处理器
    preprocessor = StrokeEEGPreprocessor(sfreq=sfreq, n_channels=n_channels)
    
    # 步骤1：去除工频噪声
    print("\n[步骤1] 工频陷波滤波（50Hz）...")
    data_notch = preprocessor.notch_filter(raw_data, notch_freq=50.0)
    
    # 步骤2：带通滤波（1-40Hz，保留注意力相关频段）
    print("[步骤2] 带通滤波（1-40Hz）...")
    data_filtered = preprocessor.bandpass_filter(data_notch, 1.0, 40.0)
    
    # 步骤3：伪迹去除
    print("[步骤3] 振幅阈值伪迹检测...")
    data_clean, artifact_mask, artifact_rate = preprocessor.remove_artifacts_threshold(
        data_filtered, threshold_uv=80
    )
    print(f"  伪迹率: {artifact_rate:.1%}")
    
    # 步骤4：提取注意力特征
    print("[步骤4] 提取注意力特征向量...")
    features, feature_names = preprocessor.extract_attention_features(data_clean)
    print(f"  特征维度: {len(features)}")
    
    # 步骤5：计算Alpha不对称性
    print("[步骤5] 计算Alpha波不对称性指数...")
    asymmetry = preprocessor.compute_alpha_asymmetry(data_clean)
    print(f"  顶叶Alpha不对称性: {asymmetry.get('parietal', 0):.4f}")
    print(f"  综合不对称性指数: {asymmetry.get('composite', 0):.4f}")
    
    # 步骤6：模拟P300 ERP计算
    print("[步骤6] 计算事件相关电位（P300）...")
    # 模拟刺激事件时间点（每1秒一个刺激）
    event_times = [int(i * sfreq) for i in range(1, duration)]
    erp_mean, erp_epochs, times = preprocessor.compute_erp(
        data_clean, event_times, pre_stim=0.1, post_stim=0.6
    )
    print(f"  ERP形状: {erp_mean.shape}")
    
    print("\n✅ EEG 预处理模块测试通过！")
    print(f"  处理流程: 原始EEG → 陷波 → 带通 → 去伪迹 → 特征提取")
    print(f"  输出特征维度: {len(features)}")

代码解析：

上述 StrokeEEGPreprocessor 类实现了一个面向中风患者的专用 EEG 预处理管道，其核心设计理念是保守主义原则——在去除噪声的同时，最大程度保护与注意力相关的神经信号成分（尤其是 P300 等 ERP 成分）。

关键方法说明：

• bandpass_filter：使用 filtfilt（双向滤波）而非 lfilter（单向滤波），这是因为双向滤波具有零相位失真特性，对于 ERP 分析中时间精度要求极高的 P300 成分（峰值在刺激后约 300ms）至关重要
• remove_artifacts_threshold：采用"扩展窗口"策略，将阈值超越点前后各扩展 25ms，这是因为眼电伪迹通常是缓慢的大幅波动，简单的点对点阈值会遗漏边缘污染
• compute_alpha_asymmetry：这是整个系统中最具临床价值的特征之一，顶叶（P3-P4）的 Alpha 不对称性与单侧忽略严重程度相关性最高，权重设置为 0.5 反映了这一点

五、YOLOv11 视觉场景分析模块

5.1 面向康复场景的 YOLOv11 定制化

在中风康复应用中，YOLOv11 需要完成三项核心任务：

1. 场景物体检测：识别视野中的所有物体，建立"应被注意的目标清单"
2. 注意力热区生成：基于眼动数据和检测结果，生成患者实际注意分布热图
3. 忽略区域识别：比对"应注意目标"与"实际注意目标"，定位忽略区域

# yolov11_stroke_detector.py
# YOLOv11 中风康复场景检测模块
# 功能：场景目标检测 + 注意力热图生成 + 忽略区域识别

import cv2
import numpy as np
import torch
from ultralytics import YOLO
from collections import defaultdict
from typing import Dict, List, Tuple, Optional
import time


class StrokeRehabDetector:
    """
    面向中风康复的 YOLOv11 检测器
    
    核心功能：
    1. 实时目标检测（场景物体识别）
    2. 注视-目标关联分析（判断患者是否注意到特定目标）
    3. 空间忽略热图生成（可视化忽略区域分布）
    4. 时序注意力追踪（监测注意力随时间的变化）
    """
    
    def __init__(self, model_path='yolo11n.pt', device='auto', 
                 conf_threshold=0.5, iou_threshold=0.45):
        """
        初始化检测器
        
        参数：
            model_path: YOLOv11 模型路径
            device: 推理设备（'auto'/'cpu'/'cuda:0'）
            conf_threshold: 检测置信度阈值
            iou_threshold: NMS IoU 阈值
        """
        # 自动选择推理设备
        if device == 'auto':
            self.device = 'cuda:0' if torch.cuda.is_available() else 'cpu'
        else:
            self.device = device
        
        print(f"[StrokeDetector] 加载模型: {model_path} | 设备: {self.device}")
        
        # 加载 YOLOv11 模型
        self.model = YOLO(model_path)
        self.conf_threshold = conf_threshold
        self.iou_threshold = iou_threshold
        
        # 图像尺寸（YOLOv11 标准输入）
        self.img_size = 640
        
        # 注意力热图参数
        self.heatmap_decay = 0.95       # 热图时间衰减系数（模拟注意力遗忘）
        self.heatmap_sigma = 50         # 高斯扩散半径（像素）
        
        # 累积注意力热图（初始化为全零）
        self.attention_heatmap = None   # 将在首次处理时初始化
        self.neglect_heatmap = None     # 忽略区域热图
        
        # 检测历史记录（用于时序分析）
        self.detection_history = []     # 每帧检测结果历史
        self.attention_history = []     # 每帧注意力评分历史
        self.timestamp_history = []     # 时间戳历史
        
        # 忽略评分阈值（低于此值认为该区域被忽略）
        self.neglect_score_threshold = 0.3
        
        # 定义关键物体类别（日常康复场景中的常见物体）
        self.rehabilitation_objects = {
            # 餐饮类（评估进食时的注意力）
            'cup': '杯子', 'bottle': '瓶子', 'fork': '叉子', 
            'knife': '刀', 'spoon': '勺子', 'bowl': '碗',
            # 家居类（评估环境导航能力）
            'chair': '椅子', 'couch': '沙发', 'tv': '电视', 
            'laptop': '笔记本', 'book': '书',
            # 人物类（评估社交注意力）
            'person': '人',
            # 电子设备类
            'cell phone': '手机', 'remote': '遥控器',
            # 其他常见物品
            'clock': '时钟', 'vase': '花瓶'
        }
        
        print(f"[StrokeDetector] 初始化完成 | 关注类别数: {len(self.rehabilitation_objects)}")
    
    def detect_scene(self, frame: np.ndarray) -> Dict:
        """
        对单帧图像进行场景检测
        
        参数：
            frame: BGR格式图像，shape=(H, W, 3)
        
        返回：
            result: 检测结果字典，包含：
                - detections: 所有检测框列表
                - left_objects: 左侧区域物体
                - right_objects: 右侧区域物体
                - spatial_distribution: 空间分布统计
        """
        frame_h, frame_w = frame.shape[:2]
        
        # 执行 YOLOv11 推理
        results = self.model(
            frame, 
            conf=self.conf_threshold,
            iou=self.iou_threshold,
            verbose=False
        )
        
        detections = []
        left_objects = []    # 图像左半部分的物体（对应单侧忽略风险区域）
        right_objects = []   # 图像右半部分的物体
        center_x_threshold = frame_w / 2  # 左右分界线
        
        # 解析检测结果
        for result in results:
            boxes = result.boxes
            if boxes is None or len(boxes) == 0:
                continue
            
            for box in boxes:
                # 提取检测框坐标（xyxy格式）
                x1, y1, x2, y2 = box.xyxy[0].cpu().numpy().astype(int)
                conf = float(box.conf[0].cpu().numpy())
                cls_id = int(box.cls[0].cpu().numpy())
                cls_name = self.model.names[cls_id]
                
                # 计算目标中心点
                center_x = (x1 + x2) / 2
                center_y = (y1 + y2) / 2
                
                # 计算目标面积（用于显著性权重）
                area = (x2 - x1) * (y2 - y1)
                area_ratio = area / (frame_h * frame_w)  # 相对面积
                
                detection = {
                    'bbox': [x1, y1, x2, y2],           # 检测框坐标
                    'center': [center_x, center_y],       # 中心点
                    'conf': conf,                          # 置信度
                    'class_id': cls_id,                   # 类别ID
                    'class_name': cls_name,               # 类别名称
                    'class_cn': self.rehabilitation_objects.get(cls_name, cls_name),  # 中文名
                    'area_ratio': area_ratio,             # 相对面积（显著性）
                    'side': 'left' if center_x < center_x_threshold else 'right',  # 所在侧
                    'spatial_quadrant': self._get_quadrant(center_x, center_y, frame_w, frame_h)
                }
                
                detections.append(detection)
                
                # 按空间位置分类
                if center_x < center_x_threshold:
                    left_objects.append(detection)
                else:
                    right_objects.append(detection)
        
        # 空间分布统计
        spatial_distribution = {
            'total': len(detections),
            'left_count': len(left_objects),
            'right_count': len(right_objects),
            'left_right_ratio': len(left_objects) / (len(right_objects) + 1e-6),
            'left_objects': [d['class_cn'] for d in left_objects],
            'right_objects': [d['class_cn'] for d in right_objects]
        }
        
        return {
            'detections': detections,
            'left_objects': left_objects,
            'right_objects': right_objects,
            'spatial_distribution': spatial_distribution,
            'frame_size': (frame_w, frame_h)
        }
    
    def _get_quadrant(self, cx, cy, frame_w, frame_h):
        """
        获取目标所在的空间象限（4象限划分）
        
        用于更精细的空间注意力分析：
        Q1: 左上  Q2: 右上
        Q3: 左下  Q4: 右下
        """
        is_left = cx < frame_w / 2
        is_top = cy < frame_h / 2
        
        if is_left and is_top:
            return 'Q1_left_top'
        elif not is_left and is_top:
            return 'Q2_right_top'
        elif is_left and not is_top:
            return 'Q3_left_bottom'
        else:
            return 'Q4_right_bottom'
    
    def update_attention_heatmap(self, gaze_point: Tuple[float, float], 
                                  frame_size: Tuple[int, int]):
        """
        根据注视点更新注意力热图
        
        使用高斯核在注视点位置叠加注意力强度，并对历史热图进行指数衰减，
        模拟人类注意力的时间消退特性。
        
        参数：
            gaze_point: 注视点坐标（归一化到[0,1]的比例坐标）
            frame_size: 图像尺寸 (width, height)
        """
        frame_w, frame_h = frame_size
        
        # 初始化热图（首次调用）
        if self.attention_heatmap is None:
            self.attention_heatmap = np.zeros((frame_h, frame_w), dtype=np.float32)
        
        # 将归一化坐标转换为像素坐标
        gaze_x = int(gaze_point[0] * frame_w)
        gaze_y = int(gaze_point[1] * frame_h)
        
        # 边界裁剪
        gaze_x = np.clip(gaze_x, 0, frame_w - 1)
        gaze_y = np.clip(gaze_y, 0, frame_h - 1)
        
        # 热图时间衰减（模拟注意力消退）
        self.attention_heatmap *= self.heatmap_decay
        
        # 在注视点位置添加高斯激活
        # 使用高斯核模拟注意力的空间扩散效应（注视点周围区域也受到部分注意）
        sigma = self.heatmap_sigma
        
        # 创建高斯核的坐标范围
        y_min = max(0, gaze_y - 3 * sigma)
        y_max = min(frame_h, gaze_y + 3 * sigma)
        x_min = max(0, gaze_x - 3 * sigma)
        x_max = min(frame_w, gaze_x + 3 * sigma)
        
        # 生成高斯激活（向量化计算）
        y_coords = np.arange(y_min, y_max)
        x_coords = np.arange(x_min, x_max)
        if len(y_coords) > 0 and len(x_coords) > 0:
            xx, yy = np.meshgrid(x_coords, y_coords)
            gaussian = np.exp(-((xx - gaze_x)**2 + (yy - gaze_y)**2) / (2 * sigma**2))
            self.attention_heatmap[y_min:y_max, x_min:x_max] += gaussian
        
        # 归一化到 [0, 1]
        max_val = self.attention_heatmap.max()
        if max_val > 0:
            self.attention_heatmap = self.attention_heatmap / max_val
    
    def compute_neglect_score(self, scene_result: Dict) -> Dict:
        """
        计算空间忽略评分
        
        忽略评分基于以下逻辑：
        1. 场景中存在 N 个目标物体（由 YOLOv11 检测）
        2. 患者实际注视了其中 M 个目标（由眼动数据和注意力热图判断）
        3. 忽略率 = 1 - M/N，分别在左侧和右侧空间计算
        
        对于中风患者，左侧忽略率通常远高于右侧
        
        参数：
            scene_result: detect_scene() 返回的检测结果
        
        返回：
            neglect_scores: 忽略评分字典
        """
        detections = scene_result['detections']
        
        if not detections:
            return {'overall': 0.0, 'left': 0.0, 'right': 0.0, 'detail': []}
        
        if self.attention_heatmap is None:
            # 热图未初始化，无法计算忽略分数
            return {'overall': 0.5, 'left': 0.5, 'right': 0.5, 'detail': []}
        
        target_details = []
        left_scores = []
        right_scores = []
        
        for det in detections:
            x1, y1, x2, y2 = det['bbox']
            
            # 提取该目标区域的注意力热图值
            # 注意：需要边界裁剪
            h, w = self.attention_heatmap.shape
            x1_c = max(0, min(x1, w-1))
            x2_c = max(0, min(x2, w-1))
            y1_c = max(0, min(y1, h-1))
            y2_c = max(0, min(y2, h-1))
            
            if x2_c <= x1_c or y2_c <= y1_c:
                attention_score = 0.0
            else:
                # 目标区域的平均注意力强度作为该目标的注意力评分
                roi_attention = self.attention_heatmap[y1_c:y2_c, x1_c:x2_c]
                attention_score = float(roi_attention.mean())
            
            # 判断是否被忽略
            is_neglected = attention_score < self.neglect_score_threshold
            
            detail = {
                'object': det['class_cn'],
                'side': det['side'],
                'attention_score': attention_score,
                'is_neglected': is_neglected,
                'bbox': det['bbox']
            }
            target_details.append(detail)
            
            if det['side'] == 'left':
                left_scores.append(attention_score)
            else:
                right_scores.append(attention_score)
        
        # 计算忽略评分
        # 注意：忽略评分 = 平均注意力分数的补数（越高说明越被忽略）
        left_neglect = 1.0 - (np.mean(left_scores) if left_scores else 0.5)
        right_neglect = 1.0 - (np.mean(right_scores) if right_scores else 0.5)
        overall_neglect = 1.0 - np.mean([d['attention_score'] for d in target_details])
        
        # 计算不对称性（左右忽略差异）
        asymmetry = left_neglect - right_neglect  # 正值→左侧忽略更严重（典型单侧忽略）
        
        return {
            'overall': float(np.clip(overall_neglect, 0, 1)),
            'left': float(np.clip(left_neglect, 0, 1)),
            'right': float(np.clip(right_neglect, 0, 1)),
            'asymmetry': float(asymmetry),  # 正值=左侧忽略，负值=右侧忽略
            'detail': target_details,
            'neglected_objects': [d['object'] for d in target_details if d['is_neglected']]
        }
    
    def visualize_results(self, frame: np.ndarray, scene_result: Dict,
                           neglect_scores: Dict, eeg_state: str = 'unknown') -> np.ndarray:
        """
        综合可视化：检测框 + 注意力热图 + 忽略区域标记
        
        参数：
            frame: 原始图像
            scene_result: 场景检测结果
            neglect_scores: 忽略评分
            eeg_state: EEG 判定的注意力状态
        
        返回：
            vis_frame: 可视化后的图像
        """
        vis_frame = frame.copy()
        frame_h, frame_w = frame.shape[:2]
        
        # ---- 1. 叠加注意力热图 ----
        if self.attention_heatmap is not None:
            # 将热图调整到与帧相同的大小
            heatmap_resized = cv2.resize(self.attention_heatmap, (frame_w, frame_h))
            
            # 转换为彩色热图（COLORMAP_JET: 蓝→绿→红，注意力从低到高）
            heatmap_colored = cv2.applyColorMap(
                (heatmap_resized * 255).astype(np.uint8), 
                cv2.COLORMAP_JET
            )
            
            # 半透明叠加（alpha=0.4）
            vis_frame = cv2.addWeighted(vis_frame, 0.6, heatmap_colored, 0.4, 0)
        
        # ---- 2. 绘制左右分界线 ----
        mid_x = frame_w // 2
        cv2.line(vis_frame, (mid_x, 0), (mid_x, frame_h), (255, 255, 0), 2)
        cv2.putText(vis_frame, 'LEFT', (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 
                   0.8, (255, 255, 0), 2)
        cv2.putText(vis_frame, 'RIGHT', (mid_x + 10, 30), cv2.FONT_HERSHEY_SIMPLEX, 
                   0.8, (255, 255, 0), 2)
        
        # ---- 3. 绘制检测框 ----
        for det in scene_result['detections']:
            x1, y1, x2, y2 = det['bbox']
            
            # 判断该目标是否被忽略
            is_neglected = any(
                d['is_neglected'] and d['bbox'] == det['bbox'] 
                for d in neglect_scores.get('detail', [])
            )
            
            if is_neglected:
                # 被忽略的目标：红色虚线框 + 感叹号标记
                color = (0, 0, 255)  # 红色（BGR）
                cv2.rectangle(vis_frame, (x1, y1), (x2, y2), color, 3)
                
                # 添加忽略警告标记（红色背景）
                label = f"! {det['class_cn']} (忽略)"
                cv2.rectangle(vis_frame, (x1, y1-30), (x1+len(label)*12, y1), color, -1)
                cv2.putText(vis_frame, label, (x1, y1-8), 
                           cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
            else:
                # 正常注意的目标：绿色框
                color = (0, 255, 0)  # 绿色
                cv2.rectangle(vis_frame, (x1, y1), (x2, y2), color, 2)
                label = f"{det['class_cn']} {det['conf']:.2f}"
                cv2.putText(vis_frame, label, (x1, y1-5), 
                           cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
        
        # ---- 4. 绘制状态信息面板 ----
        panel_h = 120
        panel = np.zeros((panel_h, frame_w, 3), dtype=np.uint8)
        panel[:] = (30, 30, 30)  # 深灰色背景
        
        # 忽略评分显示
        cv2.putText(panel, f"左侧忽略: {neglect_scores['left']:.1%}", 
                   (10, 25), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (100, 100, 255), 2)
        cv2.putText(panel, f"右侧忽略: {neglect_scores['right']:.1%}", 
                   (10, 55), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (100, 255, 100), 2)
        cv2.putText(panel, f"总体忽略: {neglect_scores['overall']:.1%}", 
                   (10, 85), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 100), 2)
        
        # 目标数量统计
        cv2.putText(panel, f"检测目标: {scene_result['spatial_distribution']['total']}个", 
                   (frame_w//2, 25), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (200, 200, 200), 2)
        cv2.putText(panel, f"EEG状态: {eeg_state}", 
                   (frame_w//2, 55), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (200, 200, 200), 2)
        
        # 被忽略物体列表
        neglected = neglect_scores.get('neglected_objects', [])
        if neglected:
            neglect_str = f"忽略物体: {', '.join(neglected[:3])}"
            cv2.putText(panel, neglect_str, (frame_w//2, 85), 
                       cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 100, 255), 2)
        
        # 拼接面板与图像
        vis_frame = np.vstack([vis_frame, panel])
        
        return vis_frame


# ======================== 测试入口 ========================
if __name__ == '__main__':
    print("=" * 60)
    print("YOLOv11 中风康复检测器测试")
    print("=" * 60)
    
    # 初始化检测器（使用轻量级 yolo11n 模型）
    detector = StrokeRehabDetector(
        model_path='yolo11n.pt',
        conf_threshold=0.4
    )
    
    # 创建模拟测试帧（实际使用时从摄像头或视频读取）
    test_frame = np.random.randint(50, 200, (480, 640, 3), dtype=np.uint8)
    
    # 添加一些模拟物体区域（矩形色块模拟物体）
    # 左侧物体（模拟容易被忽略的左侧目标）
    test_frame[100:200, 50:150] = [0, 200, 0]      # 左上绿色块（模拟杯子）
    test_frame[250:350, 80:180] = [200, 100, 0]     # 左下蓝色块（模拟书）
    # 右侧物体
    test_frame[100:200, 450:580] = [0, 0, 200]      # 右上红色块（模拟手机）
    test_frame[280:360, 430:560] = [200, 200, 0]    # 右下黄色块（模拟遥控器）
    
    # 执行场景检测
    print("\n执行场景检测...")
    scene_result = detector.detect_scene(test_frame)
    print(f"检测到物体数: {scene_result['spatial_distribution']['total']}")
    
    # 模拟眼动数据（模拟患者主要注视右侧，忽略左侧）
    print("\n模拟注视数据（偏向右侧）...")
    frame_size = (640, 480)
    
    # 右侧注视点序列（模拟患者注视右侧）
    gaze_points_right = [(0.7 + 0.1 * np.random.randn(), 
                          0.4 + 0.1 * np.random.randn()) for _ in range(30)]
    
    for gaze in gaze_points_right:
        # 裁剪到有效范围
        gx = np.clip(gaze[0], 0.01, 0.99)
        gy = np.clip(gaze[1], 0.01, 0.99)
        detector.update_attention_heatmap((gx, gy), frame_size)
    
    # 少量左侧注视（模拟偶尔扫描）
    gaze_points_left = [(0.2 + 0.05 * np.random.randn(), 
                         0.5 + 0.05 * np.random.randn()) for _ in range(5)]
    for gaze in gaze_points_left:
        gx = np.clip(gaze[0], 0.01, 0.99)
        gy = np.clip(gaze[1], 0.01, 0.99)
        detector.update_attention_heatmap((gx, gy), frame_size)
    
    # 计算忽略评分
    print("\n计算空间忽略评分...")
    neglect_scores = detector.compute_neglect_score(scene_result)
    print(f"左侧忽略评分: {neglect_scores['left']:.1%}")
    print(f"右侧忽略评分: {neglect_scores['right']:.1%}")
    print(f"整体忽略评分: {neglect_scores['overall']:.1%}")
    print(f"不对称性指数: {neglect_scores['asymmetry']:+.3f}")
    
    if neglect_scores['asymmetry'] > 0.2:
        print("⚠️ 检测到显著左侧空间忽略模式！建议进行进一步临床评估。")
    
    print("\n✅ 检测器测试完成！")

代码深度解析：

detect_scene 方法的设计思路：该方法不仅执行标准的 YOLOv11 目标检测，还额外对每个检测目标进行空间象限分类。将图像分为左右两个半区（对应患者视野的左右侧），以及更细粒度的四象限分类，为后续的单侧忽略分析提供空间分辨率。

update_attention_heatmap 的指数衰减机制：系数 0.95 意味着每帧（约 33ms@30fps）后，热图强度衰减为原来的 95%。这模拟了人类工作记忆对注意力历史的遗忘效应——约 1 秒前（30帧）的注视点对当前热图的贡献约为 0.95^30 ≈ 0.21，基本可忽略。这种设计使热图能够动态反映患者的当前注意力状态，而非整个评估期间的累积状态。

compute_neglect_score 的临床逻辑：忽略评分的核心逻辑是"应注意 vs 实际注意"的差距量化。YOLOv11 给出"场景中存在哪些物体"（应注意），注意力热图给出"患者实际关注了哪里"（实际注意），两者的对比揭示了被忽略的目标。

六、EEG 注意力分类器：EEGNet 实现

6.1 EEGNet 架构原理

EEGNet 是专为 EEG 信号设计的紧凑型深度神经网络，由 Lawhern 等人于 2018 年提出，发表于 Journal of Neural Engineering。其核心设计思想是：用**深度可分离卷积（Depthwise Separable Convolution）**替代标准卷积，在保持分类性能的同时大幅减少参数量（相比传统 CNN 减少约 95% 的参数），使其非常适合受限的嵌入式部署场景。

# eegnet_attention_classifier.py
# EEGNet 脑电注意力状态分类器
# 用于实时判断中风患者的注意力状态：
# 0: focused（集中注意）, 1: distracted（分散注意）
# 2: neglecting_left（左侧忽略）, 3: fatigued（疲劳）

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from torch.utils.data import Dataset, DataLoader
from sklearn.metrics import confusion_matrix, classification_report
import warnings
warnings.filterwarnings('ignore')


class EEGNet(nn.Module):
    """
    EEGNet 实现
    
    参考论文：Lawhern et al., 2018, Journal of Neural Engineering
    "EEGNet: A Compact Convolutional Neural Network for EEG-based Brain-Computer Interfaces"
    
    关键超参数：
        F1: 时间滤波器数量（控制频率分辨率）
        D: 深度乘数（控制空间滤波器数量 = D × F1）
        F2: 分离卷积输出通道数（通常 = D × F1）
        
    架构特点：
        1. 时间卷积：学习频率特征（等价于带通滤波器组）
        2. 深度卷积：学习空间特征（等价于空间滤波器/波束成形）
        3. 分离卷积：学习时频联合特征
    """
    
    def __init__(self, n_classes=4, n_channels=8, n_times=128, 
                 F1=8, D=2, F2=16, dropout_rate=0.5):
        """
        初始化 EEGNet
        
        参数：
            n_classes: 分类类别数（4种注意力状态）
            n_channels: EEG 通道数（C）
            n_times: 时间点数（T），即 epoch 长度
            F1: 时间卷积滤波器数（频率分辨率）
            D: 深度乘数
            F2: 分离卷积输出通道数
            dropout_rate: Dropout 比率（防过拟合）
        """
        super(EEGNet, self).__init__()
        
        self.n_classes = n_classes
        self.n_channels = n_channels
        self.n_times = n_times
        
        # ========== Block 1：时频特征提取 ==========
        
        # 步骤1.1：时间卷积（Temporal Convolution）
        # 学习频率特征，卷积核长度 = sfreq/2 对应最低可分辨频率
        # 对于 128Hz 数据，kernel_size=64 对应最低 2Hz 分辨率
        self.temporal_conv = nn.Conv2d(
            in_channels=1,          # 输入通道：1（EEG数据视为单通道图像）
            out_channels=F1,         # F1 个时间滤波器
            kernel_size=(1, n_times // 2),  # 沿时间轴的卷积核
            padding=(0, n_times // 4),       # 保持时间维度长度
            bias=False               # 不使用偏置（BN 后的偏置无意义）
        )
        self.bn1 = nn.BatchNorm2d(F1)  # 批归一化：加速收敛，防过拟合
        
        # 步骤1.2：深度卷积（Depthwise Convolution）
        # 在每个时间滤波器输出上独立进行空间滤波（通道融合）
        # groups=F1 表示每个输入通道单独卷积（深度可分离的关键）
        self.depthwise_conv = nn.Conv2d(
            in_channels=F1,
            out_channels=D * F1,    # 输出 D×F1 个空间-频率特征图
            kernel_size=(n_channels, 1),  # 沿通道（空间）轴的卷积核
            groups=F1,               # 关键！每个频率特征独立进行空间滤波
            bias=False
        )
        self.bn2 = nn.BatchNorm2d(D * F1)
        self.elu = nn.ELU()          # ELU 激活（比 ReLU 更适合 EEG 特征）
        self.pool1 = nn.AvgPool2d(kernel_size=(1, 4))  # 时间维度下采样4倍
        self.dropout1 = nn.Dropout(p=dropout_rate)
        
        # ========== Block 2：特征细化 ==========
        
        # 分离卷积（Separable Convolution = 深度卷积 + 逐点卷积）
        # 在时间维度上学习跨频率的联合特征
        self.separable_conv = nn.Sequential(
            # 深度卷积部分（时间方向）
            nn.Conv2d(
                in_channels=D * F1,
                out_channels=D * F1,
                kernel_size=(1, 16),     # 时间卷积核（约125ms @128Hz）
                padding=(0, 8),          # 保持长度
                groups=D * F1,           # 深度卷积
                bias=False
            ),
            # 逐点卷积部分（通道混合）
            nn.Conv2d(
                in_channels=D * F1,
                out_channels=F2,         # 输出 F2 个特征图
                kernel_size=(1, 1),      # 1×1 卷积（纯通道混合）
                bias=False
            )
        )
        self.bn3 = nn.BatchNorm2d(F2)
        self.pool2 = nn.AvgPool2d(kernel_size=(1, 8))  # 时间维度再下采样8倍
        self.dropout2 = nn.Dropout(p=dropout_rate)
        
        # ========== 分类层 ==========
        
        # 计算 Flatten 后的特征维度
        # 时间维度：n_times → pool1(÷4) → pool2(÷8) = n_times/32
        flatten_size = F2 * (n_times // 32)
        
        self.classifier = nn.Linear(flatten_size, n_classes)
        
        # 初始化权重（Xavier 初始化）
        self._initialize_weights()
        
        # 统计参数量
        total_params = sum(p.numel() for p in self.parameters())
        print(f"[EEGNet] 模型初始化完成")
        print(f"  输入形状: (batch, 1, {n_channels}, {n_times})")
        print(f"  分类类别: {n_classes}")
        print(f"  总参数量: {total_params:,} ({total_params/1e3:.1f}K)")
    
    def _initialize_weights(self):
        """Xavier 均匀初始化（适合 ELU 激活函数）"""
        for m in self.modules():
            if isinstance(m, nn.Conv2d):
                nn.init.xavier_uniform_(m.weight)
            elif isinstance(m, nn.Linear):
                nn.init.xavier_uniform_(m.weight)
                nn.init.zeros_(m.bias)
            elif isinstance(m, nn.BatchNorm2d):
                nn.init.ones_(m.weight)
                nn.init.zeros_(m.bias)
    
    def forward(self, x):
        """
        前向传播
        
        参数：
            x: 输入 EEG 数据，shape=(batch, 1, n_channels, n_times)
        
        返回：
            logits: 分类 logits，shape=(batch, n_classes)
        """
        # Block 1：时频特征提取
        x = self.temporal_conv(x)       # (B, F1, C, T)
        x = self.bn1(x)
        x = self.depthwise_conv(x)      # (B, D*F1, 1, T)
        x = self.bn2(x)
        x = self.elu(x)
        x = self.pool1(x)               # (B, D*F1, 1, T/4)
        x = self.dropout1(x)
        
        # Block 2：特征细化
        x = self.separable_conv(x)      # (B, F2, 1, T/4)
        x = self.bn3(x)
        x = self.elu(x)
        x = self.pool2(x)               # (B, F2, 1, T/32)
        x = self.dropout2(x)
        
        # 分类
        x = x.flatten(start_dim=1)      # (B, F2 * T/32)
        logits = self.classifier(x)      # (B, n_classes)
        
        return logits
    
    def predict_proba(self, x):
        """
        获取各类别概率（推理时使用）
        
        参数：
            x: 输入数据，shape=(batch, 1, n_channels, n_times)
        
        返回：
            proba: 概率分布，shape=(batch, n_classes)
        """
        self.eval()
        with torch.no_grad():
            logits = self.forward(x)
            proba = F.softmax(logits, dim=1)
        return proba


class EEGAttentionDataset(Dataset):
    """
    中风患者 EEG 注意力数据集
    
    标签定义：
        0: focused        - 集中注意（正常注意力）
        1: distracted     - 分散注意（内源性分心）
        2: neglecting_left - 左侧空间忽略（目标状态，中风特有）
        3: fatigued       - 认知疲劳（需要休息）
    """
    
    LABEL_NAMES = {
        0: '集中注意',
        1: '分散注意', 
        2: '左侧忽略',
        3: '认知疲劳'
    }
    
    def __init__(self, X, y, augment=False):
        """
        初始化数据集
        
        参数：
            X: EEG数据，shape=(n_samples, n_channels, n_times)
            y: 标签，shape=(n_samples,)
            augment: 是否使用数据增强
        """
        self.X = torch.FloatTensor(X)
        self.y = torch.LongTensor(y)
        self.augment = augment
        
        # 统计各类别样本数
        unique, counts = np.unique(y, return_counts=True)
        print(f"[Dataset] 样本统计: 总数={len(y)}")
        for cls_id, cnt in zip(unique, counts):
            print(f"  {self.LABEL_NAMES.get(cls_id, cls_id)}: {cnt} ({cnt/len(y):.1%})")
    
    def __len__(self):
        return len(self.y)
    
    def __getitem__(self, idx):
        x = self.X[idx].unsqueeze(0)  # 添加通道维度: (1, n_channels, n_times)
        y = self.y[idx]
        
        if self.augment and torch.rand(1) > 0.5:
            # 数据增强：随机高斯噪声（模拟电极接触差异）
            noise_level = 0.05 * x.std()
            x = x + torch.randn_like(x) * noise_level
        
        return x, y


class EEGNetTrainer:
    """
    EEGNet 训练管理器
    
    包含训练、验证、早停、学习率调度等完整训练流程
    """
    
    def __init__(self, model: EEGNet, device='cpu', lr=1e-3, weight_decay=1e-4):
        """
        初始化训练器
        
        参数：
            model: EEGNet 模型实例
            device: 训练设备
            lr: 初始学习率
            weight_decay: L2 正则化系数（防止过拟合）
        """
        self.model = model.to(device)
        self.device = device
        
        # 使用 AdamW 优化器（比 Adam 更好的权重衰减实现）
        self.optimizer = torch.optim.AdamW(
            model.parameters(), 
            lr=lr, 
            weight_decay=weight_decay
        )
        
        # 余弦退火学习率调度（带热重启）
        self.scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts(
            self.optimizer, T_0=20, T_mult=2
        )
        
        # 类别不平衡处理：加权交叉熵损失
        # 中风患者数据中"集中注意"样本通常最多，"左侧忽略"最少
        class_weights = torch.FloatTensor([1.0, 1.5, 2.0, 1.5]).to(device)
        self.criterion = nn.CrossEntropyLoss(weight=class_weights)
        
        # 训练历史记录
        self.train_history = {'loss': [], 'acc': []}
        self.val_history = {'loss': [], 'acc': []}
        
        # 早停参数
        self.best_val_acc = 0.0
        self.patience_counter = 0
        self.best_model_state = None
    
    def train_epoch(self, dataloader: DataLoader) -> Tuple[float, float]:
        """
        训练一个 epoch
        
        返回：(平均损失, 准确率)
        """
        self.model.train()
        total_loss = 0.0
        correct = 0
        total = 0
        
        for batch_idx, (data, target) in enumerate(dataloader):
            data = data.to(self.device)
            target = target.to(self.device)
            
            # 前向传播
            self.optimizer.zero_grad()
            output = self.model(data)
            loss = self.criterion(output, target)
            
            # 反向传播
            loss.backward()
            
            # 梯度裁剪（防止梯度爆炸，对 EEG 数据尤为重要）
            nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)
            
            self.optimizer.step()
            
            # 统计
            total_loss += loss.item() * data.size(0)
            pred = output.argmax(dim=1)
            correct += pred.eq(target).sum().item()
            total += data.size(0)
        
        avg_loss = total_loss / total
        accuracy = correct / total
        
        return avg_loss, accuracy
    
    def validate(self, dataloader: DataLoader) -> Tuple[float, float]:
        """
        验证模型性能
        
        返回：(平均损失, 准确率)
        """
        self.model.eval()
        total_loss = 0.0
        correct = 0
        total = 0
        
        with torch.no_grad():
            for data, target in dataloader:
                data = data.to(self.device)
                target = target.to(self.device)
                
                output = self.model(data)
                loss = self.criterion(output, target)
                
                total_loss += loss.item() * data.size(0)
                pred = output.argmax(dim=1)
                correct += pred.eq(target).sum().item()
                total += data.size(0)
        
        avg_loss = total_loss / total
        accuracy = correct / total
        
        return avg_loss, accuracy
    
    def fit(self, train_loader: DataLoader, val_loader: DataLoader,
            n_epochs: int = 100, patience: int = 15) -> dict:
        """
        完整训练流程（带早停）
        
        参数：
            train_loader: 训练数据加载器
            val_loader: 验证数据加载器
            n_epochs: 最大训练轮数
            patience: 早停耐心值（连续多少个epoch验证集无改善则停止）
        
        返回：
            history: 训练历史记录
        """
        print(f"\n开始训练 EEGNet...")
        print(f"  最大 Epochs: {n_epochs} | 早停耐心: {patience}")
        print(f"  设备: {self.device}")
        print("-" * 50)
        
        for epoch in range(n_epochs):
            # 训练一轮
            train_loss, train_acc = self.train_epoch(train_loader)
            val_loss, val_acc = self.validate(val_loader)
            
            # 记录历史
            self.train_history['loss'].append(train_loss)
            self.train_history['acc'].append(train_acc)
            self.val_history['loss'].append(val_loss)
            self.val_history['acc'].append(val_acc)
            
            # 学习率调度
            self.scheduler.step()
            current_lr = self.optimizer.param_groups[0]['lr']
            
            # 早停检查
            if val_acc > self.best_val_acc:
                self.best_val_acc = val_acc
                self.patience_counter = 0
                # 保存最优模型权重
                self.best_model_state = {
                    k: v.clone() for k, v in self.model.state_dict().items()
                }
                status = "✅ 最优"
            else:
                self.patience_counter += 1
                status = f"⏳ 等待({self.patience_counter}/{patience})"
            
            # 每5个epoch打印一次
            if (epoch + 1) % 5 == 0 or epoch == 0:
                print(f"Epoch [{epoch+1:3d}/{n_epochs}] "
                      f"训练损失:{train_loss:.4f} 训练准确率:{train_acc:.3f} | "
                      f"验证损失:{val_loss:.4f} 验证准确率:{val_acc:.3f} | "
                      f"LR:{current_lr:.6f} {status}")
            
            # 早停
            if self.patience_counter >= patience:
                print(f"\n早停触发（连续{patience}轮验证集无改善）")
                print(f"最优验证准确率: {self.best_val_acc:.4f}")
                break
        
        # 恢复最优模型
        if self.best_model_state is not None:
            self.model.load_state_dict(self.best_model_state)
            print(f"\n已恢复最优模型 (验证准确率: {self.best_val_acc:.4f})")
        
        return {
            'train': self.train_history,
            'val': self.val_history,
            'best_val_acc': self.best_val_acc
        }


# ======================== 完整训练测试 ========================
def run_eegnet_training_demo():
    """EEGNet 训练完整演示"""
    from typing import Tuple
    
    print("=" * 60)
    print("EEGNet 注意力分类器训练演示")
    print("=" * 60)
    
    # 超参数设置
    n_samples_per_class = 100   # 每类100个样本（实际项目建议500+）
    n_channels = 8               # EEG通道数
    n_times = 128                # 每个epoch的时间点数（对应约0.5秒@250Hz）
    n_classes = 4                # 分类类别数
    batch_size = 16              # 批大小
    
    # ---- 生成模拟 EEG 数据 ----
    # 在实际项目中，这里替换为真实的患者EEG数据
    np.random.seed(42)
    
    all_X = []
    all_y = []
    
    for cls_id in range(n_classes):
        # 不同类别有不同的频率特征（模拟真实EEG差异）
        X_cls = np.zeros((n_samples_per_class, n_channels, n_times))
        
        for i in range(n_samples_per_class):
            t = np.linspace(0, 0.5, n_times)  # 0.5秒时间轴
            
            if cls_id == 0:  # 集中注意：高beta，低alpha
                base_signal = 5 * np.sin(2 * np.pi * 20 * t)  # 强Beta
            elif cls_id == 1:  # 分散注意：高alpha
                base_signal = 15 * np.sin(2 * np.pi * 10 * t)  # 强Alpha
            elif cls_id == 2:  # 左侧忽略：右侧alpha不对称
                base_signal = 10 * np.sin(2 * np.pi * 10 * t)
            else:  # 认知疲劳：高theta
                base_signal = 8 * np.sin(2 * np.pi * 6 * t)   # 强Theta
            
            for ch in range(n_channels):
                # 添加通道特异性差异和随机噪声
                channel_gain = 1.0 + 0.3 * np.random.randn()
                if cls_id == 2 and ch >= 4:  # 忽略状态：右侧通道alpha更强
                    channel_gain *= 1.5
                
                X_cls[i, ch] = base_signal * channel_gain + np.random.randn(n_times) * 3
        
        all_X.append(X_cls)
        all_y.extend([cls_id] * n_samples_per_class)
    
    X = np.concatenate(all_X, axis=0)  # shape: (400, 8, 128)
    y = np.array(all_y)                 # shape: (400,)
    
    # 数据归一化（z-score，沿时间轴）
    X_mean = X.mean(axis=2, keepdims=True)
    X_std = X.std(axis=2, keepdims=True) + 1e-8
    X_normalized = (X - X_mean) / X_std
    
    # 训练/验证集划分（80/20）
    n_total = len(y)
    n_train = int(n_total * 0.8)
    
    # 随机打乱
    shuffle_idx = np.random.permutation(n_total)
    X_shuffled = X_normalized[shuffle_idx]
    y_shuffled = y[shuffle_idx]
    
    X_train = X_shuffled[:n_train]
    y_train = y_shuffled[:n_train]
    X_val = X_shuffled[n_train:]
    y_val = y_shuffled[n_train:]
    
    # 创建数据集和加载器
    train_dataset = EEGAttentionDataset(X_train, y_train, augment=True)
    val_dataset = EEGAttentionDataset(X_val, y_val, augment=False)
    
    train_loader = DataLoader(train_dataset, batch_size=batch_size, 
                              shuffle=True, num_workers=0)
    val_loader = DataLoader(val_dataset, batch_size=batch_size, 
                            shuffle=False, num_workers=0)
    
    # 创建模型
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    model = EEGNet(
        n_classes=n_classes,
        n_channels=n_channels,
        n_times=n_times,
        F1=8, D=2, F2=16,
        dropout_rate=0.5
    )
    
    # 创建训练器
    trainer = EEGNetTrainer(model, device=device, lr=1e-3)
    
    # 开始训练（演示只训练30轮）
    history = trainer.fit(train_loader, val_loader, n_epochs=30, patience=10)
    
    # 评估最终模型
    print("\n" + "=" * 50)
    print("最终评估结果")
    print("=" * 50)
    
    model.eval()
    all_preds = []
    all_targets = []
    
    with torch.no_grad():
        for data, target in val_loader:
            data = data.to(device)
            output = model(data)
            pred = output.argmax(dim=1).cpu().numpy()
            all_preds.extend(pred)
            all_targets.extend(target.numpy())
    
    # 分类报告
    class_names = ['集中注意', '分散注意', '左侧忽略', '认知疲劳']
    print("\n分类报告：")
    print(classification_report(all_targets, all_preds, 
                                 target_names=class_names, digits=4))
    
    # 保存模型
    torch.save({
        'model_state_dict': model.state_dict(),
        'model_config': {
            'n_classes': n_classes,
            'n_channels': n_channels,
            'n_times': n_times,
            'F1': 8, 'D': 2, 'F2': 16
        },
        'best_val_acc': history['best_val_acc'],
        'class_names': class_names
    }, 'eegnet_stroke_rehab.pth')
    
    print(f"\n模型已保存至: eegnet_stroke_rehab.pth")
    print(f"最优验证准确率: {history['best_val_acc']:.4f}")
    
    return model, history


if __name__ == '__main__':
    from typing import Tuple  # 确保 Tuple 在作用域内
    model, history = run_eegnet_training_demo()

EEGNet 架构深度解析：

为什么使用 EEGNet 而非 ResNet/LSTM？

在中风康复应用中，模型的选择需要权衡三个维度：

1. 分类性能：能够准确区分注意力状态
2. 实时性：在低算力的边缘设备（康复科平板/工控机）上达到实时推理（>30Hz）
3. 可解释性：临床医生需要理解模型判断依据

EEGNet 的优势在于其神经科学意义明确的架构设计：第一个时间卷积层学习到的滤波器实际上对应神经科学中的频率分解（类似傅里叶变换），第二个深度卷积层对应空间滤波（类似独立成分分析 ICA），这使得模型的中间特征对神经科学研究者而言具有可解释性。

权重初始化的重要性：Xavier 初始化能确保前向传播时各层的输出方差大致相当，避免梯度消失（所有层激活值趋近于0）或梯度爆炸（激活值过大）。对于 EEG 这类低信噪比数据，良好的初始化对收敛速度和最终性能有显著影响。

类别不平衡处理：在中风患者的真实数据中，"集中注意"样本（患者状态好时）通常远多于"左侧忽略"样本（患者状态差时），这会导致模型偏向多数类。使用加权交叉熵损失（weight=[1.0, 1.5, 2.0, 1.5]）可以对少数类（左侧忽略）赋予更高的惩罚权重，提升对临床关键状态的检测灵敏度。

七、多模态融合注意力评估系统

7.1 融合架构设计

相关示意图绘制如下，仅供参考：

# multimodal_fusion_system.py
# 多模态融合注意力评估系统
# 整合 YOLOv11 视觉分析 + EEGNet 神经分析，生成综合忽略评分

import numpy as np
import json
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional, Tuple
from datetime import datetime
import time


@dataclass
class AttentionAssessment:
    """
    单次注意力评估结果数据类
    
    包含视觉、神经、综合三个维度的评估指标
    """
    timestamp: float          # 评估时间戳（Unix时间）
    
    # 视觉维度指标
    visual_left_neglect: float    # 视觉左侧忽略评分 [0,1]
    visual_right_neglect: float   # 视觉右侧忽略评分 [0,1]
    detected_objects: int         # 场景中检测到的目标数
    neglected_objects: List[str]  # 被忽略的物体列表
    
    # 神经维度指标
    eeg_attention_state: str      # EEG判定的注意力状态
    eeg_state_confidence: float   # EEG分类置信度
    alpha_asymmetry: float        # Alpha波不对称性指数
    p300_amplitude_left: float    # 左侧刺激P300幅值 (μV)
    p300_amplitude_right: float   # 右侧刺激P300幅值 (μV)
    
    # 综合评估指标
    composite_neglect_score: float  # 综合忽略评分 [0,100]
    cognitive_fatigue_level: int    # 认知疲劳等级 [0,5]
    rehabilitation_recommendation: str  # 康复建议


class MultimodalFusionSystem:
    """
    多模态融合注意力评估系统
    
    融合策略：加权贝叶斯融合（Weighted Bayesian Fusion）
    
    权重分配依据（参考临床研究文献）：
    - 眼动数据（视觉忽略评分）权重 0.4：直接观测行为，最直观
    - EEG 注意力状态权重 0.35：神经层面的客观指标
    - Alpha 不对称性权重 0.15：半球间失衡的生理指标
    - P300 差异权重 0.10：时间分辨率最高，但信噪比最低
    """
    
    def __init__(self):
        # 各模态权重（基于文献中的效度研究）
        self.fusion_weights = {
            'visual_neglect': 0.40,    # 眼动+视觉分析权重
            'eeg_state': 0.35,         # EEG分类状态权重
            'alpha_asymmetry': 0.15,   # Alpha不对称性权重
            'p300_difference': 0.10    # P300幅值差权重
        }
        
        # 评估历史（用于趋势分析）
        self.assessment_history: List[AttentionAssessment] = []
        
        # 各模态指标的历史滑动窗口（用于平滑噪声）
        self.window_size = 5    # 5次评估的滑动平均
        self.visual_scores_window = []
        self.eeg_scores_window = []
        
        # 阈值参数（基于临床研究设定）
        self.SEVERE_NEGLECT_THRESHOLD = 65    # 严重忽略：综合评分 > 65
        self.MODERATE_NEGLECT_THRESHOLD = 40  # 中度忽略：综合评分 40-65
        self.FATIGUE_THRESHOLD = 3            # 疲劳级别 ≥ 3 建议休息
        
        print("[融合系统] 多模态注意力评估系统初始化完成")
    
    def _normalize_eeg_state_to_score(self, eeg_state: str, 
                                       eeg_probabilities: np.ndarray) -> float:
        """
        将 EEGNet 的分类结果转换为忽略评分 [0,1]
        
        转换逻辑：
        - 'focused'（集中）→ 低忽略分 (0.1)
        - 'distracted'（分散）→ 中低忽略分 (0.4)  
        - 'neglecting_left'（左侧忽略）→ 高忽略分 (0.9)
        - 'fatigued'（疲劳）→ 中忽略分 (0.6，疲劳容易导致忽略）
        
        参数：
            eeg_state: 分类状态字符串
            eeg_probabilities: 各类别概率 shape=(4,)
        
        返回：
            neglect_score: 忽略评分 [0,1]
        """
        # 基础映射分数（基于临床研究中各状态与忽略的相关性）
        base_scores = {
            'focused': 0.1,
            'distracted': 0.4,
            'neglecting_left': 0.9,
            'fatigued': 0.6
        }
        
        base_score = base_scores.get(eeg_state, 0.5)
        
        # 用概率的加权平均细化（避免二值化损失信息）
        # P(neglecting_left) 越高，即使分类为其他状态，也应该反映一定忽略风险
        if len(eeg_probabilities) == 4:
            probability_weights = np.array([0.1, 0.4, 0.9, 0.6])
            prob_weighted_score = np.dot(eeg_probabilities, probability_weights)
            
            # 基础分（70%权重）+ 概率加权（30%权重）
            final_score = 0.7 * base_score + 0.3 * prob_weighted_score
        else:
            final_score = base_score
        
        return float(np.clip(final_score, 0, 1))
    
    def _normalize_alpha_asymmetry(self, asymmetry: float) -> float:
        """
        将 Alpha 不对称性指数归一化到 [0,1] 的忽略评分
        
        Alpha 不对称性参考范围：
        - 健康成人：-0.2 ~ +0.2（基本对称，轻微个体差异）
        - 轻度忽略：+0.2 ~ +0.5（右侧 alpha 轻度升高→右侧激活弱→左侧轻度忽略）
        - 中度忽略：+0.5 ~ +1.0
        - 严重忽略：> +1.0
        
        参数：
            asymmetry: Alpha不对称性指数 ln(R) - ln(L)
        
        返回：
            neglect_score: 归一化后的忽略评分 [0,1]
        """
        # 使用 Sigmoid 函数进行非线性映射（平滑过渡）
        # 将 asymmetry=0 映射到 neglect_score=0.1（轻微基准忽略）
        # 将 asymmetry=1.0 映射到 neglect_score≈0.8（严重忽略）
        normalized = 1 / (1 + np.exp(-2 * (asymmetry - 0.3)))
        
        # 裁剪到有效范围
        return float(np.clip(normalized, 0, 1))
    
    def _normalize_p300_difference(self, p300_left: float, 
                                    p300_right: float) -> float:
        """
        将 P300 幅值差归一化为忽略评分
        
        正常人：左右刺激诱发的 P300 幅值基本相当（差异 < 2μV）
        忽略患者：左侧刺激 P300 幅值显著低于右侧（差异 > 5μV）
        
        参数：
            p300_left: 左侧位置刺激诱发的P300幅值（μV）
            p300_right: 右侧位置刺激诱发的P300幅值（μV）
        
        返回：
            neglect_score: 忽略评分 [0,1]
        """
        # P300 幅值差（右-左，正值表示左侧被忽略）
        p300_difference = p300_right - p300_left
        
        # 基于临床参考值进行归一化
        # 差异=0μV → 评分0.1（正常），差异=10μV → 评分≈0.9（严重忽略）
        normalized = np.clip(p300_difference / 10.0, 0, 1)
        
        # 若左侧P300实际高于右侧（提示右侧忽略），则给低分
        if p300_left > p300_right:
            normalized = max(0.0, normalized - 0.1)
        
        return float(normalized)
    
    def compute_composite_score(self, 
                                  visual_neglect_left: float,
                                  visual_neglect_right: float,
                                  eeg_state: str,
                                  eeg_probabilities: np.ndarray,
                                  alpha_asymmetry: float,
                                  p300_left: float,
                                  p300_right: float) -> float:
        """
        计算综合忽略评分（核心融合函数）
        
        融合策略：加权平均融合（各模态独立估计后加权求和）
        
        参数：
            visual_neglect_left: 视觉左侧忽略评分 [0,1]
            visual_neglect_right: 视觉右侧忽略评分 [0,1]
            eeg_state: EEG分类状态
            eeg_probabilities: EEG各类别概率
            alpha_asymmetry: Alpha不对称性指数
            p300_left: 左侧P300幅值
            p300_right: 右侧P300幅值
        
        返回：
            composite_score: 综合评分 [0,100]，越高表示忽略越严重
        """
        # 1. 视觉维度评分
        # 对于单侧忽略，主要关注左侧忽略（中风后最常见）
        # 右侧忽略也有，但权重略低（临床上右侧忽略较少，且预后更好）
        visual_score = 0.7 * visual_neglect_left + 0.3 * visual_neglect_right
        
        # 2. EEG状态评分
        eeg_score = self._normalize_eeg_state_to_score(eeg_state, eeg_probabilities)
        
        # 3. Alpha不对称性评分
        asymmetry_score = self._normalize_alpha_asymmetry(alpha_asymmetry)
        
        # 4. P300差异评分
        p300_score = self._normalize_p300_difference(p300_left, p300_right)
        
        # 加权融合
        w = self.fusion_weights
        composite_normalized = (
            w['visual_neglect'] * visual_score +
            w['eeg_state'] * eeg_score +
            w['alpha_asymmetry'] * asymmetry_score +
            w['p300_difference'] * p300_score
        )
        
        # 滑动平均平滑（减少单次噪声）
        self.visual_scores_window.append(visual_score)
        self.eeg_scores_window.append(eeg_score)
        
        if len(self.visual_scores_window) > self.window_size:
            self.visual_scores_window.pop(0)
            self.eeg_scores_window.pop(0)
        
        smoothed_visual = np.mean(self.visual_scores_window)
        smoothed_eeg = np.mean(self.eeg_scores_window)
        
        # 最终融合（原始值70%权重 + 平滑值30%权重）
        final_normalized = (
            0.7 * composite_normalized + 
            0.3 * (w['visual_neglect'] * smoothed_visual + w['eeg_state'] * smoothed_eeg) / 
            (w['visual_neglect'] + w['eeg_state'])
        )
        
        # 转换为0-100分制（更符合临床习惯）
        return float(np.clip(final_normalized * 100, 0, 100))
    
    def _assess_cognitive_fatigue(self, eeg_probabilities: np.ndarray,
                                   assessment_duration_min: float) -> int:
        """
        评估认知疲劳等级（0-5级）
        
        疲劳评级结合：EEGNet 疲劳概率 + 训练持续时间
        
        参数：
            eeg_probabilities: EEG各类别概率，shape=(4,)
            assessment_duration_min: 本次评估持续时间（分钟）
        
        返回：
            fatigue_level: 疲劳等级 0-5（0=无疲劳，5=严重疲劳）
        """
        # EEG 疲劳概率（第4类：fatigued）
        fatigue_prob = eeg_probabilities[3] if len(eeg_probabilities) > 3 else 0.0
        
        # 时间因素（康复训练超过30分钟后疲劳风险增加）
        time_factor = min(assessment_duration_min / 30.0, 1.0)
        
        # 综合疲劳分（EEG为主，时间为辅）
        fatigue_score = 0.7 * fatigue_prob + 0.3 * time_factor
        
        # 转换为等级
        if fatigue_score < 0.1:
            return 0  # 无疲劳
        elif fatigue_score < 0.25:
            return 1  # 轻微疲劳
        elif fatigue_score < 0.45:
            return 2  # 轻度疲劳
        elif fatigue_score < 0.60:
            return 3  # 中度疲劳（建议暂停）
        elif fatigue_score < 0.80:
            return 4  # 重度疲劳（需要休息）
        else:
            return 5  # 严重疲劳（立即停止训练）
    
    def _generate_recommendation(self, composite_score: float, 
                                   fatigue_level: int,
                                   neglected_objects: List[str]) -> str:
        """
        生成个性化康复训练建议
        
        参数：
            composite_score: 综合忽略评分
            fatigue_level: 疲劳等级
            neglected_objects: 本次被忽略的物体列表
        
        返回：
            recommendation: 中文康复建议字符串
        """
        recommendations = []
        
        # 疲劳优先判断
        if fatigue_level >= 4:
            return "⚠️ 检测到严重认知疲劳，请立即休息15-20分钟，勿强行继续训练"
        elif fatigue_level == 3:
            recommendations.append("建议暂停训练休息10分钟")
        
        # 基于忽略评分的建议
        if composite_score >= self.SEVERE_NEGLECT_THRESHOLD:
            recommendations.append("🔴 严重左侧忽略检测（评分≥65）")
            recommendations.append("增强左侧视觉扫描训练频率")
            recommendations.append("建议启动听觉引导模式（左侧语音提示）")
            recommendations.append("考虑增加棱镜适应疗法（Prismatic Adaptation）")
        elif composite_score >= self.MODERATE_NEGLECT_THRESHOLD:
            recommendations.append("🟡 中度左侧忽略检测（评分40-65）")
            recommendations.append("维持当前训练难度，增加左侧目标曝光频率")
            if neglected_objects:
                obj_str = '、'.join(neglected_objects[:3])
                recommendations.append(f"重点关注：{obj_str}等物体的左侧辨认")
        else:
            recommendations.append("🟢 注意力状态良好（评分<40）")
            recommendations.append("可适当增加场景复杂度（增加干扰项数量）")
            recommendations.append("尝试加入双任务训练（同时执行认知任务）")
        
        return " | ".join(recommendations) if recommendations else "继续当前训练方案"
    
    def assess(self, visual_result: Dict, eeg_features: Dict,
                session_start_time: float) -> AttentionAssessment:
        """
        执行一次完整的多模态注意力评估
        
        参数：
            visual_result: YOLOv11 检测模块输出的视觉结果字典
            eeg_features: EEG 分析模块输出的神经特征字典
            session_start_time: 本次训练会话开始时间戳
        
        返回：
            assessment: 完整的 AttentionAssessment 评估结果
        """
        current_time = time.time()
        session_duration_min = (current_time - session_start_time) / 60.0
        
        # 从视觉结果中提取关键指标
        neglect_scores = visual_result.get('neglect_scores', {})
        visual_left = neglect_scores.get('left', 0.5)
        visual_right = neglect_scores.get('right', 0.3)
        
        scene_result = visual_result.get('scene_result', {})
        detected_count = scene_result.get('spatial_distribution', {}).get('total', 0)
        neglected_objects = neglect_scores.get('neglected_objects', [])
        
        # 从EEG结果中提取关键指标
        eeg_state = eeg_features.get('attention_state', 'unknown')
        eeg_probabilities = np.array(eeg_features.get('probabilities', [0.25]*4))
        alpha_asymmetry = eeg_features.get('alpha_asymmetry', 0.0)
        p300_left = eeg_features.get('p300_amplitude_left', 8.0)    # 默认正常幅值
        p300_right = eeg_features.get('p300_amplitude_right', 8.0)
        eeg_confidence = float(np.max(eeg_probabilities))
        
        # 计算综合忽略评分
        composite_score = self.compute_composite_score(
            visual_left, visual_right,
            eeg_state, eeg_probabilities,
            alpha_asymmetry, p300_left, p300_right
        )
        
        # 评估疲劳等级
        fatigue_level = self._assess_cognitive_fatigue(
            eeg_probabilities, session_duration_min
        )
        
        # 生成康复建议
        recommendation = self._generate_recommendation(
            composite_score, fatigue_level, neglected_objects
        )
        
        # 构建评估结果
        assessment = AttentionAssessment(
            timestamp=current_time,
            visual_left_neglect=visual_left,
            visual_right_neglect=visual_right,
            detected_objects=detected_count,
            neglected_objects=neglected_objects,
            eeg_attention_state=eeg_state,
            eeg_state_confidence=eeg_confidence,
            alpha_asymmetry=alpha_asymmetry,
            p300_amplitude_left=p300_left,
            p300_amplitude_right=p300_right,
            composite_neglect_score=composite_score,
            cognitive_fatigue_level=fatigue_level,
            rehabilitation_recommendation=recommendation
        )
        
        # 记录到历史
        self.assessment_history.append(assessment)
        
        return assessment
    
    def generate_session_report(self) -> Dict:
        """
        生成本次训练会话的综合报告
        
        分析整个会话期间的注意力变化趋势，识别关键模式，
        为临床医生提供数据支撑
        
        返回：
            report: 会话报告字典
        """
        if not self.assessment_history:
            return {'error': '无评估历史数据'}
        
        # 提取时间序列数据
        scores = [a.composite_neglect_score for a in self.assessment_history]
        left_neglects = [a.visual_left_neglect for a in self.assessment_history]
        right_neglects = [a.visual_right_neglect for a in self.assessment_history]
        fatigue_levels = [a.cognitive_fatigue_level for a in self.assessment_history]
        
        # 统计指标
        score_mean = np.mean(scores)
        score_std = np.std(scores)
        score_trend = np.polyfit(range(len(scores)), scores, 1)[0]  # 线性趋势斜率
        
        # 判断改善/恶化趋势
        if score_trend < -1.0:
            trend_interpretation = "显著改善（注意力趋好）"
        elif score_trend < -0.3:
            trend_interpretation = "轻微改善"
        elif score_trend > 1.0:
            trend_interpretation = "显著恶化（建议评估疲劳或训练过量）"
        elif score_trend > 0.3:
            trend_interpretation = "轻微恶化"
        else:
            trend_interpretation = "基本稳定"
        
        # 严重忽略事件统计
        severe_events = sum(1 for s in scores if s >= self.SEVERE_NEGLECT_THRESHOLD)
        
        # 最常被忽略的物体
        all_neglected = []
        for assessment in self.assessment_history:
            all_neglected.extend(assessment.neglected_objects)
        
        from collections import Counter
        neglect_frequency = Counter(all_neglected).most_common(5)
        
        report = {
            'session_summary': {
                '总评估次数': len(self.assessment_history),
                '平均忽略评分': f"{score_mean:.1f}/100",
                '评分标准差': f"{score_std:.1f}",
                '变化趋势': f"{score_trend:+.2f}分/次 ({trend_interpretation})",
                '严重忽略事件': f"{severe_events}次 ({severe_events/len(scores):.1%})"
            },
            'spatial_analysis': {
                '平均左侧忽略': f"{np.mean(left_neglects):.1%}",
                '平均右侧忽略': f"{np.mean(right_neglects):.1%}",
                '忽略不对称性': f"{np.mean(left_neglects) - np.mean(right_neglects):+.1%}"
            },
            'fatigue_profile': {
                '最高疲劳等级': max(fatigue_levels),
                '平均疲劳等级': f"{np.mean(fatigue_levels):.1f}",
                '疲劳事件': sum(1 for f in fatigue_levels if f >= 3)
            },
            'neglected_objects_ranking': [
                {'物体': obj, '忽略次数': cnt} 
                for obj, cnt in neglect_frequency
            ],
            'clinical_recommendation': self._generate_clinical_summary(
                score_mean, score_trend, max(fatigue_levels)
            )
        }
        
        return report
    
    def _generate_clinical_summary(self, mean_score: float, trend: float, 
                                    max_fatigue: int) -> str:
        """
        生成面向临床医生的会话总结建议
        """
        summary_parts = []
        
        # 忽略严重程度评级
        if mean_score >= 65:
            summary_parts.append("本次会话患者表现出严重空间忽略症状（平均评分≥65），建议进行详细神经心理评估")
        elif mean_score >= 40:
            summary_parts.append("本次会话患者表现出中度空间忽略症状，训练效果需进一步观察")
        else:
            summary_parts.append("本次会话患者注意力控制良好，忽略症状较轻")
        
        # 疲劳建议
        if max_fatigue >= 4:
            summary_parts.append("会话中出现严重疲劳反应，建议缩短单次训练时长（≤20分钟）")
        
        # 趋势建议
        if trend < -0.5:
            summary_parts.append("评分呈显著下降趋势（注意力改善），可适当增加训练难度")
        elif trend > 0.5:
            summary_parts.append("评分呈上升趋势（注意力下降），建议检查训练强度是否过高")
        
        return "；".join(summary_parts)


# ======================== 系统集成测试 ========================
def test_multimodal_fusion():
    """多模态融合系统集成测试"""
    print("=" * 60)
    print("多模态融合注意力评估系统测试")
    print("=" * 60)
    
    # 初始化系统
    fusion_system = MultimodalFusionSystem()
    session_start = time.time()
    
    # 模拟10次评估（模拟10分钟康复训练）
    print("\n模拟10次评估...")
    
    # 模拟：训练初期忽略严重，随训练进行逐渐改善
    for trial in range(10):
        progress = trial / 9  # 0→1 表示训练进度
        
        # 模拟视觉结果（左侧忽略随训练减少）
        left_neglect = max(0.2, 0.8 - 0.5 * progress + 0.1 * np.random.randn())
        right_neglect = max(0.1, 0.2 + 0.05 * np.random.randn())
        
        visual_result = {
            'neglect_scores': {
                'left': np.clip(left_neglect, 0, 1),
                'right': np.clip(right_neglect, 0, 1),
                'neglected_objects': ['杯子', '书'] if left_neglect > 0.5 else ['书']
            },
            'scene_result': {
                'spatial_distribution': {'total': 5}
            }
        }
        
        # 模拟EEG结果（注意力状态随训练变化）
        if left_neglect > 0.6:
            eeg_state = 'neglecting_left'
            probs = [0.05, 0.10, 0.75, 0.10]
        elif left_neglect > 0.4:
            eeg_state = 'distracted'
            probs = [0.20, 0.50, 0.20, 0.10]
        else:
            eeg_state = 'focused'
            probs = [0.70, 0.15, 0.10, 0.05]
        
        eeg_features = {
            'attention_state': eeg_state,
            'probabilities': probs,
            'alpha_asymmetry': max(-0.5, 0.8 - 0.6 * progress + 0.1 * np.random.randn()),
            'p300_amplitude_left': min(15.0, 3.0 + 10.0 * progress + np.random.randn()),
            'p300_amplitude_right': 12.0 + np.random.randn()
        }
        
        # 执行评估
        assessment = fusion_system.assess(
            visual_result, eeg_features, session_start
        )
        
        print(f"\n[Trial {trial+1:2d}] 综合评分: {assessment.composite_neglect_score:.1f}/100 | "
              f"EEG状态: {assessment.eeg_attention_state} | "
              f"疲劳: {assessment.cognitive_fatigue_level}级")
        
        if trial == 0 or trial == 9:
            print(f"  左侧忽略: {assessment.visual_left_neglect:.1%} | "
                  f"Alpha不对称: {assessment.alpha_asymmetry:.3f}")
            print(f"  康复建议: {assessment.rehabilitation_recommendation}")
        
        # 模拟时间流逝
        time.sleep(0.01)  # 实际系统中约1秒/次评估
    
    # 生成会话报告
    print("\n" + "=" * 60)
    print("会话综合报告")
    print("=" * 60)
    
    report = fusion_system.generate_session_report()
    
    print("\n📊 会话总结：")
    for key, value in report['session_summary'].items():
        print(f"  {key}: {value}")
    
    print("\n🧠 空间分析：")
    for key, value in report['spatial_analysis'].items():
        print(f"  {key}: {value}")
    
    print("\n😴 疲劳概况：")
    for key, value in report['fatigue_profile'].items():
        print(f"  {key}: {value}")
    
    if report['neglected_objects_ranking']:
        print("\n📋 高频忽略物体：")
        for item in report['neglected_objects_ranking']:
            print(f"  {item['物体']}: {item['忽略次数']}次")
    
    print("\n🏥 临床建议：")
    print(f"  {report['clinical_recommendation']}")
    
    print("\n✅ 多模态融合系统测试完成！")
    return fusion_system, report


if __name__ == '__main__':
    import time  # 确保 time 在作用域内
    fusion_system, report = test_multimodal_fusion()

多模态融合代码解析：

融合权重设计的科学依据：

• 视觉忽略评分（40%）：眼动追踪是评估单侧忽略最直接的行为指标，具有高生态效度（直接观测患者"看哪里"），但可能受到患者主动代偿（刻意扫描左侧）的影响，导致低估真实忽略程度
• EEG注意力状态（35%）：神经信号不受患者主动控制的影响，反映更深层的神经状态；但信噪比相对较低，单次分类准确率约75-85%
• Alpha不对称性（15%）：具有高度的临床特异性（专门针对单侧忽略），但时间分辨率较低（需要数秒的数据才能稳定估计）
• P300差异（10%）：理论上是最精确的指标（直接测量神经对特定位置刺激的响应），但由于需要多次叠加平均，时间分辨率最低

贝叶斯思想的应用：系统使用滑动窗口平均对每次评估结果进行平滑，这体现了贝叶斯估计的思想——将历史观测（先验）与当前观测（似然）相结合，得到更稳健的后验估计。当前窗口大小为5次评估，约对应30秒的历史数据。

八、实时康复训练系统实现

8.1 自适应训练调度算法

# adaptive_training_scheduler.py
# 自适应康复训练调度器
# 根据实时评估结果动态调整训练参数

import numpy as np
import time
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum


class DifficultyLevel(Enum):
    """训练难度等级枚举"""
    EASY = 1        # 简单：少量目标，长反应窗口，强引导
    MEDIUM = 2      # 中等：适量目标，标准窗口
    HARD = 3        # 困难：多目标，有干扰物，短窗口
    EXPERT = 4      # 专家：复杂场景，动态目标，双任务


@dataclass
class TrainingParameters:
    """
    单次训练试次（Trial）的参数配置
    
    这些参数控制视觉刺激的呈现方式，直接影响训练效果
    """
    difficulty: DifficultyLevel   # 难度等级
    n_targets: int                 # 场景中目标物体数量
    n_distractors: int             # 干扰物数量
    response_window_ms: int        # 允许反应时间（毫秒）
    left_target_probability: float # 左侧出现目标的概率（忽略训练时提高此值）
    audio_guidance: bool           # 是否开启音频引导（忽略侧语音提示）
    highlight_left: bool           # 是否高亮左侧区域（用于严重忽略）
    inter_trial_interval_ms: int   # 试次间隔（毫秒）
    trial_type: str                # 试次类型：'detection'/'oddball'/'visual_search'


class AdaptiveTrainingScheduler:
    """
    自适应训练调度器
    
    核心功能：
    1. 基于实时评估结果动态调整训练难度
    2. 优化训练序列（避免疲劳、保持挑战性）
    3. 跟踪学习曲线，自动调整难度上升速率
    4. 生成训练计划报告
    """
    
    def __init__(self, initial_difficulty: DifficultyLevel = DifficultyLevel.MEDIUM):
        """
        初始化调度器
        
        参数：
            initial_difficulty: 初始训练难度（根据患者基线评估设定）
        """
        self.current_difficulty = initial_difficulty
        
        # 各难度等级的标准参数配置
        self.difficulty_configs = {
            DifficultyLevel.EASY: TrainingParameters(
                difficulty=DifficultyLevel.EASY,
                n_targets=3,                    # 少量目标
                n_distractors=0,                # 无干扰
                response_window_ms=5000,        # 5秒宽裕时间
                left_target_probability=0.7,   # 70%目标在左侧（强化左侧训练）
                audio_guidance=True,            # 开启语音引导
                highlight_left=True,            # 高亮左侧区域
                inter_trial_interval_ms=2000,  # 2秒间隔（充足休息）
                trial_type='detection'
            ),
            DifficultyLevel.MEDIUM: TrainingParameters(
                difficulty=DifficultyLevel.MEDIUM,
                n_targets=5,
                n_distractors=2,
                response_window_ms=3000,
                left_target_probability=0.5,   # 50%（平衡）
                audio_guidance=False,           # 无语音引导
                highlight_left=False,
                inter_trial_interval_ms=1500,
                trial_type='visual_search'
            ),
            DifficultyLevel.HARD: TrainingParameters(
                difficulty=DifficultyLevel.HARD,
                n_targets=8,
                n_distractors=5,
                response_window_ms=2000,
                left_target_probability=0.4,
                audio_guidance=False,
                highlight_left=False,
                inter_trial_interval_ms=1000,
                trial_type='visual_search'
            ),
            DifficultyLevel.EXPERT: TrainingParameters(
                difficulty=DifficultyLevel.EXPERT,
                n_targets=10,
                n_distractors=8,
                response_window_ms=1500,
                left_target_probability=0.35,  # 35%左侧（正常分布）
                audio_guidance=False,
                highlight_left=False,
                inter_trial_interval_ms=800,
                trial_type='oddball'            # 奇异范式（需要辨别目标/非目标）
            )
        }
        
        # 难度调整历史
        self.performance_history = []       # (评分, 时间戳) 列表
        self.difficulty_adjustments = []    # 难度调整记录
        
        # 难度调整策略参数
        self.upgrade_threshold = 35.0    # 低于此评分可升级难度
        self.downgrade_threshold = 60.0  # 高于此评分需降低难度
        self.min_trials_before_change = 3  # 至少3次评估才考虑调整
        self.stability_window = 3          # 基于最近3次评估判断稳定性
        
        print(f"[调度器] 初始化 | 起始难度: {initial_difficulty.name}")
    
    def determine_next_parameters(self, 
                                   composite_score: float,
                                   fatigue_level: int,
                                   consecutive_neglect_count: int = 0) -> TrainingParameters:
        """
        根据当前评估结果确定下一次试次的训练参数
        
        参数：
            composite_score: 综合忽略评分 [0,100]
            fatigue_level: 疲劳等级 [0,5]
            consecutive_neglect_count: 连续严重忽略次数
        
        返回：
            params: 下一次试次的训练参数
        """
        # 记录性能历史
        self.performance_history.append({
            'score': composite_score,
            'fatigue': fatigue_level,
            'timestamp': time.time()
        })
        
        # 疲劳优先处理
        if fatigue_level >= 4:
            # 严重疲劳：切换到最简单模式 + 强制引导
            params = self.difficulty_configs[DifficultyLevel.EASY]
            params.audio_guidance = True
            params.highlight_left = True
            params.inter_trial_interval_ms = 3000  # 延长间隔
            return params
        
        # 评估是否需要调整难度（至少有足够历史数据）
        if len(self.performance_history) >= self.min_trials_before_change:
            recent_scores = [p['score'] for p in 
                            self.performance_history[-self.stability_window:]]
            avg_recent_score = np.mean(recent_scores)
            score_variance = np.var(recent_scores)
            
            # 判断是否稳定（方差低表示稳定）
            is_stable = score_variance < 100  # 评分方差 < 100（即标准差 < 10分）
            
            if is_stable:
                if avg_recent_score < self.upgrade_threshold and \
                   self.current_difficulty != DifficultyLevel.EXPERT:
                    # 稳定且表现好 → 升级难度
                    self._increase_difficulty()
                elif avg_recent_score > self.downgrade_threshold and \
                     self.current_difficulty != DifficultyLevel.EASY:
                    # 稳定且表现差 → 降低难度
                    self._decrease_difficulty()
        
        # 获取当前难度配置
        params = self.difficulty_configs[self.current_difficulty]
        
        # 根据评分细化参数（在固定难度内微调）
        params = self._fine_tune_parameters(params, composite_score, consecutive_neglect_count)
        
        return params
    
    def _fine_tune_parameters(self, params: TrainingParameters,
                               score: float,
                               consecutive_neglect: int) -> TrainingParameters:
        """
        在固定难度等级内对参数进行微调
        
        不改变难度等级，但根据当前表现调整具体参数，
        实现难度的连续性（避免突变）
        
        参数：
            params: 当前难度级别的基础参数
            score: 当前综合忽略评分
            consecutive_neglect: 连续严重忽略次数
        
        返回：
            tuned_params: 微调后的参数
        """
        # 深拷贝避免修改原始配置
        import copy
        tuned = copy.deepcopy(params)
        
        # 连续严重忽略：增强左侧引导
        if consecutive_neglect >= 3 and score > 60:
            tuned.audio_guidance = True       # 开启语音提示
            tuned.left_target_probability = min(0.8, tuned.left_target_probability + 0.2)
            tuned.highlight_left = True       # 视觉高亮左侧
        
        # 严重忽略时：延长反应窗口，给予更多时间
        if score > 70:
            tuned.response_window_ms = int(tuned.response_window_ms * 1.3)
        
        # 表现良好时：轻微增加挑战（增加一个干扰物）
        if score < 30:
            tuned.n_distractors = min(tuned.n_distractors + 1, 10)
        
        return tuned
    
    def _increase_difficulty(self):
        """升级训练难度（一级）"""
        difficulty_order = [
            DifficultyLevel.EASY,
            DifficultyLevel.MEDIUM,
            DifficultyLevel.HARD,
            DifficultyLevel.EXPERT
        ]
        
        current_idx = difficulty_order.index(self.current_difficulty)
        if current_idx < len(difficulty_order) - 1:
            old_difficulty = self.current_difficulty
            self.current_difficulty = difficulty_order[current_idx + 1]
            
            self.difficulty_adjustments.append({
                'direction': 'increase',
                'from': old_difficulty.name,
                'to': self.current_difficulty.name,
                'timestamp': time.time(),
                'trigger_scores': [p['score'] for p in 
                                  self.performance_history[-self.stability_window:]]
            })
            
            print(f"[调度器] ⬆️ 难度升级: {old_difficulty.name} → {self.current_difficulty.name}")
    
    def _decrease_difficulty(self):
        """降低训练难度（一级）"""
        difficulty_order = [
            DifficultyLevel.EASY,
            DifficultyLevel.MEDIUM,
            DifficultyLevel.HARD,
            DifficultyLevel.EXPERT
        ]
        
        current_idx = difficulty_order.index(self.current_difficulty)
        if current_idx > 0:
            old_difficulty = self.current_difficulty
            self.current_difficulty = difficulty_order[current_idx - 1]
            
            self.difficulty_adjustments.append({
                'direction': 'decrease',
                'from': old_difficulty.name,
                'to': self.current_difficulty.name,
                'timestamp': time.time(),
                'trigger_scores': [p['score'] for p in 
                                  self.performance_history[-self.stability_window:]]
            })
            
            print(f"[调度器] ⬇️ 难度降级: {old_difficulty.name} → {self.current_difficulty.name}")
    
    def get_training_summary(self) -> Dict:
        """
        获取训练总结报告
        
        返回：
            summary: 包含学习曲线、难度调整记录等的综合报告
        """
        if not self.performance_history:
            return {}
        
        scores = [p['score'] for p in self.performance_history]
        
        # 计算学习进步率（线性拟合斜率）
        if len(scores) > 1:
            progress_rate = np.polyfit(range(len(scores)), scores, 1)[0]
        else:
            progress_rate = 0.0
        
        return {
            'total_trials': len(self.performance_history),
            'final_difficulty': self.current_difficulty.name,
            'difficulty_changes': len(self.difficulty_adjustments),
            'score_range': f"{min(scores):.1f} - {max(scores):.1f}",
            'average_score': f"{np.mean(scores):.1f}",
            'progress_rate': f"{progress_rate:+.2f}分/次",
            'learning_direction': "改善" if progress_rate < 0 else "退步",
            'adjustments_log': self.difficulty_adjustments
        }


# ======================== 调度器测试 ========================
if __name__ == '__main__':
    print("=" * 60)
    print("自适应训练调度器测试")
    print("=" * 60)
    
    # 从中等难度开始
    scheduler = AdaptiveTrainingScheduler(DifficultyLevel.MEDIUM)
    
    # 模拟15次试次，成绩从差到好
    test_scores = [75, 70, 65, 72, 60, 55, 50, 45, 40, 35, 30, 28, 25, 22, 20]
    test_fatigues = [1, 1, 2, 1, 2, 2, 2, 1, 1, 2, 1, 1, 2, 1, 2]
    
    print("\n模拟15次训练试次（成绩逐渐改善）：")
    consecutive_neglect = 0
    
    for i, (score, fatigue) in enumerate(zip(test_scores, test_fatigues)):
        if score > 65:
            consecutive_neglect += 1
        else:
            consecutive_neglect = 0
        
        params = scheduler.determine_next_parameters(score, fatigue, consecutive_neglect)
        
        print(f"\nTrial {i+1:2d} | 评分:{score:4.1f} | 疲劳:{fatigue}级 | "
              f"难度:{scheduler.current_difficulty.name}")
        print(f"  参数: 目标={params.n_targets}, 干扰={params.n_distractors}, "
              f"窗口={params.response_window_ms}ms, 语音={params.audio_guidance}")
    
    # 输出训练总结
    summary = scheduler.get_training_summary()
    print("\n" + "=" * 50)
    print("训练总结：")
    for key, value in summary.items():
        if key != 'adjustments_log':
            print(f"  {key}: {value}")
    
    print(f"\n难度调整日志 ({len(summary['adjustments_log'])}次):")
    for adj in summary['adjustments_log']:
        print(f"  {adj['direction']} | {adj['from']} → {adj['to']}")
    
    print("\n✅ 调度器测试完成！")

九、临床评估指标与康复进展追踪

9.1 康复效果评估框架

# rehabilitation_progress_tracker.py
# 康复进展追踪与可视化报告生成器

import numpy as np
import json
from datetime import datetime, timedelta
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass, field, asdict
import os


@dataclass 
class SessionRecord:
    """
    单次训练会话记录
    
    存储完整的会话信息，用于纵向进展分析
    """
    session_id: str               # 会话唯一标识符
    patient_id: str               # 患者ID（匿名化）
    date: str                     # 日期（YYYY-MM-DD格式）
    duration_minutes: float       # 会话时长（分钟）
    total_trials: int             # 试次总数
    
    # 核心评估指标
    mean_neglect_score: float     # 平均忽略评分 [0-100]
    mean_left_neglect: float      # 平均左侧忽略 [0-1]
    mean_right_neglect: float     # 平均右侧忽略 [0-1]
    min_neglect_score: float      # 最低忽略评分（最佳状态）
    max_neglect_score: float      # 最高忽略评分（最差状态）
    
    # 神经指标
    mean_alpha_asymmetry: float   # 平均Alpha不对称性
    mean_p300_left: float         # 平均左侧P300幅值 (μV)
    mean_p300_right: float        # 平均右侧P300幅值 (μV)
    
    # 疲劳与训练质量
    max_fatigue_level: int        # 最高疲劳等级
    rest_count: int               # 休息次数
    difficulty_at_end: str        # 结束时的难度等级
    
    # 被忽略的高频物体
    top_neglected_objects: List[str] = field(default_factory=list)
    
    # 康复师备注
    clinician_notes: str = ""


class RehabilitationProgressTracker:
    """
    中风患者康复进展追踪器
    
    功能：
    1. 管理患者的历史会话记录
    2. 计算进展指标（学习曲线、改善率等）
    3. 生成临床报告
    4. 识别康复"平台期"和"突破期"
    """
    
    def __init__(self, patient_id: str, data_dir: str = './rehab_data'):
        """
        初始化进展追踪器
        
        参数：
            patient_id: 患者匿名化ID
            data_dir: 数据存储目录
        """
        self.patient_id = patient_id
        self.data_dir = data_dir
        os.makedirs(data_dir, exist_ok=True)
        
        # 患者数据文件路径
        self.data_file = os.path.join(data_dir, f'patient_{patient_id}_sessions.json')
        
        # 加载历史数据（若存在）
        self.sessions: List[SessionRecord] = self._load_sessions()
        
        print(f"[进展追踪] 患者 {patient_id} | 历史会话: {len(self.sessions)} 次")
    
    def _load_sessions(self) -> List[SessionRecord]:
        """从文件加载历史会话记录"""
        if os.path.exists(self.data_file):
            try:
                with open(self.data_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                sessions = [SessionRecord(**s) for s in data]
                return sessions
            except Exception as e:
                print(f"[警告] 加载历史数据失败: {e}")
        return []
    
    def _save_sessions(self):
        """将会话记录保存到文件"""
        data = [asdict(s) for s in self.sessions]
        with open(self.data_file, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    def add_session(self, session: SessionRecord):
        """
        添加新的会话记录
        
        参数：
            session: 会话记录对象
        """
        self.sessions.append(session)
        self._save_sessions()
        print(f"[进展追踪] 新增会话记录 | ID: {session.session_id} | "
              f"日期: {session.date} | 评分: {session.mean_neglect_score:.1f}")
    
    def compute_progress_metrics(self) -> Dict:
        """
        计算综合康复进展指标
        
        返回：
            metrics: 进展指标字典
        """
        if len(self.sessions) < 2:
            return {'error': '历史会话不足，至少需要2次才能计算进展'}
        
        # 提取时序数据
        dates = [s.date for s in self.sessions]
        neglect_scores = [s.mean_neglect_score for s in self.sessions]
        left_neglects = [s.mean_left_neglect for s in self.sessions]
        alpha_asymmetries = [s.mean_alpha_asymmetry for s in self.sessions]
        p300_diffs = [s.mean_p300_right - s.mean_p300_left for s in self.sessions]
        
        # 1. 线性回归趋势（负斜率表示改善）
        n = len(neglect_scores)
        x = np.arange(n)
        
        neglect_trend = float(np.polyfit(x, neglect_scores, 1)[0])
        p300_trend = float(np.polyfit(x, p300_diffs, 1)[0])
        asymmetry_trend = float(np.polyfit(x, alpha_asymmetries, 1)[0])
        
        # 2. 改善率计算（首次 vs 最近3次平均）
        baseline_score = np.mean(neglect_scores[:min(3, n)])   # 基线（前3次）
        recent_score = np.mean(neglect_scores[-min(3, n):])    # 近期（后3次）
        improvement_rate = (baseline_score - recent_score) / baseline_score * 100
        
        # 3. 识别"平台期"（连续5次波动 < 5分）
        plateau_detected = False
        if n >= 5:
            recent_5 = neglect_scores[-5:]
            if np.std(recent_5) < 5 and np.abs(np.mean(recent_5) - recent_5[0]) < 5:
                plateau_detected = True
        
        # 4. P300 恢复进展（左右差值减小表示忽略改善）
        p300_recovery = float(p300_diffs[0] - p300_diffs[-1]) if len(p300_diffs) > 1 else 0
        
        # 5. 综合恢复指数（0-100，越高越好）
        # 基于：评分改善 + P300恢复 + Alpha不对称性改善
        recovery_components = []
        
        if improvement_rate > 0:
            recovery_components.append(min(improvement_rate * 2, 40))  # 最多贡献40分
        
        if p300_trend < 0:  # P300差值减小（改善）
            recovery_components.append(min(abs(p300_trend) * 10, 30))  # 最多贡献30分
        
        if asymmetry_trend < 0:  # 不对称性减小（改善）
            recovery_components.append(min(abs(asymmetry_trend) * 20, 30))  # 最多30分
        
        recovery_index = sum(recovery_components)
        
        metrics = {
            'session_count': n,
            'date_range': f"{dates[0]} → {dates[-1]}",
            'neglect_score_trend': {
                'slope': f"{neglect_trend:+.2f}分/次",
                'direction': "改善" if neglect_trend < 0 else "退步",
                'baseline': f"{baseline_score:.1f}",
                'recent': f"{recent_score:.1f}",
                'improvement_rate': f"{improvement_rate:+.1f}%"
            },
            'neural_recovery': {
                'p300_difference_change': f"{p300_recovery:+.2f}μV",
                'alpha_asymmetry_trend': f"{asymmetry_trend:+.4f}/次",
            },
            'rehabilitation_status': {
                'plateau_detected': plateau_detected,
                'plateau_advice': "建议更换训练策略以突破平台期" if plateau_detected else "训练进展正常",
                'recovery_index': f"{recovery_index:.0f}/100"
            },
            'clinical_assessment': self._generate_clinical_assessment(
                recovery_index, improvement_rate, plateau_detected
            )
        }
        
        return metrics
    
    def _generate_clinical_assessment(self, recovery_index: float,
                                       improvement_rate: float,
                                       plateau_detected: bool) -> str:
        """
        生成临床评估摘要
        """
        if recovery_index >= 60:
            assessment = "患者康复进展显著，建议提高训练难度和生态复杂度"
        elif recovery_index >= 30:
            assessment = "患者康复进展稳定，继续当前训练方案"
        elif recovery_index >= 10:
            assessment = "患者康复进展缓慢，建议结合物理康复和认知训练"
        else:
            assessment = "患者目前康复进展有限，建议重新评估康复目标和训练策略"
        
        if plateau_detected:
            assessment += "；检测到训练平台期，建议引入新的训练模式（如视觉扫描训练、棱镜适应疗法）"
        
        if improvement_rate > 30:
            assessment += f"；忽略评分改善幅度达 {improvement_rate:.0f}%，具有良好的康复预后"
        
        return assessment
    
    def generate_progress_report(self, output_format: str = 'text') -> str:
        """
        生成康复进展报告（临床文档格式）
        
        参数：
            output_format: 输出格式（'text' 或 'markdown'）
        
        返回：
            report: 格式化的进展报告字符串
        """
        metrics = self.compute_progress_metrics()
        
        if 'error' in metrics:
            return f"报告生成失败：{metrics['error']}"
        
        report_lines = [
            "=" * 60,
            f"中风患者视觉注意力康复进展报告",
            f"患者ID: {self.patient_id}（匿名化）",
            f"报告生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
            "=" * 60,
            "",
            "【一、训练概况】",
            f"  总训练次数: {metrics['session_count']} 次",
            f"  训练日期范围: {metrics['date_range']}",
            "",
            "【二、注意力忽略评分趋势】",
            f"  趋势斜率: {metrics['neglect_score_trend']['slope']} "
            f"（{metrics['neglect_score_trend']['direction']}）",
            f"  基线评分（前3次）: {metrics['neglect_score_trend']['baseline']}/100",
            f"  近期评分（后3次）: {metrics['neglect_score_trend']['recent']}/100",
            f"  改善幅度: {metrics['neglect_score_trend']['improvement_rate']}",
            "",
            "【三、神经生理指标恢复】",
            f"  P300幅值差变化: {metrics['neural_recovery']['p300_difference_change']}",
            f"  Alpha不对称性趋势: {metrics['neural_recovery']['alpha_asymmetry_trend']}",
            "",
            "【四、康复状态评估】",
            f"  综合恢复指数: {metrics['rehabilitation_status']['recovery_index']}",
            f"  平台期检测: {'是' if metrics['rehabilitation_status']['plateau_detected'] else '否'}",
            f"  建议: {metrics['rehabilitation_status']['plateau_advice']}",
            "",
            "【五、临床评估意见】",
            f"  {metrics['clinical_assessment']}",
            "",
            "=" * 60,
            "注：本报告由 YOLOv11-BCI 辅助康复系统自动生成，",
            "    需结合专业临床医生的综合评估使用。",
            "=" * 60
        ]
        
        return "\n".join(report_lines)


# ======================== 进展追踪测试 ========================
def test_progress_tracker():
    """康复进展追踪器测试"""
    print("=" * 60)
    print("康复进展追踪器测试")
    print("=" * 60)
    
    # 创建追踪器
    tracker = RehabilitationProgressTracker(
        patient_id='P2024_001', 
        data_dir='./test_rehab_data'
    )
    
    # 模拟8次会话数据（跨越4周训练）
    np.random.seed(42)
    session_dates = [
        (datetime.now() - timedelta(days=28-i*3.5)).strftime('%Y-%m-%d')
        for i in range(8)
    ]
    
    # 模拟渐进改善的康复曲线
    base_scores = [72, 68, 65, 60, 55, 48, 42, 38]  # 总体下降趋势（改善）
    base_p300_left = [4.0, 4.5, 5.0, 5.5, 6.0, 7.0, 8.0, 9.0]   # P300逐渐恢复
    base_p300_right = [12.0, 11.5, 11.0, 11.0, 10.5, 10.0, 10.0, 10.0]
    base_asymmetry = [0.9, 0.85, 0.78, 0.70, 0.62, 0.55, 0.48, 0.40]  # 不对称性改善
    
    difficulties = ['EASY', 'EASY', 'MEDIUM', 'MEDIUM', 'MEDIUM', 'HARD', 'HARD', 'HARD']
    
    print("\n添加模拟会话记录...")
    for i in range(8):
        score = base_scores[i] + np.random.randn() * 3
        
        session = SessionRecord(
            session_id=f"S{i+1:04d}",
            patient_id='P2024_001',
            date=session_dates[i],
            duration_minutes=25.0 + np.random.randn() * 3,
            total_trials=30,
            mean_neglect_score=float(np.clip(score, 0, 100)),
            mean_left_neglect=float(np.clip(score / 100 * 1.2, 0, 1)),
            mean_right_neglect=float(np.clip(score / 100 * 0.4, 0, 1)),
            min_neglect_score=float(np.clip(score - 15, 0, 100)),
            max_neglect_score=float(np.clip(score + 15, 0, 100)),
            mean_alpha_asymmetry=base_asymmetry[i] + np.random.randn() * 0.05,
            mean_p300_left=base_p300_left[i] + np.random.randn() * 0.5,
            mean_p300_right=base_p300_right[i] + np.random.randn() * 0.5,
            max_fatigue_level=2 if i < 4 else 1,
            rest_count=1 if score > 60 else 0,
            difficulty_at_end=difficulties[i],
            top_neglected_objects=['杯子', '书', '遥控器'][:3-i//3],
            clinician_notes=f"第{i+1}次会话，患者配合度良好" if i % 2 == 0 else ""
        )
        
        tracker.add_session(session)
    
    # 计算进展指标
    print("\n计算康复进展指标...")
    metrics = tracker.compute_progress_metrics()
    
    print("\n关键指标：")
    print(f"  总训练次数: {metrics['session_count']}")
    print(f"  评分趋势: {metrics['neglect_score_trend']['slope']} ({metrics['neglect_score_trend']['direction']})")
    print(f"  基线→近期: {metrics['neglect_score_trend']['baseline']} → {metrics['neglect_score_trend']['recent']}")
    print(f"  改善幅度: {metrics['neglect_score_trend']['improvement_rate']}")
    print(f"  综合恢复指数: {metrics['rehabilitation_status']['recovery_index']}")
    print(f"  平台期: {'已检测到' if metrics['rehabilitation_status']['plateau_detected'] else '未检测到'}")
    
    # 生成完整报告
    print("\n" + "=" * 60)
    report = tracker.generate_progress_report()
    print(report)
    
    # 清理测试文件
    import shutil
    if os.path.exists('./test_rehab_data'):
        shutil.rmtree('./test_rehab_data')
    
    return tracker


if __name__ == '__main__':
    tracker = test_progress_tracker()

代码解析总结：

RehabilitationProgressTracker 的设计体现了两个核心思想：

纵向数据管理：康复效果的评估不能局限于单次会话，必须在时间维度上追踪。系统将每次会话的完整评估数据持久化为 JSON 文件，支持跨会话的纵向分析。SessionRecord 数据类型涵盖了视觉、神经、行为三个维度的关键指标，为多维度进展分析提供数据基础。

科学的进展量化：compute_progress_metrics 使用线性回归而非简单的首末比较来评估趋势，因为线性回归对单次异常值（如患者某次状态特别差）具有更好的鲁棒性。"平台期检测"基于方差分析（连续5次评分的标准差），而非主观判断，这是将统计分析与临床需求结合的典型应用。

十、实时系统主控模块与系统集成

10.1 系统主控流程

# main_stroke_rehab_system.py
# 中风康复系统主控模块
# 整合 EEG 采集、YOLOv11 检测、多模态融合、自适应调度的完整系统

import numpy as np
import time
import threading
import queue
import cv2
from dataclasses import dataclass
from typing import Optional, Callable
import warnings
warnings.filterwarnings('ignore')

# 导入各子模块（实际使用时确保文件路径正确）
# from eeg_preprocessor import StrokeEEGPreprocessor
# from yolov11_stroke_detector import StrokeRehabDetector
# from eegnet_attention_classifier import EEGNet
# from multimodal_fusion_system import MultimodalFusionSystem
# from adaptive_training_scheduler import AdaptiveTrainingScheduler, DifficultyLevel
# from rehabilitation_progress_tracker import RehabilitationProgressTracker, SessionRecord


@dataclass
class SystemConfig:
    """
    系统配置参数
    
    集中管理所有系统级参数，便于部署时调整
    """
    # 硬件配置
    eeg_sfreq: int = 250          # EEG 采样率（Hz）
    eeg_n_channels: int = 8       # EEG 通道数
    camera_fps: int = 30          # 摄像头帧率
    camera_resolution: tuple = (640, 480)  # 分辨率 (W, H)
    
    # 模型配置
    yolo_model_path: str = 'yolo11n.pt'   # YOLOv11 模型路径
    eegnet_model_path: str = 'eegnet_stroke_rehab.pth'  # EEGNet 路径
    
    # 系统运行参数
    eeg_window_seconds: float = 0.5    # EEG 分析窗口（秒）
    assessment_interval: float = 1.0   # 评估间隔（秒）
    max_session_minutes: int = 30      # 最大会话时长（分钟）
    
    # 患者参数
    patient_id: str = 'P_TEST_001'
    initial_difficulty: str = 'MEDIUM'


class StrokeRehabSystem:
    """
    中风康复系统主控类
    
    采用多线程架构，各子系统独立运行、通过队列通信：
    - Thread 1: EEG 数据采集（250Hz 高频）
    - Thread 2: 视频帧采集（30fps）
    - Thread 3: EEG 特征提取与分类（1Hz）
    - Thread 4: 目标检测与忽略分析（30fps）
    - Thread 5: 多模态融合与评估（1Hz）
    - Main Thread: 显示、交互、调度
    """
    
    def __init__(self, config: SystemConfig):
        """
        初始化康复系统
        
        参数：
            config: 系统配置对象
        """
        self.config = config
        self.is_running = False
        
        # 线程间通信队列
        self.eeg_queue = queue.Queue(maxsize=10)       # EEG 数据队列
        self.frame_queue = queue.Queue(maxsize=5)       # 视频帧队列
        self.assessment_queue = queue.Queue(maxsize=10) # 评估结果队列
        
        # 会话状态变量
        self.session_start_time = None
        self.trial_count = 0
        self.consecutive_neglect = 0
        self.session_assessments = []
        
        # 系统状态标志
        self.forced_rest = False          # 是否在强制休息
        self.rest_end_time = None         # 休息结束时间
        
        print(f"[主系统] 中风康复系统初始化")
        print(f"  患者ID: {config.patient_id}")
        print(f"  最大会话时长: {config.max_session_minutes} 分钟")
        print(f"  初始难度: {config.initial_difficulty}")
    
    def _simulate_eeg_data(self, n_channels: int, n_samples: int) -> np.ndarray:
        """
        模拟 EEG 数据生成（替代真实硬件采集）
        
        在无硬件环境下，生成符合真实EEG统计特性的模拟数据，
        用于系统测试和演示
        
        参数：
            n_channels: 通道数
            n_samples: 采样点数
        
        返回：
            eeg_data: 模拟EEG数据，shape=(n_channels, n_samples)
        """
        sfreq = self.config.eeg_sfreq
        t = np.linspace(0, n_samples / sfreq, n_samples)
        
        # 生成真实感的多频段EEG（包含alpha主导特征）
        data = np.zeros((n_channels, n_samples))
        
        for ch in range(n_channels):
            # Alpha 波（8-12Hz）- EEG 主导成分
            alpha_power = 15 if ch >= 4 else 10  # 顶枕区alpha更强
            data[ch] += alpha_power * np.sin(2 * np.pi * 10 * t + ch * 0.5)
            
            # Beta 波（13-30Hz）
            data[ch] += 5 * np.sin(2 * np.pi * 20 * t + ch * 0.3)
            
            # 背景噪声
            data[ch] += np.random.randn(n_samples) * 3
        
        return data
    
    def _simulate_gaze_point(self, trial_progress: float) -> tuple:
        """
        模拟眼动注视点（基于忽略模式）
        
        模拟一个典型的左侧忽略患者：
        - 训练初期：注视点集中在右侧（忽略左侧）
        - 随训练进行：注视点逐渐向左侧扩展
        
        参数：
            trial_progress: 训练进度 [0,1]
        
        返回：
            (gaze_x, gaze_y): 归一化注视点坐标
        """
        # 随训练进度，左侧注视概率增加
        left_probability = 0.2 + 0.5 * trial_progress
        
        if np.random.rand() < left_probability:
            # 注视左侧
            gaze_x = 0.2 + 0.2 * np.random.rand()  # 左侧20-40%
        else:
            # 注视右侧
            gaze_x = 0.6 + 0.35 * np.random.rand()  # 右侧60-95%
        
        gaze_y = 0.2 + 0.6 * np.random.rand()  # 垂直方向随机
        
        return float(gaze_x), float(gaze_y)
    
    def run_demo_session(self, n_trials: int = 20):
        """
        运行演示会话（使用模拟数据，无需真实硬件）
        
        完整演示系统的评估-反馈-调整闭环
        
        参数：
            n_trials: 演示的试次数量
        """
        print("\n" + "=" * 60)
        print("中风康复系统 - 演示会话")
        print("=" * 60)
        print(f"演示模式：使用模拟EEG + 模拟眼动数据（无需真实硬件）")
        print(f"计划试次: {n_trials}")
        print("-" * 60)
        
        # 初始化子系统（演示版本，使用简化实现）
        sfreq = self.config.eeg_sfreq
        n_channels = self.config.eeg_n_channels
        window_samples = int(self.config.eeg_window_seconds * sfreq)
        
        # 会话统计
        self.session_start_time = time.time()
        all_neglect_scores = []
        all_eeg_states = []
        all_fatigue_levels = []
        
        # 难度追踪（简化版本）
        current_difficulty = self.config.initial_difficulty
        difficulty_changes = 0
        consecutive_low_score = 0  # 连续低分次数（用于升级难度）
        
        print(f"\n开始 {n_trials} 次试次评估...\n")
        
        for trial in range(n_trials):
            trial_start = time.time()
            trial_progress = trial / max(n_trials - 1, 1)  # 0→1
            
            # ---- 1. 模拟EEG数据采集 ----
            eeg_data = self._simulate_eeg_data(n_channels, window_samples)
            
            # ---- 2. EEG特征提取（简化版） ----
            # 计算各频段平均功率（简化特征）
            from scipy import signal as sp_signal
            
            def simple_band_power(data, fs, low, high):
                """快速频段功率计算"""
                b, a = sp_signal.butter(4, [low/(fs/2), high/(fs/2)], btype='band')
                filtered = sp_signal.filtfilt(b, a, data, axis=1)
                return np.mean(filtered**2)
            
            alpha_power_left = simple_band_power(
                eeg_data[:4], sfreq, 8, 12  # 左侧通道
            )
            alpha_power_right = simple_band_power(
                eeg_data[4:], sfreq, 8, 12  # 右侧通道
            )
            
            # 简化的Alpha不对称性计算
            alpha_asymmetry = float(
                np.log(alpha_power_right + 1e-10) - np.log(alpha_power_left + 1e-10)
            )
            # 添加随时间减小的趋势（模拟康复改善）
            alpha_asymmetry = max(-0.3, alpha_asymmetry - 0.05 * trial_progress) 
            
            # ---- 3. 模拟EEG状态分类 ----
            # 根据训练进度模拟状态变化（初期忽略，后期改善）
            rand_val = np.random.rand()
            
            if trial_progress < 0.3:
                # 训练初期：主要是忽略状态
                if rand_val < 0.6:
                    eeg_state = 'neglecting_left'
                    eeg_probs = np.array([0.08, 0.12, 0.70, 0.10])
                elif rand_val < 0.8:
                    eeg_state = 'distracted'
                    eeg_probs = np.array([0.15, 0.55, 0.20, 0.10])
                else:
                    eeg_state = 'focused'
                    eeg_probs = np.array([0.60, 0.20, 0.15, 0.05])
            elif trial_progress < 0.7:
                # 训练中期：逐渐过渡
                if rand_val < 0.35:
                    eeg_state = 'neglecting_left'
                    eeg_probs = np.array([0.12, 0.18, 0.60, 0.10])
                elif rand_val < 0.65:
                    eeg_state = 'distracted'
                    eeg_probs = np.array([0.25, 0.50, 0.18, 0.07])
                else:
                    eeg_state = 'focused'
                    eeg_probs = np.array([0.65, 0.20, 0.12, 0.03])
            else:
                # 训练后期：主要是集中状态
                if rand_val < 0.15:
                    eeg_state = 'neglecting_left'
                    eeg_probs = np.array([0.15, 0.20, 0.55, 0.10])
                elif rand_val < 0.35:
                    eeg_state = 'distracted'
                    eeg_probs = np.array([0.30, 0.45, 0.18, 0.07])
                else:
                    eeg_state = 'focused'
                    eeg_probs = np.array([0.72, 0.18, 0.07, 0.03])
            
            # 模拟P300幅值（随训练改善：左侧P300逐渐增大）
            p300_left = 3.0 + 7.0 * trial_progress + np.random.randn() * 1.0
            p300_right = 11.0 + np.random.randn() * 1.0
            
            # ---- 4. 模拟注视点更新 ----
            gaze_x, gaze_y = self._simulate_gaze_point(trial_progress)
            
            # ---- 5. 基于模拟数据计算视觉忽略评分 ----
            # 左侧忽略随训练进度减小
            base_left_neglect = max(0.2, 0.8 - 0.5 * trial_progress)
            left_neglect = float(np.clip(
                base_left_neglect + 0.1 * np.random.randn(), 0.1, 0.95
            ))
            right_neglect = float(np.clip(0.2 + 0.05 * np.random.randn(), 0.05, 0.4))
            
            # ---- 6. 计算综合忽略评分（简化版融合）----
            # 权重：视觉40%，EEG状态35%，Alpha不对称15%，P300 10%
            eeg_state_scores = {
                'focused': 0.1, 'distracted': 0.4, 
                'neglecting_left': 0.9, 'fatigued': 0.6
            }
            eeg_score = eeg_state_scores.get(eeg_state, 0.5)
            
            # EEG概率加权
            prob_weights = np.array([0.1, 0.4, 0.9, 0.6])
            eeg_score = 0.7 * eeg_score + 0.3 * float(np.dot(eeg_probs, prob_weights))
            
            # Alpha不对称归一化
            asymmetry_score = float(1 / (1 + np.exp(-2 * (alpha_asymmetry - 0.3))))
            
            # P300差异归一化
            p300_diff = p300_right - p300_left
            p300_score = float(np.clip(p300_diff / 10.0, 0, 1))
            
            # 视觉综合分
            visual_score = 0.7 * left_neglect + 0.3 * right_neglect
            
            # 加权融合
            composite_score = (
                0.40 * visual_score +
                0.35 * eeg_score +
                0.15 * asymmetry_score +
                0.10 * p300_score
            ) * 100
            
            composite_score = float(np.clip(composite_score, 0, 100))
            
            # ---- 7. 评估疲劳等级 ----
            fatigue_prob = float(eeg_probs[3])
            session_time_min = (time.time() - self.session_start_time) / 60
            fatigue_score = 0.7 * fatigue_prob + 0.3 * min(session_time_min / 30, 1)
            
            if fatigue_score < 0.1:
                fatigue_level = 0
            elif fatigue_score < 0.25:
                fatigue_level = 1
            elif fatigue_score < 0.45:
                fatigue_level = 2
            elif fatigue_score < 0.60:
                fatigue_level = 3
            else:
                fatigue_level = 4
            
            # ---- 8. 追踪记录 ----
            all_neglect_scores.append(composite_score)
            all_eeg_states.append(eeg_state)
            all_fatigue_levels.append(fatigue_level)
            
            # ---- 9. 难度自适应调整 ----
            if len(all_neglect_scores) >= 3:
                recent_avg = np.mean(all_neglect_scores[-3:])
                
                if recent_avg < 35 and current_difficulty != 'EXPERT':
                    consecutive_low_score += 1
                    if consecutive_low_score >= 3:
                        old_diff = current_difficulty
                        difficulties = ['EASY', 'MEDIUM', 'HARD', 'EXPERT']
                        curr_idx = difficulties.index(current_difficulty) if current_difficulty in difficulties else 1
                        if curr_idx < len(difficulties) - 1:
                            current_difficulty = difficulties[curr_idx + 1]
                            difficulty_changes += 1
                            consecutive_low_score = 0
                            print(f"  ⬆️ 难度升级: {old_diff} → {current_difficulty}")
                else:
                    consecutive_low_score = 0
                
                if recent_avg > 65 and current_difficulty != 'EASY':
                    difficulties = ['EASY', 'MEDIUM', 'HARD', 'EXPERT']
                    curr_idx = difficulties.index(current_difficulty) if current_difficulty in difficulties else 1
                    if curr_idx > 0:
                        old_diff = current_difficulty
                        current_difficulty = difficulties[curr_idx - 1]
                        difficulty_changes += 1
                        print(f"  ⬇️ 难度降级: {old_diff} → {current_difficulty}")
            
            # ---- 10. 生成康复建议 ----
            if composite_score >= 65:
                rec_level = "🔴 严重忽略"
                action = "开启语音引导 + 高亮左侧区域"
            elif composite_score >= 40:
                rec_level = "🟡 中度忽略"
                action = "增加左侧目标频率"
            else:
                rec_level = "🟢 状态良好"
                action = "可适当增加难度"
            
            # ---- 11. 输出本次试次结果 ----
            if (trial + 1) % 5 == 0 or trial == 0 or fatigue_level >= 3:
                print(f"[Trial {trial+1:2d}/{n_trials}] "
                      f"评分:{composite_score:5.1f} | "
                      f"状态:{eeg_state:16s} | "
                      f"难度:{current_difficulty:6s} | "
                      f"疲劳:{fatigue_level}级")
                print(f"           "
                      f"左忽略:{left_neglect:.1%} | "
                      f"Alpha非对称:{alpha_asymmetry:+.3f} | "
                      f"P300差:{p300_right - p300_left:+.1f}μV")
                print(f"           {rec_level} | 建议:{action}\n")
            
            # 疲劳警告
            if fatigue_level >= 3:
                print(f"  ⚠️ 疲劳警告（{fatigue_level}级），建议休息")
            
            # 模拟真实时间流逝（演示中加速）
            time.sleep(0.01)
        
        # ---- 会话结束统计 ----
        session_duration = (time.time() - self.session_start_time) / 60
        
        print("\n" + "=" * 60)
        print("会话结束统计报告")
        print("=" * 60)
        
        scores = all_neglect_scores
        print(f"\n📊 评估统计（共 {n_trials} 次试次）：")
        print(f"  初始评分: {scores[0]:.1f}/100")
        print(f"  最终评分: {scores[-1]:.1f}/100")
        print(f"  平均评分: {np.mean(scores):.1f}/100")
        print(f"  最优评分: {min(scores):.1f}/100")
        print(f"  最差评分: {max(scores):.1f}/100")
        print(f"  改善幅度: {scores[0] - scores[-1]:+.1f}分 "
              f"({'改善' if scores[0] > scores[-1] else '退步'})")
        
        # 趋势分析
        trend = float(np.polyfit(range(len(scores)), scores, 1)[0])
        print(f"\n📈 趋势分析：")
        print(f"  线性趋势斜率: {trend:+.2f}分/次")
        print(f"  整体趋势: {'✅ 持续改善' if trend < -0.5 else '⚡ 波动' if abs(trend) < 0.5 else '❌ 退步'}")
        
        # EEG状态分布
        from collections import Counter
        state_counts = Counter(all_eeg_states)
        print(f"\n🧠 EEG状态分布：")
        for state, count in state_counts.most_common():
            state_cn = {'focused':'集中', 'distracted':'分散', 
                       'neglecting_left':'左侧忽略', 'fatigued':'疲劳'}.get(state, state)
            print(f"  {state_cn}: {count}次 ({count/n_trials:.1%})")
        
        print(f"\n⚙️ 训练管理：")
        print(f"  难度调整次数: {difficulty_changes}")
        print(f"  结束难度: {current_difficulty}")
        print(f"  最高疲劳等级: {max(all_fatigue_levels)}")
        print(f"  会话实际时长: {session_duration:.1f}分钟")
        
        # 总体评价
        overall_improvement = scores[0] - np.mean(scores[-3:])
        if overall_improvement > 20:
            overall_verdict = "✅ 本次训练效果显著，患者表现出明显的注意力改善"
        elif overall_improvement > 8:
            overall_verdict = "🟡 本次训练效果良好，建议维持当前训练方案"
        else:
            overall_verdict = "⚠️ 本次训练改善有限，建议评估训练方案的适合性"
        
        print(f"\n🏥 临床总评：{overall_verdict}")
        print("=" * 60)
        
        return {
            'n_trials': n_trials,
            'mean_score': float(np.mean(scores)),
            'final_score': float(scores[-1]),
            'improvement': float(scores[0] - scores[-1]),
            'trend': trend,
            'difficulty_changes': difficulty_changes
        }


# ======================== 系统主入口 ========================
if __name__ == '__main__':
    # 系统配置
    config = SystemConfig(
        patient_id='DEMO_P001',
        initial_difficulty='MEDIUM',
        max_session_minutes=30
    )
    
    # 初始化并运行演示会话
    system = StrokeRehabSystem(config)
    result = system.run_demo_session(n_trials=20)
    
    print(f"\n✅ 演示完成 | 综合结果：")
    print(f"  平均忽略评分: {result['mean_score']:.1f}")
    print(f"  总体改善: {result['improvement']:+.1f}分")
    print(f"  难度调整: {result['difficulty_changes']}次")

十一、系统性能评估与伦理考量

11.1 系统性能基准测试

# performance_benchmark.py
# 系统性能基准测试
# 评估实时处理能力、内存占用、分类准确率等关键指标

import numpy as np
import time
import torch
import psutil
import os
from typing import Dict, List


def benchmark_system_performance(n_trials: int = 100) -> Dict:
    """
    系统性能综合基准测试
    
    测试项目：
    1. YOLOv11 推理速度（FPS）
    2. EEG 特征提取速度（samples/sec）
    3. 多模态融合延迟（ms）
    4. 内存占用
    5. GPU 利用率（若可用）
    
    参数：
        n_trials: 测试迭代次数
    
    返回：
        benchmark_results: 性能测试结果字典
    """
    print("=" * 60)
    print("系统性能基准测试")
    print("=" * 60)
    
    results = {}
    
    # ---- 测试1: EEG 信号处理速度 ----
    print("\n[1/4] EEG 信号处理速度测试...")
    
    sfreq = 250
    n_channels = 8
    window_seconds = 0.5
    window_samples = int(window_seconds * sfreq)
    
    # 生成测试数据
    test_eeg = np.random.randn(n_channels, window_samples)
    
    from scipy import signal as sp_signal
    
    def process_eeg_window(data, sfreq):
        """EEG单窗口处理流程"""
        # 陷波滤波
        b, a = sp_signal.iirnotch(50, 30, sfreq)
        data = sp_signal.filtfilt(b, a, data, axis=1)
        
        # 带通滤波
        b2, a2 = sp_signal.butter(4, [1/(sfreq/2), 40/(sfreq/2)], btype='band')
        data = sp_signal.filtfilt(b2, a2, data, axis=1)
        
        # 简单特征提取
        alpha_power = np.mean(data**2)
        return alpha_power
    
    # 预热
    for _ in range(5):
        process_eeg_window(test_eeg.copy(), sfreq)
    
    # 正式测试
    eeg_times = []
    for _ in range(n_trials):
        start = time.perf_counter()
        process_eeg_window(test_eeg.copy(), sfreq)
        eeg_times.append((time.perf_counter() - start) * 1000)  # ms
    
    eeg_mean = np.mean(eeg_times)
    eeg_std = np.std(eeg_times)
    eeg_fps = 1000 / eeg_mean
    
    results['eeg_processing'] = {
        '平均延迟': f"{eeg_mean:.2f} ms",
        '延迟标准差': f"{eeg_std:.2f} ms",
        '等效吞吐率': f"{eeg_fps:.1f} windows/sec",
        '实时性': "✅ 满足" if eeg_mean < 50 else "❌ 不满足"  # <50ms才能实时
    }
    print(f"  平均处理延迟: {eeg_mean:.2f}ms | 吞吐率: {eeg_fps:.1f} windows/sec")
    
    # ---- 测试2: EEGNet 推理速度 ----
    print("\n[2/4] EEGNet 推理速度测试...")
    
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    
    # 创建简化 EEGNet 进行测试
    class SimpleEEGNet(torch.nn.Module):
        def __init__(self):
            super().__init__()
            self.conv1 = torch.nn.Conv2d(1, 8, (1, 64), padding=(0, 32), bias=False)
            self.bn1 = torch.nn.BatchNorm2d(8)
            self.conv2 = torch.nn.Conv2d(8, 16, (8, 1), groups=8, bias=False)
            self.bn2 = torch.nn.BatchNorm2d(16)
            self.pool = torch.nn.AvgPool2d((1, 4))
            self.fc = torch.nn.Linear(16 * 32, 4)
            self.elu = torch.nn.ELU()
        
        def forward(self, x):
            x = self.elu(self.bn1(self.conv1(x)))
            x = self.elu(self.bn2(self.conv2(x)))
            x = self.pool(x)
            x = x.flatten(1)
            # 确保全连接层输入维度匹配
            if x.shape[1] != 16 * 32:
                # 自适应调整
                fc = torch.nn.Linear(x.shape[1], 4).to(x.device)
                x = fc(x)
            else:
                x = self.fc(x)
            return x
    
    model = SimpleEEGNet().to(device)
    model.eval()
    
    # 测试数据
    test_input = torch.randn(1, 1, 8, 128).to(device)
    
    # GPU预热
    with torch.no_grad():
        for _ in range(10):
            _ = model(test_input)
    
    if torch.cuda.is_available():
        torch.cuda.synchronize()
    
    # 正式测试
    eegnet_times = []
    with torch.no_grad():
        for _ in range(n_trials):
            start = time.perf_counter()
            _ = model(test_input)
            if torch.cuda.is_available():
                torch.cuda.synchronize()
            eegnet_times.append((time.perf_counter() - start) * 1000)
    
    eegnet_mean = np.mean(eegnet_times)
    eegnet_fps = 1000 / eegnet_mean
    
    results['eegnet_inference'] = {
        '设备': device,
        '平均延迟': f"{eegnet_mean:.3f} ms",
        '等效FPS': f"{eegnet_fps:.1f} fps",
        '实时性': "✅ 满足" if eegnet_mean < 10 else "⚠️ 接近" if eegnet_mean < 50 else "❌ 不满足"
    }
    print(f"  设备: {device} | 平均延迟: {eegnet_mean:.3f}ms | FPS: {eegnet_fps:.1f}")
    
    # ---- 测试3: YOLOv11 推理速度 ----
    print("\n[3/4] YOLOv11 推理速度测试...")
    
    try:
        from ultralytics import YOLO
        yolo_model = YOLO('yolo11n.pt')
        
        # 测试图像
        test_frame = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
        
        # 预热
        for _ in range(3):
            _ = yolo_model(test_frame, verbose=False)
        
        # 测试
        yolo_times = []
        for _ in range(min(n_trials, 20)):  # YOLOv11较慢，只测20次
            start = time.perf_counter()
            _ = yolo_model(test_frame, verbose=False)
            yolo_times.append((time.perf_counter() - start) * 1000)
        
        yolo_mean = np.mean(yolo_times)
        yolo_fps = 1000 / yolo_mean
        
        results['yolov11_inference'] = {
            '平均延迟': f"{yolo_mean:.2f} ms",
            '等效FPS': f"{yolo_fps:.1f} fps",
            '实时性（30fps需<33ms）': "✅ 满足" if yolo_mean < 33 else "⚠️ 接近"
        }
        print(f"  平均延迟: {yolo_mean:.2f}ms | FPS: {yolo_fps:.1f}")
        
    except Exception as e:
        results['yolov11_inference'] = {'状态': f'模型未找到，跳过测试 ({e})'}
        print(f"  模型未找到，跳过 ({e})")
    
    # ---- 测试4: 系统资源占用 ----
    print("\n[4/4] 系统资源占用测试...")
    
    process = psutil.Process(os.getpid())
    
    # 内存占用
    memory_mb = process.memory_info().rss / 1024 / 1024
    
    # CPU 使用率（1秒采样）
    cpu_percent = psutil.cpu_percent(interval=1.0)
    
    # 系统总内存
    total_memory_gb = psutil.virtual_memory().total / 1024**3
    
    results['system_resources'] = {
        'CPU使用率': f"{cpu_percent:.1f}%",
        '进程内存': f"{memory_mb:.0f} MB",
        '系统总内存': f"{total_memory_gb:.1f} GB",
        'GPU可用': str(torch.cuda.is_available()),
        '设备信息': f"{'CUDA GPU' if torch.cuda.is_available() else 'CPU'}"
    }
    print(f"  CPU: {cpu_percent:.1f}% | 内存: {memory_mb:.0f}MB | GPU: {torch.cuda.is_available()}")
    
    # ---- 综合评估 ----
    print("\n" + "=" * 60)
    print("综合性能评估报告")
    print("=" * 60)
    
    for module, metrics in results.items():
        module_names = {
            'eeg_processing': 'EEG 信号处理',
            'eegnet_inference': 'EEGNet 推理',
            'yolov11_inference': 'YOLOv11 推理',
            'system_resources': '系统资源'
        }
        print(f"\n▶ {module_names.get(module, module)}：")
        for key, value in metrics.items():
            print(f"  {key}: {value}")
    
    # 实时性综合判断
    print("\n" + "-" * 40)
    eeg_ok = eeg_mean < 50
    eegnet_ok = eegnet_mean < 50
    
    if eeg_ok and eegnet_ok:
        print("✅ 系统满足实时处理要求（延迟<50ms），可用于临床部署")
    else:
        print("⚠️ 系统部分模块存在延迟超标，建议优化或使用更强算力")
    
    return results


if __name__ == '__main__':
    results = benchmark_system_performance(n_trials=50)
    print("\n✅ 基准测试完成！")

性能测试结果参考值（RTX 3060 GPU 环境）：

模块	平均延迟	实时性要求	是否满足
EEG 信号处理	~8ms	<50ms	✅
EEGNet 推理（GPU）	~0.5ms	<10ms	✅
EEGNet 推理（CPU）	~12ms	<50ms	✅
YOLOv11n 推理（GPU）	~8ms	<33ms（30fps）	✅
YOLOv11n 推理（CPU）	~45ms	<33ms	⚠️ 接近
多模态融合	~2ms	<20ms	✅
端到端总延迟（GPU）	~20ms	<100ms	✅

11.2 伦理与隐私保护设计

在中风康复系统中，伦理和隐私保护是不可回避的重要议题：

数据匿名化：所有患者数据必须在采集端进行本地匿名化处理——使用随机生成的患者ID替代真实身份信息，在数据传输和存储中从不出现患者姓名、出生日期等个人识别信息。神经数据（EEG）属于高度敏感的生物特征数据，需要按照 HIPAA/GDPR 规定进行加密存储。

知情同意与自主权：系统应提供明确的使用说明和退出机制——患者可随时中止评估，中止数据不会被保留；任何评估结果不得在患者不知情的情况下与第三方共享。系统的算法建议仅供参考，最终医疗决策必须由执业医师做出。

算法公平性：EEGNet 模型的训练数据必须涵盖不同年龄、性别、损伤类型的患者群体，避免对特定群体（如老年患者、女性患者）产生系统性低估。应定期对模型进行公平性审计（Fairness Audit）。

十二、章节知识图谱与总结

12.1 本章知识点关系图

相关示意图绘制如下，仅供参考：

12.2 核心知识点总结

本节从神经科学原理出发，系统构建了一套完整的 YOLOv11 + BCI 辅助中风视觉注意力康复系统。知识体系梳理如下：

🧠 神经科学层面：

• 中风后单侧忽略的神经机制（VAN 损伤导致的注意力不对称）
• EEG 生物标记物与注意力状态的科学映射关系（P300、Alpha 不对称性等）
• 为什么多模态信号融合比单一信号更可靠（互补性与冗余性）

⚙️ 工程实现层面：

• StrokeEEGPreprocessor：保守主义的 EEG 处理策略，保护 ERP 成分
• StrokeRehabDetector：YOLOv11 扩展，实现"应注意 vs 实际注意"的比对
• EEGNet：轻量级脑电分类网络，适合边缘设备部署
• MultimodalFusionSystem：加权贝叶斯融合，量化综合忽略评分
• AdaptiveTrainingScheduler：基于评估结果的闭环训练调度
• RehabilitationProgressTracker：纵向进展追踪与临床报告

📊 临床价值层面：

• 将主观评估（量表）替换为客观量化（神经 + 行为双模态）
• 实现在真实视觉场景中的生态效度评估
• 自适应训练确保训练难度始终保持在患者的"最近发展区"

🔮 下期预告：脑机游戏交互：YOLOv11 检测驱动 VR 意念操作

在完成了严肃的临床康复应用之后，第13节我们将进入一个更具娱乐性和趣味性的应用领域——脑机游戏交互！但这绝非简单的游戏，而是将 BCI 技术与沉浸式 VR 环境深度融合的前沿探索。

下期主要内容预告

🎮 核心主题：YOLOv11 检测驱动 VR 意念操作

1. VR 与 BCI 融合架构
深入解析如何将 YOLOv11 的实时目标检测结果作为 VR 世界的"触发器"——当 YOLOv11 识别到用户正在注视某个 VR 物体时，结合 EEG 的意念信号，实现真正的"意念驱动"交互。我们将使用 Unity + XR Interaction Toolkit 构建 VR 环境，通过 Python Socket 实现与 BCI 模块的实时通信。

2. 意念操作指令编码
设计一套四选一的意念指令体系（前进/后退/左转/右转），基于 **SSVEP（稳态视觉诱发电位）**实现，用户只需凝视屏幕上特定频率闪烁的目标，大脑就会产生对应频率的神经振荡，YOLOv11 负责确认凝视目标的位置和类别。分类准确率目标：>80%（在线实时）。

3. VR 康复游戏化设计
将本节的中风康复训练"游戏化"——设计一个 VR 厨房场景，患者需要通过意念操作来"烹饪"食物，过程中 YOLOv11 实时检测食材的位置，BCI 系统监测患者的注意力分配，将康复训练融入有趣的游戏任务中，显著提升患者依从性。

4. 延迟优化与预测补偿
VR 对系统延迟极度敏感（>20ms 的延迟就会引发眩晕）。我们将探讨预测性运动补偿技术——利用 LSTM 预测用户头部运动轨迹，提前渲染下一帧，将感知延迟从 40ms 降低到 12ms 以内。

5. 意念操控的安全边界
BCI 游戏中最关键的安全问题：如何防止误操作？我们将设计多重确认机制（眼动停留时间 > 1.5s + EEG 意念强度 > 阈值 + 手部确认），确保意念指令的可靠性，避免患者因误操作产生挫败感或安全风险。

下期代码预览：

# 下期将实现的核心类（节选）
class VRBCIController:
    def __init__(self, vr_server_addr, bci_model):
        self.yolo_gaze_tracker = GazeAwareYOLO()
        self.ssvep_classifier = SSVEPClassifier(freqs=[6, 8, 10, 12])  # 4个指令频率
        self.vr_command_sender = VRCommandSocket(vr_server_addr)
        
    def process_frame(self, eeg_data, video_frame):
        # YOLOv11 检测凝视目标
        gaze_target = self.yolo_gaze_tracker.detect_gaze_object(video_frame)
        # SSVEP 解码意念指令
        command = self.ssvep_classifier.decode(eeg_data)
        # 发送到 VR 引擎
        self.vr_command_sender.send(command, gaze_target)

下期内容将彻底打破你对脑机接口只能用于医疗的刻板印象，展示 BCI + YOLOv11 在沉浸式娱乐和游戏化康复中的无限可能！

最后，希望本文围绕 YOLOv11 的实战讲解，能在以下几个方面对你有所帮助：

• 🎯 模型精度提升：通过结构改进、损失函数优化、数据增强策略等方案，尽可能提升检测效果与任务表现；
• 🚀 推理速度优化：结合量化、裁剪、蒸馏、部署加速等手段，帮助模型在实际业务场景中跑得更快、更稳；
• 🧩 工程级落地实践：从训练、验证、调参到部署优化，提供可直接复用或稍作修改即可迁移的完整思路与方案。

PS：如果你按文中步骤对 YOLOv11 进行优化后，仍然遇到问题，请不必焦虑或灰心。
YOLOv11 作为新一代目标检测模型，最终效果往往会受到 硬件环境、数据集质量、任务定义、训练配置、部署平台 等多重因素共同影响，因此不同任务之间的最优方案也并不完全相同。
如果你在实践过程中遇到：

• 新的报错 / Bug
• 精度难以提升
• 推理速度不达预期
  欢迎把 报错信息 + 关键配置截图 / 代码片段 粘贴到评论区，我们可以一起分析原因、定位瓶颈，并讨论更可行的优化方向。
  同时，如果你有更优的调参经验、结构改进思路，或者在实际项目中验证过更有效的方案，也非常欢迎分享出来，大家互相启发、共同完善 YOLOv11 的实战打法 🙌
• 当然，部分章节还会结合国内外前沿论文与 AIGC 大模型技术，对主流改进方案进行重构与再设计，内容更贴近真实工程场景，适合有落地需求的开发者深入学习与对标优化。

🧧🧧 文末福利，等你来拿！🧧🧧

文中涉及的多数技术问题，来源于我在 YOLOv11 项目中的一线实践，部分案例也来自网络与读者反馈；如有版权相关问题，欢迎第一时间联系，我会尽快处理（修改或下线）。
  部分思路与排查路径参考了全网技术社区与人工智能问答平台，在此也一并致谢。如果这些内容尚未完全解决你的问题，还请多一点理解——YOLOv11 的优化本身就是一个高度依赖场景与数据的工程问题，不存在“一招通杀”的方案。
  如果你已经在自己的任务中摸索出更高效、更稳定的优化路径，非常鼓励你：

• 在评论区简要分享你的关键思路；
• 或者整理成教程 / 系列文章。
  你的经验，可能正好就是其他开发者卡关许久所缺的那一环 💡

OK，本期关于 YOLOv11 优化与实战应用 的内容就先聊到这里。如果你还想进一步深入：

• 了解更多结构改进与训练技巧；
• 对比不同场景下的部署与加速策略；
• 系统构建一套属于自己的 YOLOv11 调优方法论；
  欢迎继续查看专栏：《YOLOv11实战：从入门到深度优化》。
  也期待这些内容，能在你的项目中真正落地见效，帮你少踩坑、多提效，下期再见 👋

码字不易，如果这篇文章对你有所启发或帮助，欢迎给我来个 一键三连（关注 + 点赞 + 收藏），这是我持续输出高质量内容的核心动力 💪

同时也推荐关注我的技术号 「猿圈奇妙屋」：

• 第一时间获取 YOLOv11 / 目标检测 / 多任务学习 等方向的进阶内容；
• 不定期分享与视觉算法、深度学习相关的最新优化方案与工程实战经验；
• 以及 BAT 等大厂面试题、技术书籍 PDF、工程模板与工具清单等实用资源。
  期待在更多维度上和你一起进步，共同提升算法与工程能力 🔧🧠

🫵 Who am I?

我是专注于 计算机视觉 / 图像识别 / 深度学习工程落地 的讲师 & 技术博主，笔名 bug菌：

• 热活于 CSDN | 稀土掘金 | InfoQ | 51CTO | 华为云开发者社区 | 阿里云开发者社区 | 腾讯云开发者社区 | 开源中国 | 博客园 | 墨天轮 等各大技术社区；
• CSDN 博客之星 Top30、华为云多年度十佳博主&卓越贡献奖、掘金多年度人气作者 Top40；
• CSDN、掘金、InfoQ、51CTO 等平台签约及优质作者；
• 全网粉丝累计 30w+。

更多高质量技术内容及成长资料，可查看这个合集入口 👉 点击查看 👈️

硬核技术号 「猿圈奇妙屋」 期待你的加入，一起进阶、一起打怪升级。

- End -