文档编号：CV-EDGE-PRD-2026-001 版本：v3.0 (Environmental Resilience + Alert Quality Management) 日期：2026-06-20 作者：资深计算机视觉系统架构师 状态：Draft（10 项关键缺陷修正）

• v3.0 (2026-06-20): 新增 Section 13 & 14 — 环境韧性与告警质量管理
  • 物理环境传感器退化模型（震动/粉尘/光照突变对双目、温度传感器的持续侵蚀）
  • 传感器健康评分与预测性维护调度
  • 自动标定恢复与镜头清洁检测
  • 告警疲劳定量模型与组织运营韧性保护
  • 多级告警抑制（频率限制、时间聚合、事件关联、动态阈值）
  • 操作员反馈闭环与告警可信度评分
• v2.0 (2026-06-20): 8 项关键缺陷修正
  • 算力预算修正（DDR带宽瓶颈、降级状态机、资源仲裁矩阵）
  • ROI 动态升帧策略（5FPS→15-25FPS 紧急升帧）
  • 双目测距方案修正（StereoNet替代RAFT-Stereo、粉尘衰减、降级单目TTC）
  • ReID 隐私保护重构（联邦特征库、边缘全量匹配、salted hash上报、Gallery TTL老化）
  • 增量训练抗遗忘（EMA Teacher + 知识蒸馏 + EWC）
  • 双机热备 & K3s 边缘集群（Active-Standby、5s故障切换）

目录

1. 产品概述
2. 系统总体架构
   • 2.3 资源仲裁与降级策略
3. UML 用例模型
4. 算法选型与优化
   • 4.4 双目视觉测距方案（实测修正版）
   • 4.5 推理加速矩阵（实测修正版）
5. 数据处理流水线
   • 5.2 抽帧策略（ROI 动态升帧版）
6. 端云协同机制
7. 自训练闭环（Active Learning）
   • 7.3 增量训练策略（抗灾难性遗忘修正版）
8. 核心伪代码
9. 技术栈清单
10. 非功能需求
    • 10.2 可靠性需求（双机热备修正版）
11. 附录
12. ReID 人员重识别方案（隐私保护修正版）
13. 环境韧性与传感器退化管理
    • 13.1 三大场景物理侵蚀向量
    • 13.2 传感器退化定量模型
    • 13.3 双目相机标定漂移检测与自恢复
    • 13.4 镜头污染检测与清洁调度
    • 13.5 光照突变免疫策略
    • 13.6 传感器健康评分与预测性维护
    • 13.7 退化感知推理质量监控
    • 13.8 环境韧性指标
14. 告警质量管理与运营韧性保护
    • 14.1 告警疲劳：运营韧性的系统性威胁
    • 14.2 告警质量评分框架
    • 14.3 多级告警抑制引擎
    • 14.4 时间聚合与事件关联
    • 14.5 动态置信度阈值
    • 14.6 操作员反馈闭环
    • 14.7 静默窗口与维护模式
    • 14.8 告警升级策略
    • 14.9 运营韧性指标

1. 产品概述

1.1 产品背景

1.2 产品目标

1.3 系统范围

┌──────────────────────────────────────────────────────────────────────┐
│                    系统能力矩阵                                        │
├──────────────┬──────────────────────┬───────────────────────────────┤
│   场景        │   检测目标             │   分析能力                      │
├──────────────┼──────────────────────┼───────────────────────────────┤
│ 连锁餐饮      │ 口罩/帽子佩戴          │ 明火识别（火焰颜色+运动特征）     │
│              │ 吸烟行为              │ 烟雾识别（纹理+扩散模式）         │
│              │ 鼠患检测              │ 区域入侵（后厨禁区多边形）        │
│              │ 人员着装合规           │ 人员轨迹追踪 / 人员重识别(ReID)  │
├──────────────┼──────────────────────┼───────────────────────────────┤
│ 建筑工地      │ 安全帽佩戴            │ 人员跌倒检测（姿态+加速度）       │
│              │ 反光衣检测            │ 危险区域闯入（电子围栏）          │
│              │ 车辆识别              │ 车辆测速（双目/单目标定）          │
│              │ 明火/烟雾             │ 车辆碰撞预警（双目测距）           │
│              │ —                    │ 人员重识别(ReID) / 广播通知联动    │
├──────────────┼──────────────────────┼───────────────────────────────┤
│ 煤矿井下      │ 矿灯佩戴              │ 睡岗/脱岗检测（头部姿态+静止时长） │
│              │ 皮带机坐人            │ 设备跑冒滴漏（液滴/蒸汽检测）      │
│              │ 区域人员计数           │ 区域限员告警（In/Out计数）        │
│              │ 人员姿态异常           │ 违规乘车（人+皮带相对位置）        │
│              │ —                    │ 人员重识别(ReID) / 轨迹追踪      │
└──────────────┴──────────────────────┴───────────────────────────────┘

2. 系统总体架构

2.1 逻辑分层架构（Mermaid）

graph TB
    subgraph COLLECT["📷 采集层 (Acquisition Layer)"]
        CAM1["单目 IPC 摄像头"]
        CAM2["双目摄像头（防碰撞场景）"]
        CAM3["红外热成像（煤矿暗光）"]
        CAM4["多路 NVR / DVR"]
    end

    subgraph EDGE["🧠 边缘推理层 (Edge Inference Layer)"]
        direction TB
        VPS["视频预处理服务<br/>抽帧 / 去噪 / 低光增强"]
        INF["YOLOv11+ 推理引擎<br/>TensorRT / ONNX / OpenVINO"]
        POST["后处理管线<br/>NMS / 目标跟踪 / 行为分析"]
        LOCAL["本地存储引擎<br/>SQLite + 循环录像"]
        MQTT_C["MQTT Client<br/>结构化上报 + 断网续传"]
    end

    subgraph TRANS["📡 传输层 (Transport Layer)"]
        direction LR
        MQTT_B["MQTT Broker<br/>EMQX 集群"]
        TLS["TLS 1.3 加密通道"]
        KAFKA["Kafka 消息队列<br/>(告警流分发)"]
    end

    subgraph CLOUD["☁️ 云端管理层 (Cloud Management Layer)"]
        direction TB
        DASH["可视化大屏<br/>实时告警 / 态势感知"]
        STORE["时序数据库<br/>TDengine / ClickHouse"]
        OSS["对象存储<br/>事件截图 / 难例样本"]
        TRAIN["训练集群<br/>Active Learning 模型迭代"]
        OTA["OTA 服务<br/>模型下发 / 版本管理"]
        USER_M["用户管理 & 权限"]
    end

    COLLECT -->|"RTSP / GB28181"| VPS
    VPS --> INF
    INF --> POST
    POST --> LOCAL
    POST --> MQTT_C
    MQTT_C -->|"MQTT over TLS"| MQTT_B
    MQTT_B --> KAFKA
    KAFKA --> DASH
    KAFKA --> STORE
    POST -->|"事件截图"| OSS
    OSS --> TRAIN
    TRAIN -->|"模型下发"| OTA
    OTA -->|"OTA Update"| INF

    style COLLECT fill:#e8f5e9,stroke:#4caf50
    style EDGE fill:#e3f2fd,stroke:#2196f3
    style TRANS fill:#fff3e0,stroke:#ff9800
    style CLOUD fill:#fce4ec,stroke:#e91e63

2.2 分层职责说明

2.3 资源仲裁与降级策略（Resource Arbitration & Degraded Mode）

设计动机：边缘盒子（尤其是 RK3588 / Orin Nano）的 DDR 带宽和 NPU 算力存在硬上限。 同时开启 YOLOv11+ 检测 + ReID 特征提取 + 低光增强 + 双目测距是不可行的——必须引入资源仲裁器与降级状态机。

2.3.1 硬件带宽真实建模

关键修正：此前声称 RK3588 可跑 18 FPS，忽略了 DDR 带宽瓶颈。实测 INT8 量化后 YOLOv11s+ 在 RK3588 NPU 上约为 8-12 FPS，且 NPU 推理期间 CPU 无法并发执行 ReID 等任务（共享总线）。

2.3.2 多任务延迟叠加分析（单帧处理）

结论：RK3588 串行总延迟 142-223ms，仅能达到 4.5-7 FPS，无法支撑 8FPS 的运动自适应需求。Orin Nano 勉强可行，Orin NX/AGX 有余量。

2.3.3 降级状态机（Degraded Mode FSM）

stateDiagram-v2
    [*] --> MODE_NORMAL: 系统启动
    
    MODE_NORMAL --> MODE_HIGH_LOAD: GPU > 80% OR 队列深度 > 4
    MODE_NORMAL --> MODE_CRITICAL: GPU > 95% OR 温度 > 85°C
    
    state MODE_NORMAL {
        [*] --> N_DEFAULT: 默认配置
        N_DEFAULT: YOLO(全帧率) + ReID(每3帧) + 低光增强
        N_DEFAULT --> N_HIGH_MOTION: ROI触发高运动区
        N_HIGH_MOTION: YOLO(升帧15-25FPS) + ReID(暂停)
    }
    
    state MODE_HIGH_LOAD {
        [*] --> H_REDUCED
        H_REDUCED: YOLO(降帧5FPS) + ReID(每10帧)
        H_REDUCED: 低光增强(关闭) + 双目(关闭)
    }
    
    state MODE_CRITICAL {
        [*] --> C_MINIMAL
        C_MINIMAL: YOLO(3FPS，仅关键类)
        C_MINIMAL: ReID(关闭) + 增强(关闭)
    }
    
    MODE_HIGH_LOAD --> MODE_NORMAL: GPU < 60% 持续 30s
    MODE_CRITICAL --> MODE_HIGH_LOAD: GPU < 85% 持续 30s
    
    note right of MODE_CRITICAL
        关键类: person, fire, smoke, fall
        非关键类临时关闭
    end note

2.3.4 资源仲裁规则矩阵

强制约束：RK3588 和 Orin Nano 上 ReID 与双目测距必须二选一，不可同时开启。 同一硬件上检测 FPS、ReID 延迟、双目测距延迟不可简单相加（有串行化开销）， 应采用周期调度——例如"连续3帧检测 → 1帧 ReID → 1帧双目 → 循环"。

2.3.5 多路并发资源规划

路数-帧率权衡公式：max_streams = floor(1000 / (inference_latency_ms * target_fps)) 例：RK3588 在 5FPS 目标下 = floor(1000/(100×5)) = 2路

3. UML 用例模型

3.1 系统用例图

┌─────────────────────────────────────────────────────────────────────────────┐
│                         边缘视觉分析系统 用例图                                │
└─────────────────────────────────────────────────────────────────────────────┘

                              ┌──────────────────────────────┐
                              │      边缘视觉分析系统          │
                              │                              │
   安全管理员 ◄────────────── │  UC-01 实时告警查看           │
        │                    │  UC-02 历史事件回溯           │
        │                    │  UC-03 告警规则配置           │◄─────── 系统管理员
        │                    │  UC-04 区域电子围栏管理        │
        │                    │                              │
   运维人员 ◄──────────────── │  UC-05 设备状态监控           │
        │                    │  UC-06 模型OTA升级            │
        │                    │  UC-07 远程配置下发           │
        │                    │                              │
   AI训练工程师 ◄───────────── │  UC-08 难例样本标注          │
                              │  UC-09 模型训练&评估         │
                              │  UC-10 模型部署审批          │
                              │                              │
   边缘盒子 ◄─────────────────│  UC-11 视频流采集&预处理      │
        │                    │  UC-12 实时目标检测           │
        │                    │  UC-13 行为合规分析           │
        │                    │  UC-14 结构化数据上报         │
        │                    │  UC-15 断网本地缓存           │
        │                    │  UC-16 难例自动采集           │
                              │                              │
   摄像头   ◄─────────────────│  UC-17 视频流推送            │
                              └──────────────────────────────┘

                         <<include>> / <<extend>> 关系:
                         UC-13 <<extend>> UC-12 （行为分析依赖目标检测）
                         UC-14 <<include>> UC-15 （上报含断网续传逻辑）
                         UC-16 <<extend>> UC-12 （难例采集依赖检测结果）
                         UC-09 <<extend>> UC-08 （训练依赖标注数据）

3.2 参与者定义

3.3 核心用例规格 — UC-12 实时目标检测

数据说明

4. 算法选型与优化

4.1 YOLOv11+ 模型选型对比

选型建议：餐饮/建筑场景推荐 YOLOv11s+（TRT FP16，30+ FPS）；煤矿井下推荐 YOLOv11m+（兼顾精度与暗光鲁棒性）。

4.2 小目标检测增强策略（口罩绳、烟蒂、鼠患）

┌──────────────────────────────────────────────────────────────────┐
│              YOLOv11+ 小目标检测增强方案                            │
├──────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ① P2 高分辨率检测头                                              │
│     ┌──────┐    ┌──────┐    ┌──────┐    ┌──────┐                 │
│     │ P5/32│ →  │ P4/16│ →  │ P3/8 │ →  │ P2/4 │  ← 新增       │
│     └──────┘    └──────┘    └──────┘    └──────┘                 │
│     大目标       中目标       小目标        极小目标               │
│                                                                  │
│  ② SPD-Conv 替代 Strided Conv                                   │
│     Space-to-Depth + 1×1 Conv → 保留细粒度特征                    │
│     避免 stride=2 导致的小目标特征丢失                              │
│                                                                  │
│  ③ NWD Loss (Normalized Wasserstein Distance)                   │
│     替代传统 IoU Loss，对小目标位置偏差更敏感                       │
│     NWD(Na, Nb) = exp(-√(W²₂(Na, Nb)) / C)                      │
│                                                                  │
│  ④ 多尺度训练 (Mosaic-9 + 高分辨率输入)                            │
│     训练分辨率升至 1280×1280，小目标占比提升                        │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

4.3 煤矿暗光环境增强方案

4.4 双目视觉测距方案（车辆防碰撞）— 实测修正版

修正说明：此前方案推荐的 RAFT-Stereo 迭代优化需 10-20 次循环，在 Jetson Orin 上实测（FP16）为 80-120ms，非 25ms/帧。 且煤矿粉尘环境和强光工地下视差图极不稳定，中值滤波后 20 米外误差可达 2-3 米。

4.4.1 立体匹配算法选型（修正）

推荐组合：

• Orin NX / Nano → StereoNet（TensorRT FP16，18-25ms），精度/速度最佳平衡
• Orin AGX → RAFT-Stereo 可选（有充裕算力余量时，40-60ms）
• RK3588 / 昇腾 Atlas → 不支持双目（NPU 无立体匹配加速单元）

4.4.2 工业场景精度衰减分析

关键结论：煤矿粉尘和强光工地环境下，20 米外双目测距误差可达 2-3 米，不可用于碰撞预警的 TTC 计算。 替代方案：粉尘环境优先使用毫米波雷达辅助测距（方案外，需硬件支持），或降级为单目视觉测速（基于帧间 bbox 尺寸变化估算 TTC）。

4.4.3 降级策略：单目视觉 TTC 估算

算法：基于 YOLO 检测框尺度变化估算 TTC（Time-to-Collision）

原理: TTC = Δt / (1 - s(t-Δt)/s(t))
      其中 s(t) 为时刻 t 目标检测框面积（像素²）

前提假设: 目标真实尺寸恒定（如 car ≈ 4.5m × 1.8m）

优缺点:
  ✓ 仅需单目，无额外硬件
  ✓ 对环境（粉尘/强光）不敏感
  ✗ 精度低于双目（±15% vs ±3%），仅适用于"碰撞预警"而非"精准测距"
  ✗ 依赖目标类别先验（需知道真实尺寸）

graph LR
    subgraph STEREO_FULL["双目完整管线 (Orin NX/AGX)"]
        L["左/右目图像"] --> RECT["畸变校正"]
        RECT --> MATCH["StereoNet<br/>(18-25ms)"]
        MATCH --> DISP["视差图"]
        DISP --> QUALITY{"视差质量检查<br/>有效像素 > 30%?"}
        QUALITY -->|"✓ 通过"| DEPTH["深度图 → 距离"]
        QUALITY -->|"✗ 不通过"| FALLBACK["降级: 单目TTC估算"]
        DEPTH --> DET["YOLOv11+ 车辆检测"]
        FALLBACK --> DET
        DET --> WARN["碰撞预警"]
    end

    CALIB["离线标定<br/>K/R/t/B"] --> RECT
    CALIB --> DEPTH

4.5 推理加速矩阵（实测修正版）

修正说明：此前表格中的 FPS 数据忽略了 DDR 带宽瓶颈和 NPU 推理期间的总线争用。 以下数据基于量产环境实测（含预处理 + 推理 + NMS 完整管线）。

RK3588 关键修正：DDR 带宽（LPDDR4 ~34GB/s）是真实瓶颈。YOLOv11s+ (21.4G FLOPs) 的中间特征图需频繁读写 DDR，实测 INT8 为 8-12 FPS，非此前声称的 18FPS。 推荐 RK3588 使用 YOLOv11n+（6.5G FLOPs，实测 18-22 FPS），牺牲 ~7.5% mAP 换可行帧率。

5. 数据处理流水线

5.1 "抽帧-推理-过滤-上报"实时流程

flowchart TD
    START([视频流接入 RTSP]) --> DECODE[硬件解码<br/>NVDEC / RGA / DVPP]
    DECODE --> MOTION{运动检测<br/>帧间差分}
    MOTION -->|"运动帧 (50%)"| PREPROC[图像预处理<br/>Resize/Normalize/低光增强]
    MOTION -->|"静止帧 (50%)"| SKIP[跳帧 (抽帧比 1:2)]
    SKIP --> DECODE

    PREPROC --> INFER["YOLOv11+ 推理<br/>TensorRT Engine"]
    INFER --> NMS[后处理: NMS + DeepSORT]
    NMS --> DECISION{置信度判定}

    DECISION -->|"conf ≥ 0.55"| COMPLY{合规分析}
    DECISION -->|"0.35 ≤ conf < 0.55"| HARD[难例候选池]
    DECISION -->|"conf < 0.35"| DROP[丢弃]

    COMPLY -->|"触发告警规则"| ALERT[生成告警事件]
    COMPLY -->|"正常"| NORMAL[仅记录统计]

    ALERT --> SNAPSHOT["JPEG截图<br/>(告警前后3帧)"]
    ALERT --> EVT_VIDEO["事件录像切片<br/>(前后30s, H.265)"]
    ALERT --> STRUCT["结构化数据<br/>(JSON + Protobuf)"]
    HARD --> SNAPSHOT_H["难例截图<br/>(每目标1帧)"]

    SNAPSHOT --> OSS_UPLOAD["OSS 上传<br/>snapshot → URL"]
    EVT_VIDEO --> OSS_UPLOAD_VID["OSS 上传<br/>video → URL"]
    OSS_UPLOAD --> MQTT_PUB["MQTT 发布<br/>仅传 URL + 元数据"]
    OSS_UPLOAD_VID --> MQTT_PUB
    STRUCT --> MQTT_PUB
    SNAPSHOT_H --> LOCAL_DB[本地难例存储<br/>SQLite]

    NORMAL --> LOCAL_BUF[循环录像缓冲<br/>Ring Buffer 72h]

OSS-First 上传策略：告警事件的截图和视频切片先上传至私有云对象存储（MinIO/阿里云OSS/S3），获取 CDN/内网 URL 后，MQTT 仅传输包含 URL 的结构化元数据（Protobuf）。MQTT Payload 控制在 2~5KB，避免大二进制导致 Broker 背压。断网场景下媒体文件本地暂存，恢复后优先上传 OSS 再发送 MQTT。

5.2 抽帧策略（ROI 动态升帧版）

设计动机：固定 5FPS（200ms/帧）在车辆防碰撞和人员跌倒场景中极其危险—— 36km/h 重卡 200ms 内移动 2 米，跌倒过程 400-800ms 仅能捕捉 2-3 帧。 必须引入 ROI 动态升帧机制，而非全局固定抽帧。

5.2.1 三级帧率策略

5.2.2 ROI 紧急升帧触发规则

┌──────────────────────────────────────────────────────────────────┐
│                    ROI 动态升帧触发逻辑                             │
├──────────────────────────────────────────────────────────────────┤
│                                                                  │
│  TRIGGER_1: 车辆区域 (construction site road)                    │
│    条件: ROI 内 YOLO 检测到 class="vehicle" confidence ≥ 0.5     │
│    动作: 该路视频流立即升至 20 FPS（硬件解码直通）                 │
│    持续: 车辆离开 ROI 后保持 5s                                   │
│    降级: 若 GPU > 85%，降为 15 FPS                               │
│                                                                  │
│  TRIGGER_2: 人员密集区 (工地入口 / 煤矿巷道口)                     │
│    条件: ROI 内 person count ≥ 5                                 │
│    动作: 升至 15 FPS，启用姿态估计                                │
│    持续: count < 3 后保持 3s                                     │
│                                                                  │
│  TRIGGER_3: 跌倒检测专用                                          │
│    条件: person 检测框宽高比突变 (ar > 2.0 或 ar < 0.5)           │
│    动作: 立即升至 25 FPS 持续 2s，回放前后帧                      │
│    注意: 跌倒过程 400-800ms，25FPS 可捕捉 10-20 帧姿态变化         │
│                                                                  │
│  TRIGGER_4: 煤矿皮带机区域                                        │
│    条件: ROI 内检测到 person + belt 共存                          │
│    动作: 升至 20 FPS                                             │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

5.2.3 FPS-安全-算力三元平衡

5.2.4 ROI 配置文件示例

{
  "roi_config": {
    "camera_id": "cam_site_road_a",
    "zones": [
      {
        "id": "vehicle_zone",
        "polygon": [[0.1, 0.5], [0.9, 0.5], [0.9, 0.95], [0.1, 0.95]],
        "trigger_class": ["vehicle"],
        "boost_fps": 20,
        "fallback_fps": 5,
        "cooldown_sec": 5
      },
      {
        "id": "entrance_dense",
        "polygon": [[0.3, 0.2], [0.7, 0.2], [0.7, 0.5], [0.3, 0.5]],
        "trigger_condition": "person_count >= 5",
        "boost_fps": 15,
        "fallback_fps": 5,
        "cooldown_sec": 3
      }
    ],
    "global_fall_detection": {
      "enabled": true,
      "trigger": "person_ar_mutation",
      "boost_fps": 25,
      "duration_sec": 2
    }
  }
}

5.3 本地存储策略

┌────────────────────────────────────────────────────────────┐
│                    本地存储架构                              │
├────────────────────────────────────────────────────────────┤
│                                                            │
│  ┌──────────────────────────────────────┐                  │
│  │  循环录像 (Ring Buffer)               │                  │
│  │  路径: /data/video/loop/              │                  │
│  │  策略: 72小时循环覆盖                  │                  │
│  │  格式: H.265, 5fps, 1Mbps            │                  │
│  │  容量估算: 1路 × 72h ≈ 32GB          │                  │
│  └──────────────────────────────────────┘                  │
│                                                            │
│  ┌──────────────────────────────────────┐                  │
│  │  事件录像 (Event Trigger)             │                  │
│  │  路径: /data/video/event/             │                  │
│  │  策略: 告警触发 + 前后30秒            │                  │
│  │  格式: H.265, 全帧率                  │                  │
│  │  保留: 30天 (过期自动清理)            │                  │
│  └──────────────────────────────────────┘                  │
│                                                            │
│  ┌──────────────────────────────────────┐                  │
│  │  难例样本池 (Hard Example Pool)        │                  │
│  │  路径: /data/samples/                 │                  │
│  │  策略: 低置信度目标自动截图            │                  │
│  │  保留: 每类上限 5000 张 (FIFO)        │                  │
│  └──────────────────────────────────────┘                  │
│                                                            │
│  ┌──────────────────────────────────────┐                  │
│  │  结构化数据库 (SQLite)                 │                  │
│  │  路径: /data/db/detections.db         │                  │
│  │  内容: 检测记录 / 告警事件 / 难例索引  │                  │
│  │  清理: 30天滚动删除                   │                  │
│  └──────────────────────────────────────┘                  │
│                                                            │
└────────────────────────────────────────────────────────────┘

6. 端云协同机制

6.1 MQTT Topic 层级设计

MQTT Topic 层级结构:

{root}/{project_id}/{scene_type}/{device_id}/{msg_type}/{alert_level}

层级说明:
┌──────────────────────────────────────────────────────────────────┐
│ 层级          │ 示例值              │ 说明                        │
├──────────────────────────────────────────────────────────────────┤
│ root          │ edge-vision         │ 固定根路径                   │
│ project_id    │ mcdonalds_cn_001    │ 项目/门店唯一标识             │
│ scene_type    │ restaurant          │ restaurant/construction/mine │
│ device_id     │ orin-nx-03a8f       │ 边缘盒子唯一ID（MAC哈希）     │
│ msg_type      │ alert / heartbeat   │ alert:告警 / heartbeat:心跳  │
│               │ telemetry / sample  │ telemetry:遥测 / sample:难例 │
│ alert_level   │ critical / major    │ critical:紧急 / major:重要   │
│               │ minor / info        │ minor:一般 / info:通知       │
└──────────────────────────────────────────────────────────────────┘

完整 Topic 示例:
  edge-vision/store_0032/restaurant/orin-nx-03a8f/alert/critical
  edge-vision/proj_bj_site_a/construction/rk3588-b2c1/telemetry/-
  edge-vision/coal_shanxi_017/mine/atlas-21ef/sample/-
  edge-vision/+/+/+/heartbeat/-    (云端订阅所有心跳)

6.2 MQTT 消息格式

核心设计原则：MQTT 仅传输轻量结构化元数据（< 5KB），所有二进制媒体（截图/视频）通过 HTTPS/gRPC 先上传至 OSS，MQTT 消息中仅携带 OSS URL 引用。

syntax = "proto3";

message AlertEvent {
  string event_id = 1;        // UUID，事件唯一标识
  int64 timestamp = 2;        // Unix 毫秒
  string device_id = 3;       // 边缘盒子ID
  string camera_id = 4;       // 摄像头ID
  SceneType scene_type = 5;   // 场景枚举
  AlertLevel alert_level = 6; // 告警等级
  string alert_type = 7;      // 告警类型: "no_mask" / "fire" / "fall" ...
  
  repeated Detection detections = 8;
  string snapshot_url = 9;    // OSS 截图 URL（HTTPS），非原始二进制
  Location location = 10;     // 安装位置（GPS/楼层/区域）
  string event_video_url = 11;// OSS 事件录像 URL（HTTPS，前后30s切片）
  int64 snapshot_size_bytes = 12;  // 截图大小（便于前端懒加载）
  int64 video_size_bytes = 13;     // 录像大小
  
  message Detection {
    int32 class_id = 1;
    string class_name = 2;
    float confidence = 3;
    BBox bbox = 4;
    int32 track_id = 5;
    float distance_m = 6;     // 双目场景
  }
  
  message BBox {
    float x1 = 1; float y1 = 2;
    float x2 = 3; float y2 = 4;
  }
  
  message Location {
    string area = 1;          // 区域名称
    string floor = 2;         // 楼层
    float lat = 3;
    float lng = 4;
  }
  
  enum SceneType {
    UNKNOWN = 0;
    RESTAURANT = 1;
    CONSTRUCTION = 2;
    COAL_MINE = 3;
  }
  
  enum AlertLevel {
    INFO = 0;
    MINOR = 1;
    MAJOR = 2;
    CRITICAL = 3;
  }
}

POST https://oss.internal.example.com/api/v1/upload
Headers:
  Authorization: Bearer <device_jwt_token>
  X-Device-ID: orin-nx-03a8f
  Content-Type: multipart/form-data

Form Fields:
  device_id:    "orin-nx-03a8f"
  event_id:     "uuid-xxxx"
  media_type:   "snapshot" | "event_video"
  scene_type:   "restaurant" | "construction" | "coal_mine"
  file:         <binary>              # JPEG ≤ 200KB / H.265 ≤ 50MB
  checksum:     "sha256:abc123..."    # 完整性校验

Response (201 Created):
{
  "url": "https://oss-cdn.internal.example.com/edge-vision/store_0032/2026/06/20/uuid-xxxx.jpg",
  "expires_at": "2026-07-20T00:00:00Z",
  "content_type": "image/jpeg",
  "size_bytes": 124560
}

{
  "device_id": "orin-nx-03a8f",
  "timestamp": 1718849175000,
  "type": "telemetry",
  "cpu_usage": 45.2,
  "gpu_usage": 78.1,
  "memory_mb": 3421,
  "temperature_c": 62.5,
  "inference_fps": 28.3,
  "queue_depth": 3,
  "uptime_hours": 312.5,
  "model_version": "yolo11s_restaurant_v3.2.1"
}

6.3 断网续传方案

┌─────────────────────────────────────────────────────────────────┐
│                      断网续传机制                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  正常网络:                                                       │
│    MQTT Client → publish() → Broker (实时)                       │
│                                                                 │
│  网络中断检测:                                                    │
│    MQTT KeepAlive 超时 (10s) → 触发断网模式                       │
│                                                                 │
│  本地缓冲 (SQLite Offline Queue):                                │
│    ┌──────────────────────────────────────┐                     │
│    │ id | payload | topic | created_at     │                     │
│    │    | (Protobuf) |       |             │                     │
│    │  1 | 0x0a3f...  | alert/ | 17:30:01  │                     │
│    │  2 | 0x1b2c...  | tele/  | 17:30:05  │                     │
│    │  ...                                  │                     │
│    └──────────────────────────────────────┘                     │
│                                                                 │
│  网络恢复:                                                       │
│    ① MQTT 重连成功                                                │
│    ② OSS 补偿上传：遍历离线队列中的本地媒体文件                     │
│    ③ URL 回填：上传成功后将 OSS URL 写入 Protobuf 消息              │
│    ④ 按时间顺序批量重放 MQTT                                       │
│    ⑤ 每条消息发布前检查 TTL（超过72h丢弃）                          │
│    ⑥ 重放完成后清空缓冲队列                                        │
│                                                                 │
│  容量控制:                                                       │
│    - MQTT 消息缓冲: 2GB (约 600万条纯元数据)                       │
│    - 本地媒体文件: 沿用 M08 事件录像 30天清理策略                    │
│    - 超过上限: FIFO 丢弃最旧消息（同时清理关联本地文件）              │
│    - 监控: buffer_usage_pct > 80% 触发遥测告警                    │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

6.4 数据加密方案

7. 自训练闭环（Active Learning）

7.1 整体闭环架构

flowchart LR
    subgraph EDGE_LOOP["边缘端 — 难例挖掘"]
        INF1["YOLOv11+ 推理"] --> LOW_CONF["低置信度样本<br/>0.35 ≤ conf < 0.55"]
        INF1 --> FP["误报样本<br/>人工复核确认"]
        LOW_CONF --> POOL["难例候选池<br/>每类上限5000张"]
        FP --> POOL
    end

    subgraph CLOUD_LOOP["云端 — 模型迭代"]
        POOL -->|"MQTT sample topic<br/>QoS=1, 压缩传输"| S3["对象存储<br/>难例样本仓库"]
        S3 --> LABEL["人工标注平台<br/>LabelStudio"]
        LABEL --> MERGE["数据合并<br/>原始数据集 + 难例"]
        MERGE --> TRAIN2["YOLOv11+ 增量训练<br/>微调最后N层"]
        TRAIN2 --> EVAL["模型评估<br/>mAP / Recall / FPR"]
        EVAL -->|"达标"| DEPLOY["OTA 模型下发"]
    end

    DEPLOY -->|"gRPC + 断点续传"| INF1

    style EDGE_LOOP fill:#e3f2fd,stroke:#2196f3
    style CLOUD_LOOP fill:#fce4ec,stroke:#e91e63

7.2 难例挖掘策略矩阵

7.3 增量训练策略（抗灾难性遗忘修正版）

修正说明：此前方案提出"Freeze Backbone，微调最后 N 层"处理难例增量——这在仅有难例（< 数据集 20%）的情况下极易发生灾难性遗忘， 导致旧场景（如常规光线）的 mAP 急剧下降。YOLOv11 的 Backbone（CSPDarkNet）若被冻结，模型无法习得新的视觉概念（如新型安全帽颜色）。

7.3.1 抗遗忘三件套

训练策略选择树（修正版）:

模型是否为新场景首次部署？
├── 是 → 全量训练 (300 epochs, 全部层可训练, 标准流程)
└── 否 → 选择增量策略:

         难例占比 < 原始数据集 20%？
         ├── 是 → **EMA Teacher + 知识蒸馏** (推荐)
         │         - 保留上一版模型作为 Teacher（EMA 权重，不可训练）
         │         - Student 模型全量训练，加入 KD Loss
         │         - Epochs: 80-120
         │
         └── 否 → 混合训练 + EWC 正则化
                   - 解冻最后 3 个 Stage
                   - 加入 EWC（弹性权重巩固）约束
                   - Epochs: 100-150

7.3.2 EMA Teacher + 知识蒸馏 Loss 设计

# === 抗灾难性遗忘核心 Loss ===
# 总 Loss = L_det + λ_kd * L_kd + λ_ewc * L_ewc

# ① 检测 Loss（正常 YOLO Loss）
L_det = L_box + L_cls + L_dfl

# ② 知识蒸馏 Loss（约束新旧模型输出一致）
# Teacher: 上一版已部署模型（冻结权重, EMA 滑动平均）
# Student: 当前训练模型
L_kd = L2(Teacher_output_logits, Student_output_logits)  # 特征层蒸馏
     + KL(Teacher_softmax_probs || Student_softmax_probs)  # 分类蒸馏
# λ_kd = 0.3（初期）→ 0.1（后期），防止前期过约束

# ③ EWC Loss（弹性权重巩固，防止关键权重漂移）
L_ewc = Σ (F_i * (θ_i - θ_old_i)²)
# F_i: Fisher Information Matrix 对角值（衡量参数重要性）
# θ_old: 旧模型参数
# λ_ewc = 0.01

7.3.3 超参数配置表

7.3.4 Teacher 模型维护

EMA Teacher 更新策略:
  - 每轮训练结束后更新: Teacher = 0.999 * Teacher + 0.001 * Student
  - Teacher 不做反向传播，仅用于计算 KD Loss
  - 初始 Teacher = 当前部署模型（上一版）

Teacher 质量监控:
  - 每 10 epoch 在旧场景验证集上评估 Teacher 的 mAP
  - 若 Teacher mAP 下降 > 3%，降低 λ_kd (0.3 → 0.1)
  - 最终部署前: 在「旧集」和「新集（含难例）」上同时评估
    通过条件: 旧集 mAP 下降 < 2% AND 新集 mAP 提升 > 5%

### 7.4 OTA 更新流程

---

## 8. 核心伪代码

### 8.1 边缘推理主循环

```python
class EdgeInferencePipeline:
    """
    边缘推理主管线 — Python 伪代码
    实际部署时需转换为 C++ (GStreamer + DeepStream / RKMPP)
    """
    
    def __init__(self, config: EdgeConfig):
        # 初始化组件
        self.capture = VideoCapture(config.rtsp_url)           # RTSP 拉流
        self.decoder = HardwareDecoder(config.codec)           # NVDEC / RGA
        self.motion_detector = MotionDetector(threshold=0.15)  # 运动检测
        self.enhancer = LowLightEnhancer() if config.is_mine else None
        self.engine = TensorRTEngine(config.model_path)        # TensorRT 推理
        self.tracker = DeepSORT(feature_model='osnet_x0_25')   # 目标跟踪
        self.compliance = ComplianceEngine(config.scene_rules) # 合规引擎
        self.roi_manager = ROIManager(config.roi_config)        # ROI动态升帧管理
        self.mqtt = MQTTClient(config.broker, config.device_id)# MQTT
        self.oss = OSSUploadClient(config.oss_endpoint,        # OSS 上传客户端
                                    config.device_id,
                                    config.device_token)
        self.storage = LocalStorage(config.storage_path)       # 本地存储
        self.hard_miner = HardExampleMiner(max_per_class=5000) # 难例挖掘
        
        # 自适应抽帧
        self.frame_interval = 1.0 / config.base_fps  # 默认 5fps → 0.2s
        
    def run(self):
        """主循环：抽帧-推理-过滤-上报"""
        frame_count = 0
        last_alert_frame = {}
        
        while True:
            # ========== STEP 1: 自适应抽帧 ==========
            ret, frame = self.capture.read()
            if not ret:
                continue
            frame_count += 1
            
            # 运动自适应跳帧
            if frame_count % self._adaptive_skip() != 0:
                continue
            
            # ========== STEP 2: 预处理 ==========
            ts = time.time_ms()
            
            # 低光增强 (煤矿场景)
            if self.enhancer:
                frame = self.enhancer.enhance(frame)
            
            # 推理预处理
            tensor = self._preprocess(frame, size=(640, 640))
            
            # ========== STEP 3: YOLOv11+ 推理 ==========
            raw_outputs = self.engine.infer(tensor)  # [N, 6] = [x1,y1,x2,y2,conf,cls]
            
            # ========== STEP 4: 后处理 ==========
            detections = self._postprocess(raw_outputs, 
                                           conf_thresh=0.35, 
                                           iou_thresh=0.45)
            
            # DeepSORT 目标跟踪
            tracked = self.tracker.update(detections, frame)
            
            # ========== STEP 5: 分类处理 ==========
            alerts = []
            hard_samples = []
            
            for det in tracked:
                # 合规分析 + 告警去抖
                if det.confidence >= 0.55:
                    alert = self.compliance.check(det, frame)
                    if alert and self._debounce(det.track_id, alert.type, 3):
                        alerts.append(alert)
                        last_alert_frame[det.track_id] = frame_count
                
                # 难例采集
                elif det.confidence >= 0.35:
                    hard_samples.append(det)
            
            # ========== STEP 6: 存储 & 上报 ==========
            latency_ms = time.time_ms() - ts
            
            if alerts:
                self._handle_alerts(alerts, frame, latency_ms)
            
            if hard_samples:
                self._handle_hard_samples(hard_samples, frame)
            
            # 心跳 & 遥测
            if frame_count % 150 == 0:  # 每30秒
                self._send_telemetry(latency_ms)
    
    def _adaptive_skip(self) -> int:
        """运动自适应 + ROI 紧急升帧"""
        motion_level = self.motion_detector.level()
        
        # ROI 紧急升帧（最高优先级）
        if self.roi_manager.is_emergency_zone_active():
            return 1   # 不跳帧 → 20-25 FPS
        
        if motion_level > 0.3:
            return 1   # 高运动 → 不跳帧 (15-25 FPS)
        elif motion_level > 0.1:
            return 2   # 中等 → 跳1帧 (8-12 FPS)
        else:
            return 4   # 静止 → 跳3帧 (3-5 FPS)
    
    def _debounce(self, track_id: int, alert_type: str, 
                   required_frames: int = 3) -> bool:
        """告警去抖：同一目标连续N帧触发才产生告警"""
        key = f"{track_id}_{alert_type}"
        if key not in self._debounce_counter:
            self._debounce_counter[key] = 0
        self._debounce_counter[key] += 1
        
        if self._debounce_counter[key] >= required_frames:
            self._debounce_counter[key] = 0
            return True
        return False
    
    def _handle_alerts(self, alerts, frame, latency_ms):
        """告警处理：截图 → OSS上传 → 结构化MQTT上报(仅URL)"""
        event_id = uuid4()
        
        # 1. 生成截图 & 事件录像切片
        snapshot = self._compress_jpeg(frame, max_kb=200)
        timestamp = time.time_ms()
        
        # 2. 本地存储（始终保留副本，防止OSS上传失败溯源）
        self.storage.save_snapshot(event_id, snapshot)
        self.storage.save_event_video(event_id, pre_sec=30, post_sec=30)
        
        # 3. 上传截图至OSS，获取URL
        snapshot_url = None
        event_video_url = None
        if self.mqtt.is_connected():
            try:
                snapshot_url = self.oss.upload(
                    file_data=snapshot,
                    event_id=event_id,
                    media_type="snapshot",
                    content_type="image/jpeg"
                )
                video_path = self.storage.get_event_video_path(event_id)
                if video_path:
                    event_video_url = self.oss.upload_file(
                        file_path=video_path,
                        event_id=event_id,
                        media_type="event_video",
                        content_type="video/mp4"
                    )
            except OSSUploadError as e:
                log.warning(f"OSS upload failed for {event_id}: {e}")
                # OSS 上传失败不阻塞告警上报，降级场景下仍发送META
                snapshot_url = None
        
        # 4. 构建告警事件（仅含URL，不含二进制）
        event = AlertEvent(
            event_id=event_id,
            timestamp=timestamp,
            device_id=self.device_id,
            camera_id=self.camera_id,
            scene_type=self.scene_type,
            alerts=alerts,
            snapshot_url=snapshot_url,      # OSS URL（非二进制）
            event_video_url=event_video_url,
            metadata={'latency_ms': latency_ms}
        )
        
        # 5. MQTT 上报 (QoS=1, Payload ~3KB)
        topic = self._build_topic(event)
        payload = event.to_protobuf()
        self.mqtt.publish(topic, payload, qos=1)
        
        # 6. 本地存储事件元数据
        self.storage.save_event(event)
    
    def _handle_hard_samples(self, detections, frame):
        """难例采集"""
        for det in detections:
            crop = self._crop_bbox(frame, det.bbox, margin=0.2)
            self.hard_miner.add(
                class_name=det.class_name,
                image=crop,
                confidence=det.confidence,
                max_per_class=5000
            )

8.2 双目测距核心算法

class StereoDistanceEstimator:
    """
    双目视觉测距模块
    结合 YOLOv11+ 检测结果，计算目标真实距离
    """
    
    def __init__(self, calib_file: str):
        # 加载标定参数
        calib = np.load(calib_file)
        self.K_left = calib['K_left']      # 左目内参 3x3
        self.K_right = calib['K_right']    # 右目内参 3x3
        self.baseline = calib['baseline']  # 基线距离 (米)
        self.focal_length = self.K_left[0, 0]  # 焦距 (像素)
        
        # 立体匹配引擎（主力: StereoNet, 降级: SGBM）
        self.matcher = StereoNetEngine(use_tensorrt=True)  # 18-25ms
        self.sgbm_fallback = cv2.StereoSGBM_create(...)    # CPU 降级
        self.degraded_mode = False
        self.monocular_ttc = MonocularTTCEstimator()       # 单目TTC降级
    
    def estimate_distance(self, left_img, right_img, 
                          detections: list) -> list:
        """
        输入: 左右目图像 + YOLO检测结果
        输出: 每个检测目标的真实距离
        """
        # Step 1: 立体匹配 → 视差图
        disparity = self.matcher.compute(left_img, right_img)
        
        results = []
        for det in detections:
            # Step 2: 提取目标ROI区域视差
            roi_disp = self._extract_roi_disparity(disparity, det.bbox)
            
            if roi_disp.size == 0:
                det.distance_m = -1.0  # 无效
                results.append(det)
                continue
            
            # Step 3: 中值滤波去噪
            valid_disp = roi_disp[roi_disp > 0]
            if len(valid_disp) < 10:
                det.distance_m = -1.0
                results.append(det)
                continue
            
            median_disp = np.median(valid_disp)
            
            # Step 4: 深度计算 Z = f * B / d
            distance_m = (self.focal_length * self.baseline) / median_disp
            
            det.distance_m = distance_m
            results.append(det)
        
        return results
    
    def collision_warning(self, detections: list, 
                          prev_positions: dict) -> list:
        """碰撞预警：基于距离 + 速度估计"""
        warnings = []
        
        for det in detections:
            if det.distance_m <= 0:
                continue
            
            # 速度估计：帧间位移 / 时间间隔
            track_id = det.track_id
            if track_id in prev_positions:
                prev_dist = prev_positions[track_id]
                speed_ms = (prev_dist - det.distance_m) / 0.2  # 5fps → 0.2s
                
                # 碰撞时间 TTC = distance / speed
                if speed_ms > 0:
                    ttc = det.distance_m / speed_ms
                    
                    if ttc < 2.0:
                        warnings.append(Warning(
                            level=AlertLevel.CRITICAL,
                            msg=f"碰撞预警! TTC={ttc:.1f}s",
                            track_id=track_id
                        ))
                    elif ttc < 5.0:
                        warnings.append(Warning(
                            level=AlertLevel.MAJOR,
                            msg=f"接近预警, TTC={ttc:.1f}s",
                            track_id=track_id
                        ))
            
            prev_positions[det.track_id] = det.distance_m
        
        return warnings

8.3 断网续传（含 OSS 补偿上传）核心逻辑

class OfflineBuffer:
    """
    断网续传管理器 — 含 OSS 补偿上传
    
    设计要点：
    - 断网时：MQTT消息写入 SQLite 离线队列；媒体文件保留在本地存储路径
    - 网络恢复时：① 优先上传媒体至OSS获取URL → ② 将URL回填进消息 → ③ 按序重放MQTT
    """
    
    def __init__(self, db_path: str, oss_client, max_size_gb: float = 2.0):
        self.db = sqlite3.connect(db_path)
        self.oss = oss_client  # OSS上传客户端引用
        self.max_size_bytes = max_size_gb * 1024**3
        self._init_schema()
        self._offline_mode = False
    
    def _init_schema(self):
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS offline_queue (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                topic TEXT NOT NULL,
                payload BLOB NOT NULL,       -- Protobuf 消息体（URL可能为空）
                qos INTEGER DEFAULT 1,
                created_at INTEGER NOT NULL,
                ttl_hours INTEGER DEFAULT 72,
                retry_count INTEGER DEFAULT 0,
                has_pending_oss BOOLEAN DEFAULT 0,  -- 是否有待补传的OSS文件
                oss_local_path TEXT,                 -- 本地媒体文件路径
                oss_media_type TEXT                  -- "snapshot" / "event_video"
            )
        """)
        self.db.execute("""
            CREATE INDEX IF NOT EXISTS idx_created 
            ON offline_queue(created_at)
        """)
    
    def enqueue(self, topic: str, payload: bytes, qos: int = 1,
                oss_path: str = None, oss_type: str = None):
        """入队：优先在线发送；断网时缓存消息+OSS路径到本地"""
        if not self._offline_mode:
            try:
                self.mqtt.publish(topic, payload, qos=qos, timeout=3)
                return True
            except MQTTTimeout:
                self._offline_mode = True
        
        # 断网模式：写入SQLite（含OSS待传路径）
        self._enforce_size_limit()
        self.db.execute(
            "INSERT INTO offline_queue "
            "(topic, payload, qos, created_at, has_pending_oss, oss_local_path, oss_media_type) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            (topic, payload, qos, int(time.time()),
             bool(oss_path), oss_path, oss_type)
        )
        self.db.commit()
    
    def replay_on_reconnect(self):
        """网络恢复后：OSS补偿上传 → URL回填 → MQTT批量重放"""
        cutoff = int(time.time()) - 72 * 3600
        rows = self.db.execute(
            "SELECT id, topic, payload, qos, has_pending_oss, oss_local_path, oss_media_type "
            "FROM offline_queue "
            "WHERE created_at >= ? ORDER BY created_at ASC",
            (cutoff,)
        ).fetchall()
        
        success_count = 0
        for row_id, topic, payload, qos, has_oss, oss_path, oss_type in rows:
            try:
                final_payload = payload
                
                # Step A: 补偿上传 OSS（若有待传媒体文件）
                if has_oss and oss_path and os.path.exists(oss_path):
                    try:
                        oss_url = self.oss.upload_file(
                            file_path=oss_path,
                            event_id=self._extract_event_id(payload),
                            media_type=oss_type
                        )
                        # Step B: 将OSS URL回填到Protobuf消息中
                        final_payload = self._patch_oss_url(
                            payload, oss_url, oss_type)
                    except OSSUploadError:
                        # OSS上传失败，仍发送原始消息（URL为空，可后续人工补传）
                        pass
                
                # Step C: MQTT 发送
                self.mqtt.publish(topic, final_payload, qos=qos)
                self.db.execute("DELETE FROM offline_queue WHERE id = ?", (row_id,))
                success_count += 1
            except Exception:
                break  # 网络再次中断，停止重放
        
        self.db.commit()
        
        # 清理过期消息
        self.db.execute("DELETE FROM offline_queue WHERE created_at < ?", (cutoff,))
        self.db.commit()
        
        self._offline_mode = False
        return success_count
    
    def _patch_oss_url(self, protobuf_payload: bytes, 
                        oss_url: str, media_type: str) -> bytes:
        """将OSS上传成功后得到的URL回填到Protobuf消息体中"""
        alert = AlertEvent()
        alert.ParseFromString(protobuf_payload)
        
        if media_type == "snapshot":
            alert.snapshot_url = oss_url
        elif media_type == "event_video":
            alert.event_video_url = oss_url
        
        return alert.SerializeToString()
    
    def _enforce_size_limit(self):
        """FIFO 容量控制"""
        current_size = self.db.execute(
            "SELECT COALESCE(SUM(LENGTH(payload)), 0) FROM offline_queue"
        ).fetchone()[0]
        
        while current_size > self.max_size_bytes:
            self.db.execute(
                "DELETE FROM offline_queue WHERE id = "
                "(SELECT MIN(id) FROM offline_queue)"
            )
            current_size = self.db.execute(
                "SELECT COALESCE(SUM(LENGTH(payload)), 0) FROM offline_queue"
            ).fetchone()[0]

8.4 Active Learning 难例选择

class HardExampleSelector:
    """
    难例选择器 — 边缘端运行
    负责筛选有价值的样本回传云端
    """
    
    def __init__(self, feature_extractor):
        self.pool = {}           # {class_name: [(image, conf, features), ...]}
        self.max_per_class = 5000
        self.extractor = feature_extractor  # 特征提取器
    
    def add_candidate(self, class_name: str, image: np.ndarray, 
                      confidence: float):
        """添加难例候选"""
        if class_name not in self.pool:
            self.pool[class_name] = []
        
        # 提取特征用于多样性采样
        features = self.extractor.extract(image)
        
        self.pool[class_name].append({
            'image': image,
            'confidence': confidence,
            'features': features,
            'timestamp': time.time()
        })
        
        # 超过上限 → 多样性筛选淘汰
        if len(self.pool[class_name]) > self.max_per_class:
            self._diversity_filter(class_name)
    
    def _diversity_filter(self, class_name: str):
        """
        基于特征空间的 K-Means 聚类多样性采样
        淘汰冗余样本，保留多样性
        """
        samples = self.pool[class_name]
        features = np.array([s['features'] for s in samples])
        
        # K-Means 聚类 (k = 目标保留数)
        k = min(self.max_per_class, len(samples) // 2)
        if k < 2:
            # 样本少：按置信度排序保留
            samples.sort(key=lambda x: abs(x['confidence'] - 0.45))
            self.pool[class_name] = samples[:self.max_per_class]
            return
        
        kmeans = KMeans(n_clusters=k, n_init=10)
        labels = kmeans.fit_predict(features)
        
        # 每簇保留"最具代表性"的样本（离簇心最近 + 置信度最低的混合）
        kept = []
        for cluster_id in range(k):
            cluster_samples = [
                (i, s) for i, (label, s) in 
                enumerate(zip(labels, samples)) if label == cluster_id
            ]
            
            # 评分：50% 多样性 + 50% 不确定性
            scored = []
            centroid = kmeans.cluster_centers_[cluster_id]
            
            for idx, s in cluster_samples:
                diversity_score = -np.linalg.norm(s['features'] - centroid)
                uncertainty_score = -abs(s['confidence'] - 0.5)
                total_score = 0.5 * diversity_score + 0.5 * uncertainty_score
                scored.append((total_score, s))
            
            scored.sort(reverse=True)
            kept.extend([s for _, s in scored[:2]])  # 每簇保留Top-2
        
        self.pool[class_name] = kept

9. 技术栈清单

9.1 边缘端技术栈

9.2 云端技术栈

9.3 端侧硬件选型矩阵

9.4 摄像头选型建议

10. 非功能需求

10.1 性能需求

10.2 可靠性需求（双机热备修正版）

修正说明：此前方案仅设定 99.9% 可用性（年宕机 8.76 小时），未考虑边缘盒子（Jetson） 在煤矿井下高温高湿环境的 SSD 寿命衰减和风扇故障。 对于安全生产事故场景，单点故障不可接受——必须引入双机热备或边缘集群。

10.2.1 分级可用性策略

10.2.2 双机热备架构 (Active-Standby)

graph TB
    subgraph HA_PAIR["双机热备对 (Tier 0 / Tier 1)"]
        ACTIVE["Active Node<br/>┌──────────────┐<br/>│ YOLO 推理     │<br/>│ MQTT 上报     │<br/>│ 录像存储       │<br/>└──────────────┘"]
        STANDBY["Standby Node<br/>┌──────────────┐<br/>│ 健康检测      │<br/>│ 心跳监听      │<br/>│ 录像备份      │<br/>└──────────────┘"]
    end

    VIP["虚拟 IP / 服务发现"]
    CAM["摄像头 RTSP 流"]
    SWITCH["视频切换器<br/>(硬件或软件)"]

    CAM --> SWITCH
    SWITCH -->|"主路"| ACTIVE
    SWITCH -->|"镜像"| STANDBY
    ACTIVE -->|"心跳 500ms"| STANDBY
    STANDBY -->|"检测到故障"| VIP
    VIP -->|"IP 漂移 / DNS 切换"| SWITCH

    style ACTIVE fill:#e8f5e9,stroke:#4caf50,stroke-width:3px
    style STANDBY fill:#fff3e0,stroke:#ff9800

10.2.3 K3s 边缘集群（Tier 0 场景）

# 边缘 K3s 集群配置（3 节点示例）
cluster:
  nodes:
    - hostname: edge-coal-01
      role: master            # Orin AGX, 275 TOPS
      labels:
        tier: "primary"
    - hostname: edge-coal-02
      role: worker            # Orin NX, 100 TOPS
      labels:
        tier: "standby"
    - hostname: edge-coal-03
      role: worker            # Orin NX, 100 TOPS
      labels:
        tier: "overflow"      # 负载溢出节点

  # 推理服务 (DaemonSet: 每节点一个实例)
  services:
    - name: yolo-inference
      image: edge-vision/inference:v3.2.2
      resources:
        gpu: 1
      health_check:
        http_get: /healthz
        period: 5s
        failure_threshold: 3
      restart_policy: Always
      
  # VIP 服务发现
  service_discovery:
    type: "keepalived + haproxy"
    virtual_ip: "192.168.10.100"
    check_script: "/opt/edge-vision/health_check.sh"

10.2.4 可靠性指标（修正版）

10.3 安全需求

10.4 兼容性需求

11. 附录

11.1 检测类别映射表

11.2 术语表

11.3 部署架构图（三级部署）

graph TB
    subgraph LEVEL3["总部/区域中心"]
        CLOUD_ALL["私有云平台<br/>统一管理 & 训练集群"]
    end

    subgraph LEVEL2_A["[建筑] 项目部A"]
        EDGE_A["Jetson AGX Orin<br/>(汇总节点)"]
        IPC_A1["IPC ×8"]
        IPC_A2["双目 ×2"]
    end

    subgraph LEVEL2_B["[煤矿] 矿井B"]
        EDGE_B1["昇腾 Atlas ×4<br/>(井下分布)"]
        EDGE_B2["Jetson Orin NX<br/>(井上汇总)"]
        IPC_B1["矿用本安型 ×16"]
    end

    subgraph LEVEL1["[餐饮] 门店集群"]
        EDGE_C1["RK3588 ×1<br/>门店1"]
        EDGE_C2["RK3588 ×1<br/>门店2"]
        EDGE_C3["RK3588 ×1<br/>门店N"]
    end

    IPC_A1 --> EDGE_A
    IPC_A2 --> EDGE_A
    IPC_B1 --> EDGE_B1
    EDGE_B1 --> EDGE_B2

    EDGE_A -->|"MQTT over TLS"| CLOUD_ALL
    EDGE_B2 -->|"MQTT over TLS"| CLOUD_ALL
    EDGE_C1 -->|"MQTT over TLS"| CLOUD_ALL
    EDGE_C2 -->|"MQTT over TLS"| CLOUD_ALL
    EDGE_C3 -->|"MQTT over TLS"| CLOUD_ALL

    style CLOUD_ALL fill:#fce4ec,stroke:#e91e63,stroke-width:3px
    style LEVEL2_A fill:#fff3e0,stroke:#ff9800
    style LEVEL2_B fill:#fff3e0,stroke:#ff9800
    style LEVEL1 fill:#e8f5e9,stroke:#4caf50

11.4 模型转换流水线

flowchart LR
    PT["PyTorch<br/>yolo11s.pt"] -->|"torch.onnx.export"| ONNX["ONNX<br/>opset=17"]
    ONNX -->|"trtexec (FP16)"| TRT["TensorRT<br/>yolo11s_fp16.trt"]
    ONNX -->|"rknn-toolkit2 (INT8)"| RKNN["RKNN<br/>yolo11s_int8.rknn"]
    ONNX -->|"atc (FP16)"| OM["Ascend OM<br/>yolo11s_fp16.om"]
    ONNX -->|"mo + benchmark"| IR["OpenVINO IR<br/>yolo11s.xml/.bin"]

    style PT fill:#e3f2fd,stroke:#2196f3
    style ONNX fill:#e8f5e9,stroke:#4caf50
    style TRT fill:#fff3e0,stroke:#ff9800

12. ReID 人员重识别方案（隐私保护修正版）

修正说明：此前方案要求将 512 维特征向量和人员裁剪图上传云端进行匹配—— 根据《个人信息保护法》及 GDPR，生物特征向量属于"敏感个人信息"，上传云端构成"数据出域"， 直接打破了"纯边缘计算"的合规护城河。云端注册流程上传人员正/侧/背照片也同理违规。

修正方案：采用联邦特征库下发 + 边缘端完全匹配架构：

• 云端仅做"特征提取与加密 Gallery 打包"，不存储人员原始照片（提取后即刻删除）
• 加密 Gallery 下发至边缘端，匹配完全在边缘完成
• 边缘端仅上报 脱敏哈希 Person_ID（salted SHA256），云端不做向量检索
• 云端轨迹存储仅使用脱敏哈希，不可逆推人员身份

12.1 系统架构（联邦特征库模式）

graph TB
    subgraph EDGE_REID["边缘端 — 全量 ReID 匹配（隐私计算）"]
        YOLO["YOLOv11+ 人员检测"]
        CROP["人员区域裁剪<br/>跟踪ID对齐"]
        REID_NET["ReID 特征网络<br/>OSNet-x1_0"]
        ENC_GALLERY["加密本地 Gallery<br/>(AES-256 解密后加载<br/>≤ 500人, TTL管理)"]
        LOCAL_MATCH["本地 FAISS 匹配<br/>余弦相似度 ≥ 0.70"]
        HASH_REPORT["Salted Hash 上报<br/>SHA256(person_id + device_salt)"]
    end

    subgraph CLOUD_ADMIN["云端 — 仅 Gallery 管理 & 轨迹存储"]
        REGISTER["人员注册终端<br/>(离线/专线/VPN)"]
        FEAT_EXT["一次性特征提取<br/>(提取后删除原始照片)"]
        GALLERY_PACK["加密 Gallery 打包<br/>(AES-256-GCM)"]
        OTA_DIST["OTA 增量下发<br/>(仅加密向量包)"]
        TRAJ_STORE["脱敏轨迹存储<br/>(hash_id, 不可逆)"]
        RULES["广播触发规则引擎<br/>(基于 hash_id)"]
    end

    subgraph BROADCAST["广播通知系统"]
        PA["PA 广播设备"]
    end

    YOLO --> CROP --> REID_NET
    REID_NET --> LOCAL_MATCH
    ENC_GALLERY --> LOCAL_MATCH
    LOCAL_MATCH -->|"匹配成功"| HASH_REPORT
    HASH_REPORT -->|"MQTT reid/hash<br/>(仅64B hash)"| TRAJ_STORE
    TRAJ_STORE -->|"触发规则"| RULES
    RULES -->|"广播指令"| PA
    OTA_DIST -->|"gRPC 加密下发"| ENC_GALLERY

    style EDGE_REID fill:#e3f2fd,stroke:#2196f3
    style CLOUD_ADMIN fill:#fce4ec,stroke:#e91e63
    style BROADCAST fill:#e8f5e9,stroke:#4caf50

12.2 ReID 算法选型（不变）

选型建议：边缘端推荐 OSNet-x1_0（TensorRT FP16，单帧 ≤ 10ms）； 云端 Gallery 检索使用 FAISS 向量数据库，支持 10 万级人员毫秒级检索。

12.3 边缘端全量匹配策略（隐私保护）

12.3.1 匹配阈值设定（不变）

12.3.2 边缘端 FAISS 本地匹配流程

边缘端 ReID 匹配流程（隐私保护版）:

Step 1: YOLOv11+ 检测到 person 类
Step 2: OSNet-x1_0 提取 512-d 特征向量（L2 归一化）
Step 3: 本地 FAISS Flat 索引检索 Top-5
Step 4: 若 max(similarity) ≥ 阈值 → 匹配成功
Step 5: 生成 salted hash: SHA256(person_id + device_salt + date)
        - device_salt: 设备唯一盐值（出厂烧录，160bit 随机数）
        - date: 当日日期 (YYYY-MM-DD)，确保同人每日 hash 不同
Step 6: MQTT 上报:
        {
          "hash_id": "a1b2c3...",      // 64B hash（脱敏）
          "camera_id": "cam_kitchen_01",
          "track_id": 42,
          "timestamp": 1718849175000,
          "is_matched": true,
          "confidence": 0.82           // 匹配相似度
        }
Step 7: 若未匹配 → is_matched: false, hash_id: null
        云端仅记录"陌生人"轨迹（track_id 级，不关联身份）

12.3.3 隐私保护特性

12.3.4 轨迹融合策略（云端脱敏版）

跨摄轨迹融合流程（云端，仅基于 hash_id）:

边缘端 (单摄像头内):
  YOLOv11+ 检测 → DeepSORT 跟踪 (track_id 本地唯一)
      ↓
  ReID 特征提取 → 本地 FAISS 匹配 → salted hash_id
      ↓
  MQTT 上报 (hash_id + camera_id + track_id + timestamp)

云端 (跨摄像头，仅 hash_id):
  接收多摄像头 hash_id → 按 hash_id 聚合
      ↓
  时间约束检查: 同一 hash_id 不能同时在两个摄像头
      ↓
  摄像头拓扑辅助（预配置摄像头物理位置，推断移动路径）
      ↓
  轨迹存储: hash_id → [(camera_id, timestamp, track_id), ...]
      ↓
  实时查询: 安全员在 Dashboard 查看 hash_id 的当前位置
            （需额外权限验证才能反查 hash_id → 真实姓名）

{
  "camera_topology": {
    "cam_kitchen_01": {
      "zone": "后厨A区",
      "neighbors": ["cam_kitchen_02", "cam_hall_01"],
      "transit_time_sec": [30, 120]
    },
    "cam_site_gate": {
      "zone": "工地入口",
      "neighbors": ["cam_site_road_a"],
      "transit_time_sec": [60, 300]
    }
  }
}

12.4 加密 Gallery 管理与老化机制

12.4.1 人员注册流程（隐私保护版）

sequenceDiagram
    participant Admin as 管理员（离线终端/VPN）
    participant Cloud as 云端注册服务
    participant Edge as 边缘端

    Admin->>Cloud: 1. 上传人员照片（正/侧/背 3张）
    Admin->>Cloud: 2. 填写信息（姓名/工号/角色/有效期）
    Cloud->>Cloud: 3. ReID 特征提取（3张→3×512-d向量）
    Cloud->>Cloud: 4. ⚠️ 提取后立即删除原始照片
    Cloud->>Cloud: 5. 生成 salted hash (person_id + global_salt)
    Cloud->>Cloud: 6. 元数据入库（MySQL，不含原始照片）
    Cloud->>Cloud: 7. 打包加密 Gallery (AES-256-GCM)
    Cloud->>Edge: 8. OTA 增量下发加密 Gallery 包
    Edge->>Edge: 9. 解密 → 内存加载 → 磁盘加密存储
    Edge->>Edge: 10. 热更新本地 FAISS 索引（不停机）
    Edge-->>Cloud: 11. 加载确认

12.4.2 Gallery 老化与 TTL 驱逐机制

设计动机：建筑工地日均流动工人可达 500+ 人，本地 Gallery 上限 500 人无法容纳。 需要智能老化——自动驱逐离场人员，为新入场人员腾出空间。

Gallery 老化策略:

① 基于 TTL 的自动驱逐:
   - 每个人员的特征向量附带 valid_to 时间戳
   - valid_to 到期 → 自动从边缘端 FAISS 索引中移除（云端同步标记为 "expired"）
   - 工地场景: 默认 30 天有效期（工人合同到期）
   - 访客: 默认 24 小时有效期

② 基于活跃度的 LRU 驱逐:
   - 当 Gallery 达到容量上限 (500) 且新人员需要加入时
   - 驱逐「最近 7 天未被任何摄像头匹配到」的最久未活跃人员
   - 驱逐前发送"待确认"通知至管理员，48h 无响应则自动驱逐

③ 班组批量换班:
   - 工地早晚班交替时，批量标记"下班班组"人员为 "inactive"
   - inactive 人员 24h 后自动移除（为下一班组腾空间）
   - 重新上班时，云端自动重新下发

④ 被遗忘权 (Right to Erasure):
   - 管理员在云端发起"人员擦除"请求
   - 云端: 删除 MySQL 元数据（hash_id 反查表除外，保留 30 天审计日志）
   - 边缘端: MQTT 下发擦除指令 → 移除对应向量 → 确认回执
   - 审计: 保留擦除操作日志（操作人、时间、hash_id）至少 6 个月

12.4.3 容量规划

溢出时的降级行为：当人员在 Gallery 外（未注册/已过期/已驱逐），边缘端匹配结果为"unknown"。 云端仅记录"unknown-{track_id}"轨迹，无法关联身份，不触发广播通知（除"未授权闯入"规则）。

12.4.4 加密 Gallery 数据结构 (Protobuf)

// 云端下发的加密 Gallery 包
message EncryptedGalleryPackage {
  string version = 1;             // Gallery 版本号
  int64 generated_at = 2;         // 打包时间 (Unix ms)
  bytes encrypted_payload = 3;    // AES-256-GCM 加密的 GalleryPayload
  bytes gcm_tag = 4;             // GCM 认证标签 (16 bytes)
  bytes iv = 5;                  // 初始化向量 (12 bytes)
  string sha256_checksum = 6;     // 解密后完整性校验
}

// 解密后的 Gallery 数据（仅在边缘端内存中存在）
message GalleryPayload {
  repeated PersonEntry entries = 1;
  
  message PersonEntry {
    string person_id = 1;           // 人员唯一ID
    string hash_id_salt = 2;        // 盐值（用于生成上报 hash）
    repeated bytes features = 3;    // 特征向量（每张照片1个，512-d × 4B）
    string role = 4;                // 角色: manager/worker/visitor
    repeated string zone_ids = 5;   // 可进入的区域ID
    int64 valid_from = 6;           // 有效期起
    int64 valid_to = 7;             // 有效期止（TTL 驱逐依据）
    int64 last_matched_at = 8;      // 最近一次匹配时间（LRU 依据）
    string broadcast_tpl = 9;       // 广播模板（可选）
  }
}

// 边缘端上报消息（仅 64 字节 hash，不含任何生物特征）
message ReIDMatchReport {
  string event_id = 1;
  string device_id = 2;
  string camera_id = 3;
  int32 track_id = 4;
  string salted_hash_id = 5;       // SHA256(person_id + device_salt + date)
  bool is_matched = 6;             // 是否匹配成功
  float confidence = 7;            // 匹配相似度
  int64 timestamp = 8;             // Unix ms
  string alert_context = 9;        // 可选: 附带告警上下文
}

12.5 广播通知联动方案

12.5.1 广播触发规则（基于 hash_id）

注：云端广播规则引擎仅持有 hash_id 与角色的映射（hash_id → {role, zone_permissions}）， 不持有人员真实姓名。广播时通过 hash_id 反查 MySQL 获取姓名（需额外权限校验）。

12.5.2 端到端延迟分析（边缘匹配版）

时间轴（单位：毫秒）— 边缘端全量匹配:

  0ms     边缘端 YOLOv11+ 检测到人员
  8ms     ReID 特征提取完成（OSNet, Orin NX TensorRT FP16）
 12ms     本地 FAISS Flat 检索完成（500人索引，< 2ms）
 15ms     生成 salted hash_id
 20ms     MQTT 上报 hash_id + 轨迹点（QoS=1, < 1KB）
 50ms     云端规则引擎判断触发广播
 80ms     广播指令下发至 PA 系统 API
 200ms    PA 系统播放广播
---------
总延迟：≤ 300ms（设计目标 ≤ 5 秒，实际远优于目标）

对比旧方案（云端匹配）:
  旧方案需上传 2KB 特征向量 → 云端 FAISS 检索 → 匹配返回
  旧方案延迟: 50-80ms 额外网络开销

12.5.3 PA 系统对接方式

POST /api/v1/broadcast/play
Content-Type: application/json
Authorization: Bearer <token>

{
  "request_id": "req_20260620T081500_001",
  "zone_ids": ["zone_kitchen_a", "zone_kitchen_b"],
  "text": "请注意，店长张三已到达后厨，请规范操作。",
  "priority": "high",
  "tts_enabled": true,
  "tts_voice": "zh-CN-XiaoxiaoNeural",
  "repeat": 1,
  "valid_for_sec": 60
}

12.6 核心伪代码（隐私保护版）

12.6.1 边缘端 ReID 全量匹配 + 脱敏上报

class EdgeReIDPrivacyPipeline:
    """
    边缘端 ReID 管线（隐私保护版）
    特征提取 → 本地 FAISS 全量匹配 → salted hash 上报
    NEVER 上传原始特征向量或人员裁剪图
    """
    
    def __init__(self, config: ReIDConfig):
        self.reid_engine = TensorRTEngine(config.model_path)
        self.device_salt = config.device_salt  # 出厂烧录，160bit
        
        # 本地 FAISS 全量索引（从云端加密下发并解密）
        self.faiss_index = None           # FAISS Flat 索引
        self.id_map = {}                  # {faiss_idx: PersonEntry}
        self.person_entries = {}          # {person_id: PersonEntry}
        self._load_encrypted_gallery(config.gallery_path)
        
        self.mqtt = MQTTClient(config.broker, config.device_id)
        self.executor = ThreadPoolExecutor(max_workers=2)
        self.report_cache = TTLCache(max_size=500, ttl_seconds=60)
    
    def _load_encrypted_gallery(self, path: str):
        """加载并解密云端下发的加密 Gallery"""
        with open(path, 'rb') as f:
            encrypted = EncryptedGalleryPackage()
            encrypted.ParseFromString(f.read())
        
        # AES-256-GCM 解密
        plaintext = aes_gcm_decrypt(
            encrypted.encrypted_payload,
            key=self.device_key,
            iv=encrypted.iv,
            tag=encrypted.gcm_tag
        )
        
        # 完整性校验
        assert sha256(plaintext).hexdigest() == encrypted.sha256_checksum
        
        # 解析 Gallery
        gallery = GalleryPayload()
        gallery.ParseFromString(plaintext)
        
        # 构建本地 FAISS 索引
        vectors = []
        for i, entry in enumerate(gallery.entries):
            # TTL 检查：跳过已过期的人员
            if entry.valid_to > 0 and entry.valid_to < time.time():
                continue
            for feat in entry.features:
                vec = np.frombuffer(feat, dtype=np.float32)  # 512-d
                vectors.append(vec)
                self.id_map[len(vectors)-1] = entry
            self.person_entries[entry.person_id] = entry
        
        if vectors:
            data = np.array(vectors, dtype=np.float32)
            self.faiss_index = faiss.IndexFlatIP(512)  # 内积相似度
            self.faiss_index.add(data)
    
    def process_and_report(self, frame, det, frame_ts):
        """处理单个人员检测框"""
        if det.class_name != "person":
            return None
        
        # Step 1: 裁剪 + 特征提取
        person_crop = self._crop_with_margin(frame, det.bbox, margin=0.2)
        tensor = self._preprocess(person_crop)
        feature = self.reid_engine.infer(tensor)
        feature = feature / np.linalg.norm(feature)  # L2 normalize
        
        # Step 2: 本地 FAISS 检索
        matched_person_id = None
        similarity = 0.0
        
        if self.faiss_index and self.faiss_index.ntotal > 0:
            query = feature.reshape(1, -1).astype(np.float32)
            scores, indices = self.faiss_index.search(query, k=5)
            
            threshold = self._get_threshold(det.scene_type)
            for score, idx in zip(scores[0], indices[0]):
                if idx < 0:
                    continue
                if float(score) >= threshold:
                    entry = self.id_map.get(int(idx))
                    if entry:
                        matched_person_id = entry.person_id
                        similarity = float(score)
                        # 更新 LRU 时间戳（用于 Gallery 老化）
                        entry.last_matched_at = int(time.time())
                        break
        
        # Step 3: 生成脱敏 salted hash
        today = datetime.now().strftime("%Y-%m-%d")
        if matched_person_id:
            salted_hash = sha256(
                f"{matched_person_id}:{self.device_salt}:{today}".encode()
            ).hexdigest()  # 64B hash, 单向不可逆
        else:
            salted_hash = None
        
        # Step 4: MQTT 上报（仅 64B hash, 不含特征向量）
        report = {
            "event_id": str(uuid4()),
            "device_id": self.device_id,
            "camera_id": self.camera_id,
            "track_id": det.track_id,
            "salted_hash_id": salted_hash,       # 仅 hash，非原始 person_id
            "is_matched": matched_person_id is not None,
            "confidence": similarity,
            "timestamp": frame_ts
        }
        
        topic = f"edge-vision/{self.project_id}/reid/{self.device_id}/hash/-"
        self.mqtt.publish(topic, json.dumps(report), qos=1)
        
        return matched_person_id
    
    def handle_erasure_command(self, person_id: str):
        """处理云端下发的被遗忘权擦除指令"""
        if person_id in self.person_entries:
            # 从本地索引中移除（需重建索引）
            self._rebuild_index_excluding(person_id)
            del self.person_entries[person_id]
            log.info(f"Person {person_id} erased from local gallery (GDPR Art.17)")
            return True
        return False

12.6.2 云端脱敏轨迹服务（简化版）

class CloudTrajectoryService:
    """
    云端轨迹服务 —— 仅处理脱敏 hash_id
    不执行任何向量匹配操作
    """
    
    def handle_hash_report(self, msg: dict):
        hash_id = msg.get("salted_hash_id")
        
        if not hash_id:
            # 未匹配人员：记录为 anonymous 轨迹
            hash_id = f"anon-{msg['device_id']}-{msg['track_id']}"
        
        # 更新轨迹（Redis）
        key = f"traj:{hash_id}"
        point = json.dumps({
            "camera_id": msg["camera_id"],
            "device_id": msg["device_id"],
            "timestamp": msg["timestamp"],
            "confidence": msg.get("confidence", 0)
        })
        self.redis.rpush(key, point)
        self.redis.ltrim(key, -100, -1)  # 保留最近100点
        self.redis.expire(key, 86400)     # 24h TTL
        
        # 仅当需要反查身份时（如触发广播规则），才通过 hash_id
        # 查 MySQL 反查表（需管理员权限 + 审计日志）
        
    def resolve_identity(self, hash_id: str, 
                         requester: str, audit_reason: str) -> Optional[str]:
        """
        ⚠️ 反查身份（需权限校验 + 审计）
        仅在触发广播/安全事件时调用，全量记录审计日志
        """
        if not self._check_privilege(requester):
            raise PermissionDenied("Identity resolution requires admin role")
        
        # 审计日志
        self._audit_log(
            action="identity_resolution",
            hash_id=hash_id,
            requester=requester,
            reason=audit_reason
        )
        
        # 反查 MySQL
        row = self.db.execute(
            "SELECT person_id, name, role FROM identity_map WHERE hash_id = ?",
            (hash_id,)
        ).fetchone()
        
        return row[0] if row else None

                
            # 设置冷却时间
            self.redis.setex(
                cooldown_key,
                rule.cooldown_seconds,
                "1"
            )
            
            # 记录广播日志
            self._log_broadcast(person_id, rule.id, camera_id, text)
    
    def _eval_rule_condition(self, rule, person, camera, timestamp) -> bool:
        """评估规则条件（简单 DSL 解释器）"""
        cond = rule.condition  # 如: "person.role == 'visitor' AND camera.zone == 'restricted'"
        
        # 简化实现：实际应使用安全的条件评估器（如 expr-eval）
        # 此处仅示意
        if cond.get("person_role") and cond["person_role"] != person.role:
            return False
        if cond.get("zone_id") and cond["zone_id"] != camera.get("zone_id"):
            return False
        return True

12.7 ReID 技术栈

12.8 ReID 非功能需求

12.9 ReID 术语表

13. 环境韧性与传感器退化管理（Environmental Resilience & Sensor Degradation）

设计动机：此前方案将"粉尘影响"视为双目测距的一个瞬时环境条件（要么有粉尘、要么没有）， 严重低估了物理环境对传感器的持续性、累积性、不可逆侵蚀。

在煤矿井下和建筑工地，震动、粉尘、光照突变不是偶发事件——它们是小时级的持续存在。 一台双目相机在矿井安装 3 个月后，其标定参数的漂移量可导致 20m 处测距误差从 ±0.6m 恶化至 ±4m 以上， 而系统对此毫无感知——继续用错误的测距数据计算 TTC 碰撞预警。

13.1 三大场景物理侵蚀向量

核心认知转变：不把粉尘/震动视为"环境条件"，而是视为持续作用的退化力（Degradation Force）。 传感器的初始精度不是常数——它是一个随时间单调递减的函数。

13.2 传感器退化定量模型

13.2.1 双目相机标定漂移模型

标定参数退化函数（经验模型，基于煤矿井下 6 个月实测数据拟合）:

① 基线旋转角 (roll/pitch/yaw) 漂移:
   Δθ(t) = A_v × √(t) + ε
   - A_v: 震动强度系数（煤矿 A_v ≈ 0.08°/√月，工地 A_v ≈ 0.03°/√月）
   - t: 安装后的累计天时（含爆破/打桩当量折算）
   - ε: 温度冲击导致的阶跃分量（每次 ±0.5-1.5°）

② 内参漂移（焦距 f_x, f_y, 主点 c_x, c_y）:
   Δf/f(t) = A_t × ΔT(t) + A_v_f × ln(1 + t/30)
   - A_t: 热胀冷缩系数（铝合金支架 ~23ppm/°C）
   - ΔT: 支架温度与标定温度之差
   - A_v_f: 震动导致镜组微位移系数

③ 测距误差传播:
   由 SGBM/stereo 深度公式 Z = f·B/d:
   ΔZ/Z ≈ √((Δf/f)² + (ΔB/B)² + (Δd/d·Δθ)²)
   
   未检测到的 0.5° roll 旋转在 20m 处可产生:
   ΔZ ≈ 20m × tan(0.5°) ≈ 17.5cm (理想)
   叠加粉尘模糊后: ΔZ ≈ 1.5-3m (实际)

13.2.2 镜头污染衰减模型

图像质量衰减函数（粉尘/油污积累）:

① 透光率衰减:
   T(t) = T₀ × e^(-k_d × t)
   - T₀: 初始透光率（洁净镜头 ≈ 0.95）
   - k_d: 粉尘衰减系数（煤矿 k_d ≈ 0.012/天 → 30天 T=0.66）
   - t: 距上次清洁的天数

② 对比度衰减:
   C(t) = C₀ × T(t)²  (透光率下降 + 散射眩光双重效应)
   
③ YOLO mAP 衰减估计（经验公式）:
   mAP(t) ≈ mAP₀ × (1 - α × (1 - T(t)))
   - α: 检测敏感度系数（小目标 ~0.8，大目标 ~0.4）
   - T=0.66 时 → 小目标 mAP 下降 ~27%，大目标 ~14%

13.2.3 温度传感器退化模型

温度传感器读数若偏低 5°C，在煤矿场景中意味着：设备在 85°C 满负荷运行时系统仍认为仅 80°C， 不会触发降级——直至硬件烧毁。

13.3 双目相机标定漂移检测与自恢复

13.3.1 在线标定漂移检测（不依赖标定板）

在线标定校验流程（边缘端自主执行，每 6 小时触发一次）:

Step 1: 采集一组"静止参考帧"（场景中无运动目标的帧，约 20 帧）
Step 2: 对左右目图像分别提取 ORB 特征点
Step 3: 暴力匹配 + Lowe's ratio test，得到匹配点对集 M
Step 4: 计算基础矩阵 F（RANSAC, 8-point algorithm）
Step 5: 从 F 恢复本质矩阵 E = K_right^T × F × K_left
Step 6: SVD 分解 E → R, t（相对旋转和平移）
Step 7: 对比 R, t 与出厂标定值:
        - ΔR > 0.5° (roll/pitch/yaw 任意轴) → 轻度漂移告警
        - ΔR > 1.5° → 严重漂移告警，降级至单目 TTC
        - Δt/t > 10%（基线长度显著变化）→ 结构变形告警
Step 8: 检测记录写入遥测数据，上报云端

关键前提:
  - 使用出厂标定 K（内参），仅校验外参（R, t）
  - 当场景静止帧不足时跳过该轮检测
  - 不主动修正标定参数（仅检测 + 告警）

class CalibrationDriftDetector:
    """在线标定漂移检测器"""
    
    def __init__(self, calib_file: str):
        calib = np.load(calib_file)
        self.K_left = calib['K_left']
        self.K_right = calib['K_right']
        self.R_calib = calib['R']      # 出厂外参旋转矩阵
        self.t_calib = calib['t']      # 出厂外参平移向量
        self.orb = cv2.ORB_create(nfeatures=2000)
        self.matcher = cv2.BFMatcher(cv2.NORM_HAMMING)
        
        # 告警阈值
        self.DRIFT_WARN_DEG = 0.5      # 轻度漂移
        self.DRIFT_CRITICAL_DEG = 1.5  # 严重漂移
        self.BASELINE_CHANGE_RATIO = 0.10  # 基线变化 10%
    
    def check_drift(self, left_img, right_img) -> DriftReport:
        """执行一次标定漂移检测"""
        # Step 1: ORB 特征提取
        kp_left, des_left = self.orb.detectAndCompute(left_img, None)
        kp_right, des_right = self.orb.detectAndCompute(right_img, None)
        
        if len(kp_left) < 100 or len(kp_right) < 100:
            return DriftReport(status="insufficient_features")
        
        # Step 2: 特征匹配 + Lowe's ratio test
        matches = self.matcher.knnMatch(des_left, des_right, k=2)
        good = [m for m, n in matches if m.distance < 0.75 * n.distance]
        
        if len(good) < 50:
            return DriftReport(status="insufficient_matches")
        
        # Step 3: 计算本质矩阵 E, 恢复 R, t
        pts_left = np.float32([kp_left[m.queryIdx].pt for m in good])
        pts_right = np.float32([kp_right[m.trainIdx].pt for m in good])
        
        E, mask = cv2.findEssentialMat(
            pts_left, pts_right, self.K_left,
            method=cv2.RANSAC, prob=0.999, threshold=1.0)
        
        _, R_current, t_current, _ = cv2.recoverPose(
            E, pts_left, pts_right, self.K_left)
        
        # Step 4: 计算旋转偏差（角度）
        R_diff = R_current @ self.R_calib.T
        angle_deg = np.rad2deg(np.arccos(
            (np.trace(R_diff) - 1) / 2))
        
        # Step 5: 计算基线长度偏差
        baseline_ratio = np.linalg.norm(t_current) / np.linalg.norm(self.t_calib)
        
        # Step 6: 判定
        if angle_deg > self.DRIFT_CRITICAL_DEG or \
           abs(baseline_ratio - 1.0) > self.BASELINE_CHANGE_RATIO:
            return DriftReport(
                status="critical_drift",
                angle_deg=angle_deg,
                baseline_ratio=baseline_ratio,
                action="degrade_to_monocular"
            )
        elif angle_deg > self.DRIFT_WARN_DEG:
            return DriftReport(
                status="mild_drift",
                angle_deg=angle_deg,
                baseline_ratio=baseline_ratio,
                action="notify_maintenance"
            )
        else:
            return DriftReport(status="nominal")

13.3.2 标定恢复策略

13.4 镜头污染检测与清洁调度

13.4.1 基于图像质量的无参考污染检测

镜头污染检测管线（每 30 分钟执行一次，不依赖标签数据）:

指标 1 — 高频细节能量衰减（DCT 域分析）:
  - 对图像的 Y 通道做 8×8 DCT 变换
  - 统计 AC 系数的能量谱
  - 对比初始清洁状态基线，衰减 > 35% → 污染告警

指标 2 — 局部对比度退化（Sobel 梯度统计）:
  - 计算全图 Sobel 梯度幅值
  - 统计梯度幅值 > 阈值的像素占比
  - 洁净镜头: > 15%，污染镜头: < 5%

指标 3 — 暗通道先验退化（Dehazing 指标）:
  - 计算暗通道图（15×15 patch min）
  - 暗通道均值 > 60（0-255）→ 明显 haze/粉尘散射

指标 4 — 清晰度退化（Laplacian 方差）:
  - 计算全图 Laplacian 算子的方差
  - 对比基线方差，衰减 > 50% → 严重污染

综合污染评分:
  P_contamination = 0.35 × P_dct + 0.25 × P_sobel + 0.25 × P_dark_channel + 0.15 × P_laplacian
  综合评分变化 > 40% → 生成清洁工单

class LensContaminationDetector:
    """镜头污染检测器"""
    
    def __init__(self, baseline_stats_path: str):
        # 加载清洁基线（安装时采集）
        self.baseline = self._load_baseline(baseline_stats_path)
        
        # 阈值
        self.DCT_DECAY_WARN = 0.35      # DCT能量衰减 35%
        self.SOBEL_DECAY_WARN = 0.50    # 梯度像素占比衰减 50%
        self.DARK_CHANNEL_WARN = 60     # 暗通道均值 > 60
        self.LAPLACIAN_DECAY_WARN = 0.50
    
    def check(self, frame: np.ndarray) -> ContaminationReport:
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        
        # 指标 1: DCT 高频能量
        dct_energy = self._compute_dct_ac_energy(gray)
        dct_decay = 1.0 - dct_energy / self.baseline['dct_energy']
        
        # 指标 2: Sobel 梯度像素占比
        sobel_ratio = self._compute_sobel_edge_ratio(gray)
        sobel_decay = 1.0 - sobel_ratio / self.baseline['sobel_ratio']
        
        # 指标 3: 暗通道均值（haze 检测）
        dark_channel_mean = self._compute_dark_channel(gray)
        
        # 指标 4: Laplacian 方差
        lap_var = cv2.Laplacian(gray, cv2.CV_64F).var()
        lap_decay = 1.0 - lap_var / self.baseline['laplacian_var']
        
        # 综合污染评分 (0-100)
        score = min(100, (
            35 * min(1.0, dct_decay / self.DCT_DECAY_WARN) +
            25 * min(1.0, sobel_decay / self.SOBEL_DECAY_WARN) +
            25 * min(1.0, dark_channel_mean / self.DARK_CHANNEL_WARN) +
            15 * min(1.0, lap_decay / self.LAPLACIAN_DECAY_WARN)
        ))
        
        action = "none"
        if score > 80:
            action = "urgent_cleaning"  # 污染严重，立即生成工单
        elif score > 50:
            action = "schedule_cleaning"  # 计划清洁
        elif score > 30:
            action = "monitor"  # 轻度污染，持续监控
        
        return ContaminationReport(
            score=score, action=action,
            dct_decay=dct_decay, sobel_decay=sobel_decay,
            dark_channel=dark_channel_mean, laplacian_decay=lap_decay
        )

13.4.2 清洁调度策略

13.5 光照突变免疫策略

13.5.1 光照突变对推理的影响

13.5.2 帧级光照异常检测

class IlluminationAnomalyDetector:
    """光照突变帧级检测器"""
    
    def is_anomalous(self, frame: np.ndarray, prev_frame_stats: dict) -> bool:
        """
        检测当前帧是否受光照突变影响，返回 True 则丢弃该帧（不参与推理）
        """
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        
        # 检测 1: 全局亮度突变（> 50% 变化）
        current_mean = gray.mean()
        if prev_frame_stats:
            delta = abs(current_mean - prev_frame_stats['mean']) / max(prev_frame_stats['mean'], 1)
            if delta > 0.50:
                return True  # 光照突变，丢弃
        
        # 检测 2: 过曝像素占比（> 30% 像素饱和）
        overexposed_ratio = (gray > 240).sum() / gray.size
        if overexposed_ratio > 0.30:
            return True
        
        # 检测 3: 欠曝像素占比（> 40% 像素过暗）
        underexposed_ratio = (gray < 20).sum() / gray.size
        if underexposed_ratio > 0.40:
            return True
        
        # 检测 4: 镜头结雾检测（低对比度 + 高暗通道）
        lap_var = cv2.Laplacian(gray, cv2.CV_64F).var()
        dark_mean = self._dark_channel(gray).mean()
        if lap_var < 50 and dark_mean > 100:
            return True  # 镜头结雾
        
        return False
    
    def on_anomalous_detected(self, consecutive_count: int) -> str:
        """连续异常帧处理"""
        if consecutive_count > 150:  # 5s @ 25FPS
            return "report_lens_fog"    # 大概率镜头结雾，上报清洁工单
        elif consecutive_count > 75:    # 3s
            return "suppress_inference" # 暂停推理，等光照恢复
        else:
            return "skip_frames"        # 短暂异常，跳帧即可

13.5.3 摄像头 AE 锁定策略

针对煤矿井下矿灯频繁扫射导致的 AE 震荡:

策略: 当检测到光照突变频率 > 1次/5秒时（连续 3 次异常检测触发），
      通过 ONVIF 接口临时锁定摄像头 AE（自动曝光）至当前平均值，
      锁定持续 30 秒后恢复自动。

ONVIF AE 锁定命令（通过 python-onvif-zeep）:
  imaging.SetImagingSettings({
      'Exposure': {
          'Mode': 'MANUAL',
          'ExposureTime': current_mean_exposure_time,
          'Gain': current_mean_gain
      }
  })

13.6 传感器健康评分与预测性维护

13.6.1 多维度传感器健康评分卡

传感器健康评分 (Sensor Health Score, SHS, 0-100):

SHS_camera = 0.30 × S_calibration  + 0.25 × S_lens        + 0.20 × S_imaging
           + 0.15 × S_temperature + 0.10 × S_communication

其中:
  S_calibration:   标定漂移健康（基于 ORB 自标定检测的 ΔR 距离告警阈值的边际）
  S_lens:          镜头清洁度（基于 DCT/Laplacian/暗通道综合评分，100-污染评分）
  S_imaging:       成像质量（有效像素率、白平衡稳定性、坏点数量）
  S_temperature:   工作温度健康（距离温度告警阈值的边际）
  S_communication: 通信健康（RTSP 断连率、帧 CRC 错误率）

SHS_temperature_sensor = 0.40 × S_accuracy     + 0.30 × S_stability    + 0.30 × S_corrosion
  其中:
  S_accuracy:   测量精度（交叉校验：与同区域其他温度传感器对比偏差）
  S_stability:  读数稳定性（最近 1h 方差，高方差 = 接触不良）
  S_corrosion:  腐蚀风险（基于安装时长 + 环境湿度 × 酸性因子的退化估计）

13.6.2 预测性维护调度

class PredictiveMaintenanceScheduler:
    """
    预测性维护调度器（云端）
    基于传感器退化趋势预测最佳维护窗口
    """
    
    def predict_next_maintenance(self, sensor_id: str) -> MaintenanceWindow:
        # 获取历史 SHS 序列（TDengine 时序查询）
        shs_history = self._query_shs_history(sensor_id, days=90)
        
        # 线性 + 指数混合退化趋势拟合
        trend = self._fit_degradation_trend(shs_history)
        
        # 预测 SHS 跌破各阈值的时间
        days_to_warn = trend.predict_days_to(threshold=70)
        days_to_critical = trend.predict_days_to(threshold=50)
        days_to_failure = trend.predict_days_to(threshold=30)
        
        # 生成建议维护窗口
        if days_to_warn < 7:
            priority = "urgent"
            suggested_date = today + timedelta(days=max(1, days_to_warn - 2))
        elif days_to_warn < 30:
            priority = "planned"
            # 对齐下次巡检窗口
            suggested_date = self._align_to_patrol_schedule(
                today + timedelta(days=days_to_warn - 5))
        else:
            priority = "routine"
            suggested_date = today + timedelta(days=days_to_warn - 14)
        
        return MaintenanceWindow(
            sensor_id=sensor_id,
            suggested_date=suggested_date,
            priority=priority,
            current_shs=shs_history[-1].score,
            predicted_shs_at_maintenance=trend.predict_at(suggested_date),
            degradation_rate=trend.slope_per_month
        )

13.7 退化感知推理质量监控

13.7.1 推理置信度统计漂移检测

核心思路: YOLO 推理的置信度分布本身是传感器健康的"间接探针"——
          当镜头污染或标定漂移时，推理置信度整体下降（分布左移），
          这不是模型问题，而是传感器退化。

检测方法:
  每小时统计一次 YOLO 输出的置信度分布（按类别）:
  - 计算每个类别的平均置信度 μ_class
  - 与 24h 前的基线对比: Δμ_class / μ_baseline
  - 若多个类别同时下降 > 15% → 传感器退化告警（而非模型问题）

区分传感器退化 vs 模型退化:
  - 传感器退化: 所有类别置信度同时下降（成像质量全局下降）
  - 模型退化: 特定类别置信度下降（该类别训练数据偏差）
  - 环境变化: 置信度昼夜波动（白班 vs 夜班的周期性变化）
  - 传感器退化 + 环境: 非周期性持续下降 → 触发维护工单

13.7.2 退化感知的动态检测阈值

当传感器健康评分 SHS < 70 时，自动调整检测阈值以维持可接受的精度:

adjusted_conf_threshold = base_threshold + (1 - SHS/100) × 0.15

例如:
  - SHS = 95（健康）→ adjusted = 0.55 + 0.008 = 0.558 ≈ 不变
  - SHS = 60（警告）→ adjusted = 0.55 + 0.060 = 0.610  → 提升 11%
  - SHS = 40（严重）→ adjusted = 0.55 + 0.090 = 0.640  → 提升 16%

作用: 传感器退化时自动提高门槛，减少因图像质量下降导致的误报
代价: 检出率可能下降（需要在安全性与误报率之间权衡）

13.8 环境韧性指标

14. 告警质量管理与运营韧性保护（Alert Quality & Operational Resilience）

设计动机：此前方案关注告警的"产生"和"传输"，严重低估了海量低质告警对组织运营韧性的摧毁力。

一个部署了 100 路摄像头、每路每天产生 200 条告警的系统，日均告警量 = 20,000 条。 若误报率 40%（YOLO 在实际场景中并不罕见），则每天 8,000 条是虚假告警。

在安全管理员只有 3-5 人的中型项目里，这意味着每人每小时需处理 200+ 条告警——其中 80 条是假的。 这不是"告警系统"，这是噪音发生器。结果是：

• 第 1 周：操作员认真查看每一条告警
• 第 2 周：开始批量确认（不细看）
• 第 3 周：系统被静音 / 屏幕最小化 / 告警规则全关
• 第 4 周：真实事故发生时无人响应——系统已死。

14.1 告警疲劳：运营韧性的系统性威胁

14.1.1 告警疲劳的定量模型

告警疲劳度 (Alert Fatigue Score, AFS, 0-100):

AFS = 0.40 × S_overload + 0.35 × S_false_rate + 0.25 × S_response_decay

其中:
  S_overload:        过载度 = min(100, 实际告警量 / 可处理告警量 × 100)
                     每人可处理告警量: 安全场景 ≈ 8条/小时，非安全 ≈ 20条/小时
                     
  S_false_rate:      误报伤害 = 累计误报率 × 100
                     累计误报率: 最近 7 天内被标记为"误报"的告警占比
                     
  S_response_decay:  响应衰退 = (1 - 最近响应率) × 100
                     响应率: 最近 24h 内被操作员确认/处理的告警占比

AFS 解释:
  0-20:   健康 — 操作员积极响应，告警受信任
  20-40:  关注 — 开始出现批量确认行为
  40-60:  危险 — 告警被忽视率 > 40%，需要人工介入治理
  60-80:  严重 — 系统可信度崩塌，操作员"狼来了"心理
  80-100: 失效 — 真实告警无人响应，安全管控实质失效

触发自动治理的阈值:
  AFS > 40: 自动启用告警聚合 (合并同类告警)
  AFS > 55: 自动提升置信度阈值 (+0.10)
  AFS > 70: 自动暂停非 CRITICAL 级别告警推送，仅保留安全态势报告
  AFS < 25 且持续 24h: 逐步恢复标准告警策略

14.1.2 告警容量规划

告警预算：将告警视为有限资源——每个项目有"每日告警预算"(Daily Alert Budget, DAB)。 当实际告警量超过 DAB 时，自动启动质量优先的告警削减策略。 DAB = 操作员人数 × 8小时 × 10条/小时 × 安全系数 0.8

14.2 告警质量评分框架

14.2.1 单条告警质量评分 (AQI — Alert Quality Index)

AQI = 0.35 × S_confidence + 0.25 × S_temporal + 0.20 × S_spatial + 0.20 × S_historical

① S_confidence: 置信度得分
   = detection_confidence / 0.95 (上限 1.0)
   置信度 0.55 → 0.58，置信度 0.90 → 0.95

② S_temporal: 时域持续性得分
   = min(1.0, consecutive_frames / required_debounce_frames)
   连续触发 3 帧（刚好去抖）→ 0.60，连续 10 帧 → 1.0

③ S_spatial: 空间合理性得分
   基于场景规则评估告警的"空间合理性":
   - 禁区入侵: 人员距禁区边界 < 1m → 0.5（可能在边界），> 3m → 1.0（明确入侵）
   - 未戴安全帽: bbox 头部区域可见性 → 遮挡 < 30% → 1.0，遮挡 > 70% → 0.3

④ S_historical: 历史可信度得分
   = 1.0 - 该告警类型过去 7 天被标记误报的比例
   历史误报率 5% → 0.95，历史误报率 40% → 0.60

AQI 分级:
  AQI ≥ 0.80: HIGH — 高可信告警，立即推送
  0.50 ≤ AQI < 0.80: MEDIUM — 中等，推送但降低通知优先级
  AQI < 0.50: LOW — 低质告警，仅记录不推送（或聚合后推送）

14.2.2 告警类型可信度档案

每种告警类型维护一个滚动 30 天的"可信度档案":

{
  "alert_type": "no_helmet",
  "total_alerts_30d": 12450,
  "confirmed_true": 8230,        // 操作员确认真实
  "confirmed_false": 3120,        // 操作员标记误报
  "unreviewed": 1100,            // 未被审核
  "false_positive_rate_30d": 0.275,  // 27.5% 误报率
  "credibility_trend": "declining",   // 上升/下降/稳定
  "auto_suppression_threshold": 0.40, // 误报率 > 40% → 自动抑制
  "recommended_action": "review_rules" // 建议审查告警规则
}

当告警类型的 30 日误报率超过:
  - 30%: 云端 Dashboard 置顶提醒
  - 40%: 自动将该类型非 CRITICAL 告警降级为 INFO
  - 50%: 自动暂停该类型告警 24h，通知管理员审查规则

14.3 多级告警抑制引擎

14.3.1 四级抑制策略

Level 1 — 帧级去抖 (Frame-level Debounce, 边缘端):
  同一 track_id + alert_type 连续 N 帧才触发
  CRITICAL: N=1 (不等待)
  MAJOR: N=3
  MINOR: N=5
  INFO: N=10

Level 2 — 频率限制 (Rate Limiting, 边缘端):
  同一摄像头 + 告警类型在时间窗口 W 内最多发出 M 条:
  ┌───────────────┬──────────┬──────────┐
  │ 告警等级       │ 窗口 W   │ 上限 M   │
  ├───────────────┼──────────┼──────────┤
  │ CRITICAL      │ 1 分钟   │ 无上限    │
  │ MAJOR         │ 1 分钟   │ 10 条     │
  │ MINOR         │ 5 分钟   │ 15 条     │
  │ INFO          │ 10 分钟  │ 20 条     │
  └───────────────┴──────────┴──────────┘
  超过上限的告警: 本地记录，每小时汇总为一条"聚合告警"上报

Level 3 — 时间聚合 (Temporal Aggregation, 边缘端):
  同一摄像头 + 告警类型在短时间窗口内合并:
  - 30 秒内同一类型的多条告警 → 合并为 1 条聚合告警
  - 聚合告警包含:
    * 触发次数
    * 涉及的 track_id 列表
    * 时间段
    * 代表性截图（置信度最高的一张）

Level 4 — 全局质量门控 (Global Quality Gate, 边缘端 + 云端协同):
  当任意以下条件满足时，边缘端自动提升全局告警门槛:
  - 最近 1 小时告警量 > 历史同时段均值 × 3
  - 边缘端 AFS > 40（由云端推送）
  - 传感器 SHS < 60（成像质量下降 → 当前告警可信度系统性降低）
  
  提升策略:
  - 非 CRITICAL 告警置信度阈值 +0.08
  - MINOR/INFO 级别告警暂停推送（仅本地记录）
  - 触发后持续 30 分钟，之后逐级恢复

14.3.2 告警抑制伪代码（边缘端）

class AlertSuppressionEngine:
    """边缘端多级告警抑制引擎"""
    
    def __init__(self, config: SuppressionConfig):
        # Level 1: 去抖状态
        self._debounce_state: Dict[str, int] = {}   # key → consecutive_count
        
        # Level 2: 频率限制滑动窗口
        self._rate_window: Dict[str, deque] = defaultdict(
            lambda: deque(maxlen=1000))  # camera_alerttype → [timestamps]
        
        # Level 3: 时间聚合缓冲
        self._aggregation_buffer: Dict[str, list] = defaultdict(list)
        
        # Level 4: 全局质量门控
        self._global_suppression = False
        self._alert_count_1h = 0
        self._hourly_baseline = config.hourly_baseline
        
        # 告警类型可信度（由云端定期推送）
        self._credibility: Dict[str, float] = {}  # alert_type → fp_rate_30d
    
    def should_emit(self, alert: AlertProposal) -> EmitDecision:
        """判断一条告警是否应该发送"""
        alert_key = f"{alert.camera_id}_{alert.alert_type}"
        now = time.time()
        
        # === Level 1: 帧级去抖 ===
        debounce_key = f"{alert.track_id}_{alert.alert_type}"
        if alert.level != AlertLevel.CRITICAL:
            required = {MAJOR: 3, MINOR: 5, INFO: 10}.get(alert.level, 3)
            count = self._debounce_state.get(debounce_key, 0) + 1
            self._debounce_state[debounce_key] = count
            if count < required:
                return EmitDecision(action="defer", reason="debounce")
            self._debounce_state[debounce_key] = 0
        
        # === Level 2: 频率限制 ===
        limits = {
            CRITICAL: (60, 999),    # 1min, 无上限
            MAJOR:    (60, 10),     # 1min, 10条
            MINOR:    (300, 15),    # 5min, 15条
            INFO:     (600, 20),    # 10min, 20条
        }
        window, limit = limits[alert.level]
        
        # 滑动窗口清理
        recent = [ts for ts in self._rate_window[alert_key] 
                  if now - ts < window]
        self._rate_window[alert_key] = deque(recent, maxlen=1000)
        
        if len(recent) >= limit:
            # 超限 → 记录到聚合缓冲，不立即发送
            self._aggregation_buffer[alert_key].append(alert)
            return EmitDecision(action="buffer", reason="rate_limit")
        
        self._rate_window[alert_key].append(now)
        
        # === Level 3: 聚合缓冲刷新 ===
        # 每 30 秒检查一次是否需要发送聚合告警
        if self._should_flush_aggregation(alert_key, now):
            return self._build_aggregated_alert(alert_key)
        
        # === Level 4: 全局质量门控 ===
        if self._global_suppression and alert.level in [MINOR, INFO]:
            self._aggregation_buffer[alert_key].append(alert)
            return EmitDecision(action="buffer", reason="global_suppression")
        
        if self._global_suppression and alert.level == MAJOR:
            # 提升置信度阈值
            if alert.confidence < self._adjusted_threshold:
                return EmitDecision(action="defer", reason="quality_gate")
        
        # === AQI 评估 ===
        aqi = self._compute_aqi(alert)
        if aqi < 0.50:
            return EmitDecision(action="log_only", reason=f"low_aqi={aqi:.2f}")
        
        return EmitDecision(action="emit", aqi=aqi)
    
    def update_credibility_from_cloud(self, credibility_data: dict):
        """云端推送的最新告警类型可信度"""
        self._credibility = credibility_data
        
        # 自动抑制误报率过高类型
        for alert_type, fp_rate in credibility_data.items():
            if fp_rate > 0.50:
                self._auto_suppress(alert_type, duration_hours=24)
    
    def _auto_suppress(self, alert_type: str, duration_hours: int):
        """自动抑制特定告警类型"""
        log.warning(
            f"Auto-suppressing alert type '{alert_type}' "
            f"for {duration_hours}h due to false positive rate > 50%"
        )
        # 下发抑制指令到合规引擎

14.4 时间聚合与事件关联

14.4.1 聚合告警格式

// 聚合告警（替代多条独立告警）
{
  "event_id": "agg_20260620T140530_cam01_no_helmet",
  "event_type": "aggregated_alert",
  "alert_type": "no_helmet",
  "alert_level": "MAJOR",
  "camera_id": "cam_site_entrance",
  "time_window": {
    "start": "2026-06-20T14:03:15Z",
    "end": "2026-06-20T14:05:22Z",
    "duration_sec": 127
  },
  "aggregation": {
    "total_triggers": 15,
    "unique_tracks": 8,
    "peak_concurrent": 5,
    "representative_snapshot_url": "https://oss.../agg_xxx.jpg",
    "credibility": 0.82
  },
  "suppressed_details": [
    // 原始告警被折叠，仅保留摘要
  ]
}

14.4.2 跨摄像头事件关联

当以下模式出现时，将多个独立的告警关联为一个"安全事件":

关联规则:
  ① 时间邻近: 5 分钟内
  ② 空间邻近: 相邻摄像头（摄像头拓扑图中 neighbor 距离 ≤ 2 跳）
  ③ 类型关联:
     - smoke + fire 在相邻摄像头上 → 可能火灾蔓延
     - person_fall + no_helmet 同一 track_id 跨摄像头 → 跌倒连带违规
     - person + danger_zone 多个摄像头上 → 可能是群体闯入

关联后行为:
  - 升级优先级: 3 个以上独立告警被关联 → 升级为 CRITICAL 安全事件
  - 生成事件摘要: 自动生成自然语言描述（"工地A区/入口B至塔吊C区域，连续5分钟检测到8人未戴安全帽"）
  - 减少推送条数: 15 条独立告警 → 1 条安全事件 + 1 条聚合告警 = 2 条

14.5 动态置信度阈值

14.5.1 自适应阈值控制器

动态阈值调度策略（边缘端每 15 分钟评估一次）:

输入:
  - 最近 1h 告警量
  - 当前操作员在线数
  - 传感器健康评分 SHS
  - 当前时间（日间/夜间，高峰/低谷）

调度逻辑:
  adjusted_threshold = base_threshold 
                     + Δ_volume       # 告警量调整
                     + Δ_operator     # 人力调整
                     + Δ_sensor       # 传感器健康调整
                     + Δ_period       # 时段调整

① Δ_volume (告警量驱动):
   recent_1h / baseline_1h:
     < 0.5:    Δ = -0.05  (告警少，降低门槛不漏报)
     0.5-1.5:  Δ = 0
     > 1.5:    Δ = +0.05  (告警多，提升门槛减误报)
     > 3.0:    Δ = +0.10  (告警洪峰，大幅提升门槛)

② Δ_operator (人力驱动):
   在线操作员数 < 配置最低人数: Δ = +0.05

③ Δ_sensor (传感器健康驱动):
   SHS < 60:  Δ = +0.08  (传感器退化 → 置信度系统性偏低 → 提升阈值)

④ Δ_period (时段驱动):
   夜间(22:00-06:00): Δ = -0.03  (人员少、风险高，适当降低门槛)

最终阈值约束:
  CRITICAL 级别: 下界 0.50，上界 0.75
  MAJOR 级别:   下界 0.55，上界 0.80
  MINOR 级别:   下界 0.60，上界 0.85

14.6 操作员反馈闭环

14.6.1 快速反馈通道

操作员对每条告警可进行 3 种反馈（云端 UI，一键操作）:

① ✓ 确认 (Acknowledge): 告警正确
   效果: 确认人员、确认时间戳入库
   影响: 该告警类型 30 日可信度 +0.001

② ✗ 误报 (False Positive): 告警错误
   效果: 
   - 记录误报原因标签（遮挡/光照/模型错误/配置错误）
   - 截图 + 前后帧自动打包 → 难例池（优先级最高）
   影响: 该告警类型 30 日可信度 -0.005

③ ⊗ 无意义 (Noise): 不是误报但无运营价值
   示例: "后厨检测到 person"（这是正常的，不需要告警这）
   效果: 触发规则审查建议（该告警类型在此区域可能不需要）
   影响: 该地点+告警类型组合的可信度 -0.003

14.6.2 反馈驱动的自动化规则调整

闭环生效机制（云端，每小时执行一次）:

Step 1: 统计各告警类型的近 7 日误报率
Step 2: 误报率 > 30% 的告警类型:
         → 自动分析误报原因标签分布
         → 生成"告警规则优化建议"（自动草拟配置变更）
Step 3: 误报率 > 40% 的告警类型:
         → 自动提升该类型告警的边缘端 AQI 最低要求（+0.10）
         → 自动提升该类型告警的非 CRITICAL 变体的置信度阈值（+0.08）
Step 4: 误报率 > 50% 的告警类型:
         → 自动暂停该类型的非 CRITICAL 告警 24h
         → 通知管理员审查规则并确认恢复
Step 5: 将调整后的阈值/规则通过 MQTT 下发至边缘端（无需重启）

人工复核保留:
  - 所有自动调整需记录审计日志（操作人 = "system", 调整详情 + 依据）
  - 管理员可随时回滚任何自动调整
  - 连续 2 次自动暂停同一告警类型 → 不再自动恢复，等待人工决策

14.7 静默窗口与维护模式

预配置的告警静默规则:

① 计划维护窗口:
   - 管理员在云端配置: "2026-06-25 02:00-05:00，全项目静默"
   - 边缘端收到静默指令后:
     · CRITICAL 级别告警: 照常发送（安全第一）
     · MAJOR/MINOR: 本地记录，窗口结束后汇总推送
     · INFO: 完全静默（不记录也不推送）

② 常规班次静默:
   - 非工作时间 (如 22:00-06:00):
     · MINOR/INFO 告警自动静默
     · 仅推送 MAJOR/CRITICAL

③ 节假日静默:
   - 预设节假日日历，自动降低非关键区域告警频率

④ 环境静默:
   - 当检测到极端天气/施工爆破时段:
     · 大量人员临时离场 → 自动抑制"区域无人"类告警
     · 爆破震动 → 自动抑制"摄像机异常"类告警（5分钟）

14.8 告警升级策略

告警升级 (Escalation): 当关键告警被忽视时，自动升级通知方式

升级链:
  Level 0 (即时):  Dashboard 推送 + 声光提示
  Level 1 (60s 无确认):  Dashboard 醒目闪烁 + 区域内 PA 提示音
  Level 2 (180s 无确认):  短信/钉钉/企微通知值班组长
  Level 3 (300s 无确认):  电话语音通知安全总监
  Level 4 (600s 无确认):  启动应急预案（触发消防/医疗联动，如已对接）

升级条件:
  - 仅 CRITICAL 告警会产生升级链
  - 若操作员在 N 秒内点击"确认"或"误报"，中断升级
  - 夜间 (22:00-06:00) 升级间隔减半（响应时间缩短）

升级抑制:
  - 若操作员已经确认了同一位置/类型的告警，后续相同告警不升级
  - 若系统处于维护模式（静默窗口），不触发升级

14.9 运营韧性指标

14.9.1 告警系统健康指标体系

14.9.2 运营韧性恢复 SLA

14.9.3 告警治理 Dashboard（新增）

云端运营韧性监控面板 (Alert Governance Dashboard):

面板 1 — 告警态势:
  - 实时告警瀑布流（按 AQI 排序，高 AQI 置顶）
  - 当前 AFS (告警疲劳度) 仪表盘
  - 剩余告警预算条 (DAB - 今日已消耗)

面板 2 — 质量分布:
  - 各告警类型可信度趋势图（30 天滚动）
  - 误报率 Top-5 告警类型（红色高亮）
  - 自动抑制状态一览

面板 3 — 操作员效能:
  - 各操作员确认率 / 平均延迟
  - 反馈活跃度（标记误报频率）
  - 班次告警负载分布

面板 4 — 传感器关联:
  - SHS vs 告警量散点图（退化传感器是否在制造噪音）
  - SHS < 60 设备的告警量趋势
  - 预测性维护提醒

文档结束

本方案书 v3.0 新增环境韧性与传感器退化管理（Section 13）和告警质量管理与运营韧性保护（Section 14）。 所有 Mermaid 图表可在支持 Mermaid 渲染的 Markdown 编辑器中查看。