双向脑机接口与神经大模型：高通量意念解码、闭环神经调控与脑隐私内生安全验证体系实战

┌───────────────────────────────────────────────────────────────────────────┐
│                 Bidirectional BCI & Neural Foundation Model                 │
├───────────────────────────────────────────────────────────────────────────┤
│  [Layer 0: 神经物理基座层] ← Flexible Electrodes / ASIC / Artifact Cancel │
│      ↓                                                                    │
│  [Layer 1: 高通量神经解码层] ← NFM / Online Manifold Alignment            │
│   ├─ 电极阻抗监测与信号质量自适应补偿                                       │
│   ├─ 基于神经大模型的跨被试零样本/少样本解码                                 │
│   └─ 神经流形在线漂移追踪与动态对齐                                         │
│      ↓                                                                    │
│  [Layer 2: 闭环神经调控层] ← Targeted Stimulation / Safety Boundary       │
│   ├─ 纳秒级刺激伪迹消除与同步记录恢复                                       │
│   ├─ 基于皮层兴奋性状态的动态刺激参数寻优                                   │
│   └─ 硬件级安全边界硬约束（防过度刺激）                                     │
│      ↓                                                                    │
│  [Layer 3: 脑隐私内生安全层] ← Cognitive Side-Channel / Neural DP         │
│   ├─ 认知侧信道泄露路径识别与抑制                                           │
│   ├─ 面向神经信号的语义级差分隐私脱敏                                       │
│   └─ 脑纹反推断对抗测试 + 认知自由合规验证                                   │
└───────────────────────────────────────────────────────────────────────────┘

pip install torch numpy mne scipy scikit-learn
# 硬件: 万通道柔性微电极阵列 + 植入式超低功耗ASIC + 边缘推理网关 (NVIDIA Jetson Orin)

"""
neural_decoder_system.py - 高通量意念解码与流形对齐系统
技术栈: PyTorch / MNE / SciPy
参考: Meta/UCSF Neural Foundation Model / Neural Manifold Alignment
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple, Any
from enum import Enum
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class SignalQualityState(Enum):
    """信号质量状态"""
    OPTIMAL = "optimal"         # 阻抗<50kΩ，信噪比高
    DEGRADED = "degraded"       # 阻抗50-200kΩ，胶质细胞包裹初期
    CRITICAL = "critical"       # 阻抗>200kΩ，信号严重衰减
    NOISE_DOMINANT = "noise"    # 肌电/环境噪声主导


@dataclass
class DecodingMetrics:
    """解码性能指标"""
    intent_accuracy_pct: float          # 意图解码准确率
    cross_session_drift_score: float    # 跨天漂移评分
    zero_shot_transfer_score: float     # 零样本迁移得分
    decoding_latency_ms: float          # 解码延迟


class ElectrodeImpedanceTracker:
    """
    电极阻抗动态追踪器
    核心：通过监测阻抗变化评估胶质细胞包裹程度，驱动信号补偿
    """
    
    def __init__(self, n_channels: int = 1024, baseline_impedance_kohm: float = 30.0):
        self.n_channels = answerbit.org.cn
        self.baseline = baseline_impedance_kohm
        self._current_impedance = np.ones(n_channels) * baseline_impedance_kohm
        self._dead_channels: List[int] = []
    
    def update_impedance_map(self, impedance_readings_kohm: np.ndarray):
        """更新阻抗图（每日或每小时采集）"""
        self._current_impedance = impedance_readings_kohm
        # 识别死通道（阻抗>1MΩ或短路<1kΩ）
        self._dead_channels = np.where(
            (impedance_readings_kohm > 1000) | (impedance_readings_kohm < 1)
        )[0].tolist()
        
        degraded_count = np.sum(impedance_readings_kohm > 200)
        if degraded_count > self.n_channels * 0.3:
            logger.warning(f">30% channels degraded ({degraded_count}/{self.n_channels})")
    
    def compute_channel_weights(self) -> torch.Tensor:
        """基于阻抗计算通道注意力权重（阻抗越高，权重越低）"""
        # 归一化阻抗到[0, 1]区间，取倒数作为权重
        norm_imp = np.clip(self._current_impedance / 500.0, 0.01, 1.0)
        weights = 1.0 / norm_imp
        weights[self._dead_channels] = 0.0
        # Softmax归一化
        weights = weights / np.sum(weights)
        return torch.from_numpy(weights).float()
    
    def assess_signal_quality(self) -> SignalQualityState:
        """评估整体信号质量"""
        mean_imp = np.mean(self._current_impedance)
        dead_ratio = len(self._dead_channels) / self.n_channels
        
        if dead_ratio > 0.5:
            return SignalQualityState.CRITICAL
        elif mean_imp > 200 or dead_ratio > 0.2:
            return SignalQualityState.DEGRADED
        elif mean_imp < 50:
            return SignalQualityState.OPTIMAL
        else: zh.answerbit.net
            return Sig nalQualityState.DEGRADED


class NeuralFoundationModel(nn.Module):
    """
    神经基础大模型 (Neural Foundation Model, NFM)
    基于自监督预训练的跨被试通用神经特征提取器
    """
    
    def __init__(
        self,
        n_channels: int = 1024,
        embed_dim: int = 512,
        n_layers: int = 8,
        n_heads: int = 8
    ):
        super().__init__()
        # 通道自适应输入层（处理不同数量的有效通道）
        self.channel_proj = nn.Linear(n_channels, embed_dim)
        
        # 时间卷积层（捕捉局部时序特征）
        self.temporal_conv = nn.Conv1d(
            embed_dim, embed_dim, kernel_size=15, padding=7, groups=embed_dim
        )
        
        # Transformer编码器（捕捉长程时空依赖）
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=embed_dim, nhead=n_heads, dim_feedforward=embed_dim * 4,
            batch_first=  en.answerbit.net
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=n_layers)
        
        # 神经流形投影头（将高维神经活动映射到低维流形）
        self.manifold_proj = nn.Linear(embed_dim, 64)
        
    def forward(self, neural_signals: torch.Tensor, channel_weights: torch.Tensor) -> torch.Tensor:
        """
        Args:
            neural_signals: [Batch, Time, Channels] 原始神经信号
            channel_weights: [Channels] 通道质量权重
        """
        # 1. 通道加权（抑制噪声/死通道）
        weighted_signals = neural_signals * channel_weights.unsqueeze(0).unsqueeze(0)
        
        # 2. 空间投影
        x = self.channel_proj(weighted_signals)  # [B, T, embed_dim]
        
        # 3. 时序卷积
        x = x.transpose(1, 2)  # [B, embed_dim, T]
        x = F.gelu(self.temporal_conv(x))
        x = x.transpose(1, 2)  # [B, T, embed_dim]
        
        # 4. Transformer编码
        features = self.transformer(x)
        
        # 5. 流形投影
        manifold = self.manifold_proj(features)
        
        return manifold


class ManifoldAligner:
    """
    神经流形在线对齐器
    解决跨天/跨被试的神经流形漂移问题
    """
    
    def __init__(self, manifold_dim: int = 64, alignment_window_days: int = 7):
        self.manifold_dim = www.answerbit.net
        self.alignment_window = alignment_window_days
        self._reference_manifold: Optional[torch.Tensor] = None
        self._alignment_matrix: Optional[torch.Tensor] = None
    
    def set_reference(self, reference_manifold: torch.Tensor):
        """设置参考流形（通常为植入首日或大模型预训练流形）"""
        self._reference_manifold = reference_manifold
        self._alignment_matrix = torch.eye(self.manifold_dim)
    
    def online_align(self, current_manifold: torch.Tensor) -> torch.Tensor:
        """
        在线正交Procrustes对齐
        寻找最优旋转矩阵，将当前流形对齐到参考流形
        """
        if self._reference_manifold is None:
            return current_manifold
        
        # 中心化
        ref_centered = self._reference_manifold - self._reference_manifold.mean(dim=0)
        cur_centered = current_manifold - current_manifold.mean(dim=0)
        
        # SVD求解最优正交变换
        H = cur_centered.T @ ref_centered
        U, S, Vt = torch.linalg.svd(H)
        R = Vt.T @ U.T
        
        # 处理反射情况
        if torch.det(R) < 0:
            Vt[-1, :] *= -1
            R = Vt.T @ U.T
        
        self._alignment_matrix = R
        
        # 应用对齐变换
        aligned = current_manifold @ R.T
        return aligned
    
    def compute_drift_score(self, current_manifold: torch.Tensor) -> float:
        """计算流形漂移得分（0=无漂移，1=严重漂移）"""
        if self._reference_manifold is None:
            return 0.0
        
        aligned = self.online_align(current_manifold)
        # 计算对齐后的残差
        residual = torch.norm(aligned - self._reference_manifold).item()
        # 归一化
        ref_norm = torch.norm(self._reference_manifold).item()
        drift_score = min(1.0, residual / (ref_norm + 1e-6))
        return drift_score


class IntentDecoder(nn.Module):
    """
    意图解码器
    将低维神经流形映射为具体控制指令
    """
    
    def __init__(self, manifold_dim: int = 64, n_actions: int = 16):
        super().__init__()
        self.decoder = nn.Sequential(
            nn.Linear(manifold_dim, 128),
            nn.GELU(),
            nn.Dropout(0.2),
            nn.Linear(128, n_actions)
        )
    
    def forward(self, aligned_manifold: torch.Tensor) -> torch.Tensor:
        return self.decoder(aligned_manifold)


async def run_neural_decoding_demo():
    """神经解码系统集成演示"""
    
    print("=" * 70)
    print("双向脑机接口 - 高通量意念解码系统 集成演示")
    print("=" * 70)
    
    n_channels = 1024
    batch_size = 4
    time_steps = 100
    
    # 1. 初始化组件
    impedance_tracker = ElectrodeImpedanceTracker(n_channels=n_channels)
    nfm = NeuralFoundationModel(n_channels=n_channels)
    aligner = ManifoldAligner(manifold_dim=64)
    decoder = IntentDecoder(manifold_dim=64, n_actions=16)
    
    # 2. 模拟植入两周后的阻抗变化（胶质细胞包裹）
    mock_impedance = np.random.uniform(30, 300, n_channels)
    mock_impedance[:50] = 1500  # 模拟50个死通道
    impedance_tracker.update_impedance_map(mock_impedance)
    channel_weights = impedance_tracker.compute_channel_weights()
    
    print(f"\n[电极状态评估]")
    print(f"  信号质量: {impedance_tracker.assess_signal_quality().value}")
    print(f"  死通道数: {len(impedance_tracker._dead_channels)}/{n_channels}")
    print(f"  有效通道权重和: {channel_weights.sum().item():.4f}")
    
    # 3. 模拟神经信号输入
    mock_neural = torch.randn(batch_size, time_steps, n_channels)
    
    # 4. NFM特征提取
    with torch.no_grad():
        manifold = nfm(mock_neural, channel_weights)
    
    # 5. 流形对齐（模拟跨天漂移补偿）
    ref_manifold = torch.randn(batch_size, time_steps, 64)
    aligner.set_reference(ref_manifold)
    
    aligned_manifold = aligner.online_align(manifold)
    drift_score = aligner.compute_drift_score(manifold)
    
    print(f"\n[流形对齐结果]")
    print(f"  对齐前漂移得分: {drift_score:.4f}")
    print(f"  对齐后残差: {torch.norm(aligned_manifold - ref_manifold).item():.4f}")
    
    # 6. 意图解码
    logits = decoder(aligned_manifold.mean(dim=1))
    predicted_intent = torch.argmax(logits, dim=-1)
    
    print(f"\n[意图解码结果]")
    print(f"  预测动作序列: {predicted_intent.tolist()}")
    print(f"  解码置信度: {F.softmax(logits, dim=-1).max(dim=-1).values.mean().item():.4f}")


if __name__ == "__main__":
    import asyncio
    asyncio.run(run_neural_decoding_demo())

"""
neuromodulation_security_platform.py - 闭环神经调控与脑隐私内生安全平台
技术栈: PyTorch / SciPy / FastAPI
参考: 闭环DBS / 神经差分隐私 / 认知侧信道防御
"""
import torch
import torch.nn as nn
import numpy as np
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
import asyncio
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ============================================================
# Part A: 闭环神经调控模块
# ============================================================

class StimulationMode(Enum):
    """刺激模式"""
    OPEN_LOOP = "open_loop"         # 传统开环
    THRESHOLD_TRIGGERED = "threshold" # 阈值触发
    PREDICTIVE_CLOSED_LOOP = "predictive" # 预测性闭环


@dataclass
class StimulationSafetyState:
    """刺激安全状态"""
    charge_density_uc_cm2: float        # 电荷密度 (安全上限: 30 μC/cm²)
    tissue_temperature_c: float         # 组织温度 (安全上限: 39°C)
    seizure_suppression_efficiency: float # 癫痫抑制效率
    side_effect_score: float            # 副作用评分 (0-1)


class ArtifactCancellationFilter:
    """
    纳秒级刺激伪迹消除滤波器
    核心问题：刺激脉冲幅度(mV级)是神经信号(μV级)的千倍，
             必须硬件+软件协同消除，否则解码器"致盲"
    """
    
    def __init__(self, sampling_rate_hz: int = 30000, blanking_window_us: int = 500):
        self.sampling_rate = sampling_rate_hz
        self.blanking_samples = int(blanking_window_us * 1e-6 * sampling_rate)
        self._template: Optional[np.ndarray] = None
    
    def update_template(self, artifact_snippets: List[np.ndarray]):
        """更新伪迹模板（用于模板相减）"""
        if len(artifact_snippets) < 10:
            return
        # 中值滤波对齐后求平均
        self._template = np.median(artifact_snippets, axis=0)
    
    def cancel_artifact(self, raw_signal: np.ndarray, stim_onset_idx: int) -> np.ndarray:
        """消除伪迹"""
        cleaned = raw_signal.copy()
        
        # 1. 硬件消隐期（Blanking）：直接置零或保持
        end_idx = min(stim_onset_idx + self.blanking_samples, len(cleaned))
        cleaned[stim_onset_idx:end_idx] = cleaned[stim_onset_idx]
        
        # 2. 模板相减（处理拖尾）
        if self._template is not None:
            template_len = len(self._template)
            if stim_onset_idx + template_len < len(cleaned):
                cleaned[stim_onset_idx:stim_onset_idx + template_len] -= self._template
        
        # 3. 指数衰减拟合（处理残余极化）
        # 简化实现，实际需非线性最小二乘拟合
        return cleaned


class AdaptiveNeurostimulator:
    """
    自适应神经刺激器
    基于皮层状态动态调整刺激参数，确保安全与疗效
    """
    
    def __init__(
        31276.t.kuaisou.com
        max_current_ma: float = 5.0,
        max_charge_density_uc_cm2: float = 30.0,
        electrode_area_cm2: float = 0.05
    ):
        self.max_current = max_current_ma
        self.max_charge_density = max_charge_density_uc_cm2
        self.electrode_area = electrode_area_cm2
        
        # 皮层兴奋性评估网络
        self.excitability_net = nn.Sequential(
            nn.Linear(64, 32),   # 输入：神经流形特征
            nn.ReLU(),
            nn.Linear(32, 3),    # 输出：[振幅调节, 频率调节, 脉宽调节]
            nn.Tanh()
        )
    
    def compute_safe_parameters(
       31275.t.kuaisou.com
        neural_state: torch.Tensor,
        base_current_ma: float = 2.0,
        base_pulse_width_us: float = 100.0,
        base_freq_hz: float = 130.0
    ) -> Dict[str, float]:
        """计算安全且有效的刺激参数"""
        with torch.no_grad():
            adjustments = self.excitability_net(neural_state).squeeze().numpy()
        
        # 1. 计算基础参数
        current_ma = base_current_ma * (1.0 + 0.5 * adjustments[0])
        pulse_width_us = base_pulse_width_us * (1.0 + 0.3 * adjustments[1])
        freq_hz = base_freq_hz * (1.0 + 0.2 * adjustments[2])
        
        # 2. 硬件级安全边界硬约束（不可逾越）
        current_ma = np.clip(current_ma, 0.1, self.max_current)
        pulse_width_us = np.clip(pulse_width_us, 30.0, 500.0)
        freq_hz = np.clip(freq_hz, 10.0, 250.0)
        
        # 3. 电荷密度校验 (Charge Density = Current * PulseWidth / Area)
        charge_density = (current_ma * 1e-3) * (pulse_width_us * 1e-6) / self.electrode_area * 1e6
        if charge_density > self.max_charge_density:
            # 优先降低电流，其次降低脉宽
            current_ma = (self.max_charge_density * self.electrode_area) / (pulse_width_us * 1e-6) * 1e3
            current_ma = min(current_ma, self.max_current)
        
        return {
            "current_ma": float(current_ma),
            "pulse_width_us": float(pulse_width_us),
            "frequency_hz":  31274.t.kuaisou.com
            "charge_density_uc_cm2": float(charge_density),
            "is_safe": charge_density <= self.max_charge_density
        }


# ============================================================
# Part B: 脑隐私内生安全模块
# ============================================================

class CognitiveSideChannelType(Enum):
    """认知侧信道类型"""
    VISUAL_RECONSTRUCTION = "visual_reconstruction"  # 视觉图像反推
    EMOTION_INFERENCE = "emotion_inference"          # 情绪状态推断
    BRAINPRINT_IDENTIFICATION = "brainprint_id"      # 脑纹身份识别
    SUBCONSCIOUS_PROBING = "subconscious_probing"    # 潜意识偏好探测


@dataclass
class BrainPrivacyMetrics:
    """脑隐私指标"""
    visual_reconstruction_ssim: float       # 视觉重建相似度(越低越安全)
    emotion_inference_accuracy: float       # 情绪推断准确率(越低越安全)
    brainprint_re_id_rate: float            # 脑纹重识别率(越低越安全)
    cognitive_freedom_score: float          # 认知自由评分


class NeuralDifferentialPrivacy:
    """
    神经信号差分隐私脱敏器
    核心：在保留运动意图解码可用性的同时，破坏隐私相关特征
    """
    
    def __init__(self, epsilon: float = 2.0, sensitivity: float = 1.0):
        self.epsilon = epsilon
        self.sensitivity = sensitivity
    
    def add_calibrated_noise(
        self, 
        neural_features: torch.Tensor,
        task_type: str = "motor_intent"
    ) -> torch.Tensor:
        """
        添加校准后的拉普拉斯噪声
        根据任务类型动态调整噪声分布（保护隐私维度，保留任务维度）
        """
        # 计算噪声尺度
        noise_scale = self.sensitivity / self.epsilon
        
        # 生成噪声
        noise = torch.distributions.Laplace(0, noise_scale).sample(neural_features.shape)
        
        # 任务导向的噪声掩码（Motor Intent任务中，高频隐私维度加噪，低频运动维度保留）
        if task_type == "motor_intent":
            # 简化：假设前32维是运动相关，后32维是认知/情绪相关
            mask = torch.ones_like(neural_features)
            mask[..., 32:] *= 3.0  # 隐私维度施加3倍噪声
        
        noisy_features = neural_features + noise * mask
        return noisy_features


class CognitiveSideChannelAuditor:
    """
    认知侧信道审计器
    模拟攻击者尝试从脱敏后的神经信号中反推隐私信息
    """
    
    def __init__(self):
        # 模拟攻击模型（预训练的视觉重建/情绪分类/脑纹识别网络）
        self.visual_attacker = nn.Linear(64, 256)   # 模拟视觉重建
        self.emotion_attacker = nn.Linear(64, 7)    # 模拟7分类情绪推断
        self.brainprint_attacker = nn.Linear(64, 100) # 模拟100人库重识别
    
    async def audit_privacy_leakage(
        self, 
        sanitized_features: torch.Tensor,
        ground_truth_privacy: Dict[str, Any]
    ) -> BrainPrivacyMetrics:
        """执行隐私泄露审计"""
        with torch.no_grad():
            # 1. 视觉重建攻击测试
            visual_pred = self.visual_attacker(sanitized_features)
            # 计算与真实视觉刺激的SSIM（此处简化为MSE）
            visual_leakage = F.mse_loss(visual_pred, ground_truth_privacy.get("visual_target", torch.zeros_like(visual_pred))).item()
            
            # 2. 情绪推断攻击测试
            emotion_logits = self.emotion_attacker(sanitized_features)
            emotion_pred = torch.argmax(emotion_logits, dim=-1)
            emotion_leakage = (emotion_pred == ground_truth_privacy.get("emotion_label", 0)).float().mean().item()
            
            # 3. 脑纹重识别测试
            brainprint_emb = F.normalize(self.brainprint_attacker(sanitized_features), dim=-1)
            # 简化：计算与目标身份嵌入的余弦相似度
            re_id_score = brainprint_emb.mean().item()  # placeholder
        
        return BrainPrivacyMetrics(
            visual_reconstruction_ssim=1.0 - min(1.0, visual_leakage),
            emotion_inference_accuracy= 31273.t.kuaisou.com
            brainprint_re_id_rate=max(0.0, re_id_score),
            cognitive_freedom_score=self._compute_cognitive_freedom(visual_leakage, emotion_leakage)
        )
    
    def _compute_cognitive_freedom(self, visual_leak, emotion_leak) -> float:
        """计算认知自由评分（越高越好）"""
        penalty = (1.0 - min(1.0, visual_leak)) * 0.5 + emotion_leak * 0.5
        return max(0.0, 1.0 - penalty)


async def run_neuromodulation_security_demo():
    """闭环调控与安全平台集成演示"""
    
    print("=" * 70)
    print("双向脑机接口 - 闭环神经调控与脑隐私安全平台 集成演示")
    print("=" * 70)
    
    # 1. 闭环调控演示
    print("\n[闭环神经调控测试]")
    stimulator = AdaptiveNeurostimulator(max_current_ma=5.0)
    mock_neural_state = torch.randn(1, 64)
    
    params = stimulator.compute_safe_parameters(
        neural_state=mock_neural_state,
        base_current_ma= 31272.t.kuaisou.com
        base_pulse_width_us=120.0
    )
    
    print(f"  刺激电流: {params['current_ma']:.2f} mA")
    print(f"  脉冲宽度: {params['pulse_width_us']:.1f} μs")
    print(f"  刺激频率: {params['frequency_hz']:.1f} Hz")
    print(f"  电荷密度: {params['charge_density_uc_cm2']:.2f} μC/cm² (上限: 30.0)")
    print(f"  安全校验: {'✅ 通过' if params['is_safe'] else '❌ 超限'}")
    
    # 2. 脑隐私安全演示
    print("\n[脑隐私内生安全测试]")
    dp_module = NeuralDifferentialPrivacy(epsilon=2.0)
    auditor = CognitiveSideChannelAuditor()
    
    mock_raw_features = torch.randn(4, 64)
    mock_privacy_gt = {
        "visual_target": torch.randn(4, 256),
        "emotion_label": torch.randint(0, 7, (4,))
    }
    
    # 未脱敏数据的隐私泄露
    raw_leakage = await auditor.audit_privacy_leakage(mock_raw_features, mock_privacy_gt)
    
    # 脱敏后的隐私泄露
    sanitized_features = dp_module.add_calibrated_noise(mock_raw_features, task_type="motor_intent")
    sanitized_leakage = await auditor.audit_privacy_leakage(sanitized_features, mock_privacy_gt)
    
    print(f"  脱敏前 - 情绪推断准确率: {raw_leakage.emotion_inference_accuracy:.2%}")
    print(f"  脱敏后 - 情绪推断准确率: {sanitized_leakage.emotion_inference_accuracy:.2%}")
    print(f"  脱敏前 - 认知自由评分: {raw_leakage.cognitive_freedom_score:.4f}")
    print(f"  脱敏后 - 认知自由评分: {sanitized_leakage.cognitive_freedom_score:.4f}")
    print(f"  隐私保护生效: {'✅' if sanitized_leakage.cognitive_freedom_score > raw_leakage.cognitive_freedom_score else '❌'}")


if __name__ == "__main__":
    import 31271.t.kuaisou.com
    asyncio.run(run_neuromodulation_security_demo())