graph LR
    A[原始视频流 30fps] --> B[人脸检测与对齐 MTCNN/RetinaFace]
    B --> C[动作单元强度回归 AU Intensity Regression]
    C --> D[时序平滑与去抖动 Kalman/One-Euro Filter]
    D --> E[微表情候选片段提取 时窗 0.2-0.5s]
    E --> F[时序分类器 BiLSTM / Temporal ConvNet / Transformer]
    F --> G[疲劳相关 ME 序列编码]
    G --> H[疲劳等级回归/分类输出]

class CognitiveLoadFusion(nn.Module):
    def __init__(self, d_model=256, n_heads=4, n_layers=3):
        super().__init__()
        # 模态专用编码器
        self.face_enc   = TemporalEncoder(input_dim=17, d_model=d_model)  # AU强度序列
        self.gaze_enc   = TemporalEncoder(input_dim=6,  d_model=d_model)  # 眼动特征
        self.physio_enc = TemporalEncoder(input_dim=8,  d_model=d_model)  # HRV/EDA/呼吸
        self.behav_enc  = TemporalEncoder(input_dim=12, d_model=d_model)  # 交互日志
        
        # 跨模态注意力融合
        self.cross_attn = nn.ModuleList([
            CrossModalAttention(d_model, n_heads) for _ in range(n_layers)
        ])
        
        # 动态门控：根据模态可靠性自适应加权
        self.gate_net = nn.Sequential(
            nn.Linear(d_model * 4, 64), nn.ReLU(),
            nn.Linear(64, 4), nn.Softmax(dim=-1)
        )
        
        # 负荷回归头
        self.load_head = nn.Sequential(
            nn.Linear(d_model, 64), nn.ReLU(),
            nn.Linear(64, 1), nn.Sigmoid()  # 输出 0-1 归一化负荷
        )

    def forward(self, face, gaze, physio, behav, mask=None):
        # 编码
        f = self.face_enc(face)
        g = self.gaze_enc(gaze)
        p = self.physio_enc(physio)
        b = self.behav_enc(behav)
        
        # 跨模态交互
        mods = [f, g, p, b]
        for attn in self.cross_attn:
            mods = [attn(q, kv, mask) for q, kv in zip(mods, mods)]
        
        # 门控融合
        concat = torch.cat(mods, dim=-1)
        weights = self.gate_net(concat.mean(dim=1))  # [B, 4]
        fused = sum(w.unsqueeze(-1).unsqueeze(-1) * m for w, m in zip(weights.unbind(-1), mods))
        
        return self.load_head(fused.mean(dim=1)), weights

graph TB
    subgraph 客户端
        A[Webcam/耳机/手环] --> B[本地特征提取 ONNX Runtime]
        B --> C[仅上传脱敏特征向量 50Hz]
    end
    subgraph 边缘网关
        C --> D[流式特征聚合 Kafka/Redpanda]
        D --> E[实时推理服务 Triton Inference Server]
    end
    subgraph 控制平面
        E --> F[疲劳/负荷事件流]
        F --> G[自适应干预策略引擎]
        G --> H[会议客户端提醒/议程调整/休息建议]
    end
    subgraph 合规层
        I[GDPR/PIPL 合规审计] -.-> B
        J[联邦学习参数聚合] -.-> E
    end

# 伪代码：主动学习选样策略 (基于模型不确定性 + 样本多样性)
def select_samples(unlabeled_pool, model, budget=500):
    # 1. 不确定性采样: MC Dropout 前向传播 T=20 次, 计算预测熵
    entropies = []
    embeddings = []
    for x in unlabeled_pool:
        preds = [model(x, dropout=True) for _ in range(20)]
        mean_pred = np.mean(preds, axis=0)
        entropy = -np.sum(mean_pred * np.log(mean_pred + 1e-8))
        entropies.append(entropy)
        embeddings.append(model.get_embedding(x)) # 获取倒数第二层特征
    
    # 2. 核心集贪心选择 (Core-set) 保证多样性
    selected_idx = []
    remaining = set(range(len(unlabeled_pool)))
    # 先选熵最高的种子
    seed = np.argmax(entropies)
    selected_idx.append(seed)
    remaining.remove(seed)
    
    for _ in range(budget - 1):
        # 计算剩余样本到已选集合的最小距离
        dists = np.array([min(np.linalg.norm(embeddings[i] - embeddings[j]) 
                              for j in selected_idx) for i in remaining])
        # 综合得分: α * 归一化熵 + (1-α) * 归一化距离
        scores = 0.6 * normalize([entropies[i] for i in remaining]) + 
                 0.4 * normalize(dists)
        best = list(remaining)[np.argmax(scores)]
        selected_idx.append(best)
        remaining.remove(best)
    return [unlabeled_pool[i] for i in selected_idx]

# 合成数据生成: 条件扩散模型生成极端疲劳/负荷样本
# 条件 c = [疲劳等级, 光照类型, 姿态角, 种族属性]
# 训练目标: 扩充长尾分布 (高负荷样本 < 5%)

// 伪代码：Fused Dilated Residual Block Kernel (INT8)
__global__ void fused_dilated_residual_block_kernel(
    const int8_t* __restrict__ input,   // [C, T]
    const int8_t* __restrict__ w1, const int32_t* __restrict__ b1, // Conv1
    const float* __restrict__ bn1_scale, const float* __restrict__ bn1_bias,
    const int8_t* __restrict__ w2, const int32_t* __restrict__ b2, // Conv2
    const float* __restrict__ bn2_scale, const float* __restrict__ bn2_bias,
    const int8_t* __restrict__ residual, // 残差分支 (可能需 1x1 投影)
    int8_t* __restrict__ output,         // [C, T]
    int C, int T, int K, int D1, int D2, // 通道、时长、核大小、膨胀率
    float output_scale, int8_t output_zp  // 量化参数
) {
    // 1. 共享内存加载权重 (常驻)
    // 2. 寄存器分块计算: 单线程处理 4 个输出通道 x 8 个时间步
    // 3. 在线融合: Conv -> BN -> ReLU -> Conv -> BN -> Add -> ReLU -> Quantize
    // 4. 避免中间结果写回 Global Memory, 寄存器/共享内存直通
    // 5. 非对齐内存访问优化: 使用 ld.global.ca.v4.i8 / st.global.ca.v4.i8
}

graph LR
    subgraph Camera HAL
        A[YUV420 NV12 Buffer] -->|Zero-Copy dmabuf| B[ISP 预处理: 裁剪/归一化/转 RGB]
    end
    subgraph NPU Runtime (TFLite / MNN / ONNX Runtime)
        B --> C[Face Detect + Align]
        C --> D[AU Regressor HRNet-S]
        D --> E[TCN Micro-Exp Classifier]
        E --> F[Fatigue State Estimator Seq-VAE]
    end
    subgraph CPU (Main Thread)
        F --> G[特征向量打包 Protobuf]
        G --> H[gRPC/QUIC 发送至网关]
    end
    %% 异步双缓冲
    style A fill:#e1f5fe
    style B fill:#e1f5fe
    style C fill:#fff3e0
    style D fill:#fff3e0
    style E fill:#fff3e0
    style F fill:#fff3e0

graph TD
    U[未观测混杂因素n基线认知能力/睡眠质量] --> F[疲劳度 F_t]
    U --> L[认知负荷 L_t]
    C[会议内容复杂度] --> L
    C --> F
    S[发言时长/频次] --> L
    S --> F
    M[微表情频次] --> F
    P[生理指标 HRV/EDA] --> F
    L --> F
    I[干预: 休息/议程调整] --> F
    I --> L
    T[时间 t] --> F
    T --> L
    T --> M
    T --> P
    
    classDef latent fill:#fce4ec,stroke:#c82c6c;
    class U latent;

from causallearn.search.ConstraintBased.PC import pc
import numpy as np

# 数据: [F_t, L_t, M_t, P_t, S_t, C_t, I_t, T] 滞后 1-3 步特征
data = np.load('meeting_causal_data.npy') # [N, D]

# 领域知识约束: 时间只能指向未来, 干预 I 不能被 F/L 反向导致
forbidden_edges = {('F_t', 'T'), ('L_t', 'T'), ('I_t', 'F_t-1'), ('I_t', 'L_t-1')}
required_edges = {('T', 'F_t'), ('T', 'L_t'), ('C_t', 'L_t')}

cg = pc(data, alpha=0.01, indep_test='fisherz', 
        background_knowledge=(forbidden_edges, required_edges))
# 输出 CPDAG (完成部分有向无环图)

graph TB
    subgraph 物理实体层
        U[用户 U_i]
    end
    subgraph 数字孪生层
        DT[认知数字孪生 CDT_i]
        subgraph 核心状态空间
            S1[特质参数 θ_traitn基线觉醒/抑制控制/工作记忆容量]
            S2[状态参数 θ_state_tn当前疲劳/负荷/情绪/动机]
            S3[环境参数 θ_env_tn会议类型/时间/设备/社交压力]
        end
        subgraph 预测引擎
            E1[短期预测器n下 30 分钟疲劳轨迹]
            E2[长期演化器n周/月认知韧性趋势]
            E3[干预模拟器nWhat-if 仿真]
        end
    end
    subgraph 同步与更新机制
        SYNC[卡尔曼滤波/粒子滤波n多模态观测融合更新 θ_state]
        META[元学习器n新会话快速适应 θ_trait]
        CL[持续学习n防止灾难性遗忘]
    end
    U -->|多模态流| SYNC
    SYNC --> DT
    DT -->|决策支持| U
    META -.->|冷启动| S1
    CL -.->|长期漂移| S1

# 伪代码：推理前置守卫
def inference_guard(video_stream):
    # 1. 活体检测 (Silent Live Detection)
    #    - 频域分析: 真实视频高频噪声特征 vs 合成视频平滑频谱
    #    - 微动一致性: 头部微动/眼动/血流信号 (rPPG) 跨模态相位锁定
    liveness_score = silent_liveness_detector(video_stream) 
    if liveness_score < 0.85: 
        raise SecurityException("SPOOF_DETECTED", detail="Virtual camera or replay attack")
    
    # 2. 对抗净化
    #    - 随机化平滑: 输入加高斯噪声 σ=0.05, 集成 10 次预测取众数
    #    - 扩散模型去噪: 预训练 DDPM 将输入投影回正常数据流形
    purified_frames = diffusion_purifier(video_stream, steps=5) # 轻量级蒸馏版
    
    return purified_frames

# 伪代码：Trimmed Mean 聚合 (剔除极值梯度)
def trimmed_mean_aggregation(client_grads, trim_ratio=0.2):
    # client_grads: List[Dict[param_name, Tensor]]
    stacked = {k: torch.stack([g[k] for g in client_grads]) for k in client_grads[0]}
    trimmed = {}
    k_trim = int(len(client_grads) * trim_ratio)
    for k, tensors in stacked.items():
        sorted_vals, _ = torch.sort(tensors, dim=0)
        trimmed[k] = sorted_vals[k_trim:-k_trim].mean(dim=0)
    return trimmed

// fatigue_telemetry.proto
syntax = "proto3";

package meeting.intelligence.v1;

message FatigueTelemetry {
  string session_id = 1;           // 会议唯一标识 (UUID v7)
  string user_id_hash = 2;         // 用户哈希 ID (HMAC-SHA256, salt 轮换)
  int64 timestamp_ms = 3;          // 事件时间戳 (Unix ms, UTC)
  
  // 核心指标 (0.0 ~ 1.0 归一化)
  float fatigue_level = 4;         // 综合疲劳度
  float cognitive_load = 5;        // 认知负荷
  float arousal_level = 6;         // 觉醒水平
  float engagement_index = 7;      // 投入度指数
  
  // 置信度与不确定性
  float epistemic_uncertainty = 8; // 认知不确定性 (模型不知)
  float aleatoric_uncertainty = 9; // 偶然不确定性 (数据噪声)
  
  // 关键驱动因素解释
  repeated FeatureAttribution top_drivers = 10; // SHAP 值 Top-K
  
  // 元数据
  DeviceContext device_context = 11;
  ModelVersion model_version = 12;
  ProcessingLatencyMs latency_breakdown = 13;
}

message FeatureAttribution {
  string feature_name = 1;  // e.g., "AU4_intensity", "HRV_LF_HF", "gaze_entropy"
  float shap_value = 2;     // 对疲劳正/负贡献
  float raw_value = 3;      // 原始特征值
  string unit = 4;          // 单位
}

message DeviceContext {
  string os = 1;
  string cpu_arch = 2;
  string npu_vendor = 3;     // "Qualcomm", "Apple", "Intel", "None"
  bool has_ir_camera = 4;
  bool has_wearable = 5;
}

message ModelVersion {
  string model_id = 1;       // e.g., "fatigue_seqvae_tcn_v3.2.1"
  string git_commit = 2;
  int64 build_timestamp = 3;
}

# OpenAPI 3.1 片段
paths:
  /v1/intervention/recommend:
    post:
      summary: 请求干预建议
      security: [{BearerAuth: []}]
      requestBody:
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/InterventionRequest'
      responses:
        '200':
          description: 干预决策
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/InterventionResponse'
        '429':
          $ref: '#/components/responses/RateLimited'

components:
  schemas:
    InterventionRequest:
      type: object
      required: [session_id, user_id_hash, current_telemetry, meeting_context]
      properties:
        session_id: {type: string, format: uuid}
        user_id_hash: {type: string}
        current_telemetry: {$ref: '#/components/schemas/FatigueTelemetry'}
        meeting_context:
          type: object
          properties:
            agenda_items: 
              type: array
              items: {type: string}
            remaining_duration_min: {type: integer}
            participant_count: {type: integer}
            meeting_type: {type: string, enum: [DECISION, SYNC, BRAINSTORM, TRAINING, ONE_ON_ONE]}
            is_host: {type: boolean}
    
    InterventionResponse:
      type: object
      required: [action, priority, reasoning, ttl_seconds]
      properties:
        action:
          type: string
          enum: [NONE, MICRO_BREAK_SUGGEST, AGENDA_COMPRESS, TOPIC_SWITCH, FORCED_PAUSE, MEETING_WRAP_UP, HEALTH_ALERT]
        priority: {type: integer, minimum: 0, maximum: 100}
        reasoning: {type: string} # 自然语言解释, 供 UI 展示
        ttl_seconds: {type: integer} # 建议有效期
        config:
          type: object
          properties:
            break_duration_min: {type: integer}
            suggested_activity: {type: string, enum: [STRETCH, HYDRATE, BREATHING, WALK, EYE_EXERCISE]}
            agenda_adjustment: {type: string} # 具体议程调整建议