MediaPipe 系列 49：IMS OMS 架构——安全带检测完整指南

前言：安全带检测的重要性

49.1 Euro NCAP 2026 要求

┌─────────────────────────────────────────────────────────────────────────┐
│                    Euro NCAP 2026 安全带检测要求                         │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│   检测场景：                                                             │
│   ├── 安全带未系（Unfastened）                                          │
│   ├── 安全带错误佩戴（Belt Misuse）                                     │
│   │   ├── 安全带在手臂下方（Under-arm）                                 │
│   │   ├── 安全带在背后（Behind-back）                                   │
│   │   └── 安全带松弛（Loose）                                           │
│   └── 儿童安全座椅安全带                                                │
│                                                                         │
│   检测要求：                                                             │
│   ├── 检测时间：< 2 秒                                                   │
│   ├── 准确率：> 95%                                                      │
│   └── 多座位：前排 + 后排                                               │
│                                                                         │
│   告警策略：                                                             │
│   ├── 未系：立即告警                                                     │
│   ├── 错误佩戴：立即告警                                                 │
│   └── 后排：仪表盘提示                                                   │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

49.2 安全带检测分类

类型	描述	风险等级	检测难度
未系	安全带未扣合	🔴 高	低
Under-arm	安全带在手臂下方	🔴 高	高
Behind-back	安全带在背后	🔴 高	中
Loose	安全带松弛	🟡 中	中
正常佩戴	安全带正确佩戴	🟢 无	-

---

五十四、检测方法

54.1 方法概述

┌─────────────────────────────────────────────────────────────────────────┐
│                    安全带检测方法                                         │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│   方法一：视觉检测                                                       │
│   ┌─────────────────────────────────────────────────────┐              │
│   │ • 检测安全带外观特征                                │              │
│   │ • 关键点：肩带、腰带、锁扣                          │              │
│   │ • 结合人体姿态判断                                  │              │
│   │                                                     │              │
│   │ 优点：无需车辆信号                                  │              │
│   │ 缺点：遮挡场景困难                                  │              │
│   └─────────────────────────────────────────────────────┘              │
│                                                                         │
│   方法二：车辆信号 + 视觉融合                                           │
│   ┌─────────────────────────────────────────────────────┐              │
│   │ • 车辆信号：安全带锁扣状态                          │              │
│   │ • 视觉：检测是否正确佩戴                            │              │
│   │ • 融合：提高准确率                                  │              │
│   │                                                     │              │
│   │ 优点：准确率高                                      │              │
│   │ 缺点：依赖车辆 CAN 信号                             │              │
│   └─────────────────────────────────────────────────────┘              │
│                                                                         │
│   方法三：深度学习分类                                                   │
│   ┌─────────────────────────────────────────────────────┐              │
│   │ • 输入：乘员图像                                    │              │
│   │ • 输出：安全带状态分类                              │              │
│   │ • 端到端训练                                        │              │
│   │                                                     │              │
│   │ 优点：泛化能力强                                    │              │
│   │ 缺点：需要大量标注数据                              │              │
│   └─────────────────────────────────────────────────────┘              │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

54.2 安全带关键点

┌─────────────────────────────────────────────────────────────┐
│                    安全带关键点定义                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   人体关键点：                                               │
│   ├── 肩膀（左/右）                                         │
│   ├── 胸部中心                                             │
│   └── 腰部两侧                                             │
│                                                             │
│   安全带关键点：                                             │
│   ├── 肩带（Shoulder Belt）                                │
│   │   ├── 上端点（靠近肩膀）                               │
│   │   └── 下端点（靠近腰部）                               │
│   ├── 腰带（Lap Belt）                                     │
│   │   ├── 左端点                                          │
│   │   └── 右端点                                          │
│   └── 锁扣（Buckle）                                       │
│                                                             │
│   检测逻辑：                                                 │
│   ├── 肩带是否从肩膀斜跨到腰部                             │
│   ├── 腰带是否在髋部位置                                   │
│   └── 安全带是否贴合身体                                   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

五十五、Seatbelt Detection Calculator

55.1 完整 Calculator

// seatbelt_detection_calculator.h
#ifndef MEDIAPIPE_CALCULATORS_OMS_SEATBELT_DETECTION_H_
#define MEDIAPIPE_CALCULATORS_OMS_SEATBELT_DETECTION_H_

#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/framework/formats/landmark.pb.h"

namespace mediapipe {

// ========== 安全带状态消息 ==========
message SeatbeltStatus {
  enum State {
    UNKNOWN = 0;
    FASTENED = 1;        // 正常佩戴
    UNFASTENED = 2;      // 未系
    UNDER_ARM = 3;       // 手臂下方
    BEHIND_BACK = 4;     // 背后
    LOOSE = 5;           // 松弛
  }
  
  State state = 1;
  float confidence = 2;
  
  // 各座位状态
  repeated bool seat_fastened = 3;  // [FL, FR, RL, RR]
  
  uint64 timestamp_ms = 4;
}

// ========== Seatbelt Detection Calculator ==========
class SeatbeltDetectionCalculator : public CalculatorBase {
 public:
  static absl::Status GetContract(CalculatorContract* cc) {
    cc->Inputs().Tag("POSE_LANDMARKS").Set<std::vector<NormalizedLandmarkList>>();
    cc->Inputs().Tag("SEATBELT_LANDMARKS").Set<std::vector<NormalizedLandmarkList>>();
    cc->Inputs().Tag("IMAGE").Set<ImageFrame>();
    cc->Outputs().Tag("SEATBELT_STATUS").Set<SeatbeltStatus>();
    cc->Outputs().Tag("ALERT").Set<bool>();
    cc->Options<SeatbeltDetectionOptions>();
    return absl::OkStatus();
  }

  absl::Status Open(CalculatorContext* cc) override {
    const auto& options = cc->Options<SeatbeltDetectionOptions>();
    
    // 阈值配置
    shoulder_belt_threshold_ = options.shoulder_belt_threshold();
    lap_belt_threshold_ = options.lap_belt_threshold();
    loose_threshold_ = options.loose_threshold();
    
    return absl::OkStatus();
  }

  absl::Status Process(CalculatorContext* cc) override {
    SeatbeltStatus status;
    status.set_timestamp_ms(cc->InputTimestamp().Value() / 1000);
    
    // ========== 获取人体关键点 ==========
    if (cc->Inputs().Tag("POSE_LANDMARKS").IsEmpty()) {
      status.set_state(SeatbeltStatus::UNKNOWN);
      cc->Outputs().Tag("SEATBELT_STATUS").AddPacket(
          MakePacket<SeatbeltStatus>(status).At(cc->InputTimestamp()));
      return absl::OkStatus();
    }
    
    const auto& poses = 
        cc->Inputs().Tag("POSE_LANDMARKS").Get<std::vector<NormalizedLandmarkList>>();
    
    if (poses.empty()) {
      status.set_state(SeatbeltStatus::UNKNOWN);
      cc->Outputs().Tag("SEATBELT_STATUS").AddPacket(
          MakePacket<SeatbeltStatus>(status).At(cc->InputTimestamp()));
      return absl::OkStatus();
    }
    
    const auto& pose = poses[0];
    
    // ========== 获取安全带关键点 ==========
    std::vector<Point3D> belt_points;
    if (!cc->Inputs().Tag("SEATBELT_LANDMARKS").IsEmpty()) {
      const auto& belts = 
          cc->Inputs().Tag("SEATBELT_LANDMARKS").Get<std::vector<NormalizedLandmarkList>>();
      if (!belts.empty()) {
        for (int i = 0; i < belts[0].landmark_size(); ++i) {
          Point3D p;
          p.x = belts[0].landmark(i).x();
          p.y = belts[0].landmark(i).y();
          p.z = belts[0].landmark(i).z();
          belt_points.push_back(p);
        }
      }
    }
    
    // ========== 提取人体关键点 ==========
    Point3D left_shoulder = GetPoint(pose, 11);
    Point3D right_shoulder = GetPoint(pose, 12);
    Point3D left_hip = GetPoint(pose, 23);
    Point3D right_hip = GetPoint(pose, 24);
    Point3D chest = GetChestCenter(left_shoulder, right_shoulder);
    
    // ========== 安全带检测 ==========
    SeatbeltStatus::State detected_state = SeatbeltStatus::UNKNOWN;
    float confidence = 0.0f;
    
    if (belt_points.empty()) {
      // 未检测到安全带 → 未系
      detected_state = SeatbeltStatus::UNFASTENED;
      confidence = 0.8f;
    } else {
      // ========== 检测肩带 ==========
      bool shoulder_belt_ok = CheckShoulderBelt(
          belt_points, left_shoulder, right_shoulder, chest);
      
      // ========== 检测腰带 ==========
      bool lap_belt_ok = CheckLapBelt(
          belt_points, left_hip, right_hip);
      
      // ========== 检测错误佩戴 ==========
      bool under_arm = CheckUnderArm(belt_points, left_shoulder, right_shoulder);
      bool behind_back = CheckBehindBack(belt_points, chest);
      bool loose = CheckLoose(belt_points, left_shoulder, right_shoulder);
      
      // ========== 状态判断 ==========
      if (under_arm) {
        detected_state = SeatbeltStatus::UNDER_ARM;
        confidence = 0.7f;
      } else if (behind_back) {
        detected_state = SeatbeltStatus::BEHIND_BACK;
        confidence = 0.7f;
      } else if (loose) {
        detected_state = SeatbeltStatus::LOOSE;
        confidence = 0.6f;
      } else if (shoulder_belt_ok && lap_belt_ok) {
        detected_state = SeatbeltStatus::FASTENED;
        confidence = 0.9f;
      } else {
        detected_state = SeatbeltStatus::UNFASTENED;
        confidence = 0.6f;
      }
    }
    
    status.set_state(detected_state);
    status.set_confidence(confidence);
    
    // ========== 告警判断 ==========
    bool alert = (detected_state == SeatbeltStatus::UNFASTENED ||
                  detected_state == SeatbeltStatus::UNDER_ARM ||
                  detected_state == SeatbeltStatus::BEHIND_BACK);
    
    cc->Outputs().Tag("SEATBELT_STATUS").AddPacket(
        MakePacket<SeatbeltStatus>(status).At(cc->InputTimestamp()));
    cc->Outputs().Tag("ALERT").AddPacket(
        MakePacket<bool>(alert).At(cc->InputTimestamp()));
    
    return absl::OkStatus();
  }

 private:
  float shoulder_belt_threshold_ = 0.2f;
  float lap_belt_threshold_ = 0.15f;
  float loose_threshold_ = 0.1f;
  
  Point3D GetPoint(const NormalizedLandmarkList& landmarks, int index) {
    Point3D p;
    if (index < landmarks.landmark_size()) {
      p.x = landmarks.landmark(index).x();
      p.y = landmarks.landmark(index).y();
      p.z = landmarks.landmark(index).z();
    }
    return p;
  }
  
  Point3D GetChestCenter(const Point3D& left_shoulder, 
                         const Point3D& right_shoulder) {
    Point3D center;
    center.x = (left_shoulder.x + right_shoulder.x) / 2.0f;
    center.y = (left_shoulder.y + right_shoulder.y) / 2.0f;
    center.z = (left_shoulder.z + right_shoulder.z) / 2.0f;
    return center;
  }
  
  bool CheckShoulderBelt(const std::vector<Point3D>& belt_points,
                         const Point3D& left_shoulder,
                         const Point3D& right_shoulder,
                         const Point3D& chest) {
    // 检测肩带是否从肩膀斜跨到腰部
    // 肩带应该穿过胸部
    
    for (const auto& belt_point : belt_points) {
      // 检查安全带点是否在肩膀和胸部之间
      float dist_to_chest = Distance2D(belt_point, chest);
      
      if (dist_to_chest < shoulder_belt_threshold_) {
        return true;
      }
    }
    
    return false;
  }
  
  bool CheckLapBelt(const std::vector<Point3D>& belt_points,
                    const Point3D& left_hip,
                    const Point3D& right_hip) {
    // 检测腰带是否在髋部位置
    
    Point3D hip_center;
    hip_center.x = (left_hip.x + right_hip.x) / 2.0f;
    hip_center.y = (left_hip.y + right_hip.y) / 2.0f;
    
    for (const auto& belt_point : belt_points) {
      float dist_to_hip = Distance2D(belt_point, hip_center);
      
      if (dist_to_hip < lap_belt_threshold_) {
        return true;
      }
    }
    
    return false;
  }
  
  bool CheckUnderArm(const std::vector<Point3D>& belt_points,
                     const Point3D& left_shoulder,
                     const Point3D& right_shoulder) {
    // 检测安全带是否在手臂下方
    // 安全带应该在肩膀上方
    
    for (const auto& belt_point : belt_points) {
      // 如果安全带点在肩膀以下且靠近腋下
      float dist_to_left = Distance2D(belt_point, left_shoulder);
      float dist_to_right = Distance2D(belt_point, right_shoulder);
      
      if (belt_point.y > left_shoulder.y + 0.1f && dist_to_left < 0.2f) {
        return true;  // 左侧 under-arm
      }
      if (belt_point.y > right_shoulder.y + 0.1f && dist_to_right < 0.2f) {
        return true;  // 右侧 under-arm
      }
    }
    
    return false;
  }
  
  bool CheckBehindBack(const std::vector<Point3D>& belt_points,
                       const Point3D& chest) {
    // 检测安全带是否在背后
    // 使用深度信息（Z 坐标）
    
    for (const auto& belt_point : belt_points) {
      // 如果安全带点的 Z 值小于胸部（更远离摄像头）
      // 且在胸部附近，说明可能在背后
      if (belt_point.z < chest.z - 0.05f) {
        return true;
      }
    }
    
    return false;
  }
  
  bool CheckLoose(const std::vector<Point3D>& belt_points,
                  const Point3D& left_shoulder,
                  const Point3D& right_shoulder) {
    // 检测安全带是否松弛
    // 松弛的安全带会远离身体
    
    for (const auto& belt_point : belt_points) {
      float dist_to_shoulder = std::min(
          Distance2D(belt_point, left_shoulder),
          Distance2D(belt_point, right_shoulder));
      
      if (dist_to_shoulder > loose_threshold_) {
        return true;
      }
    }
    
    return false;
  }
  
  float Distance2D(const Point3D& a, const Point3D& b) {
    return std::sqrt(std::pow(a.x - b.x, 2) + std::pow(a.y - b.y, 2));
  }
};

REGISTER_CALCULATOR(SeatbeltDetectionCalculator);

}  // namespace mediapipe

#endif  // MEDIAPIPE_CALCULATORS_OMS_SEATBELT_DETECTION_H_

---

五十六、Graph 配置

# ims_oms_seatbelt_detection_graph.pbtxt

input_stream: "OMS_IMAGE:oms_image"
input_stream: "CAN_SIGNAL:can_signal"
output_stream: "SEATBELT_STATUS:seatbelt_status"
output_stream: "ALERT:alert"

# ========== 1. 人体姿态检测 ==========
node {
  calculator: "PoseDetectionCalculator"
  input_stream: "IMAGE:oms_image"
  output_stream: "POSE_LANDMARKS:pose_landmarks"
  options {
    [mediapipe.PoseDetectorOptions.ext] {
      model_complexity: 1
      min_detection_confidence: 0.5
    }
  }
}

# ========== 2. 安全带检测 ==========
node {
  calculator: "SeatbeltLandmarkCalculator"
  input_stream: "IMAGE:oms_image"
  input_stream: "POSE_LANDMARKS:pose_landmarks"
  output_stream: "SEATBELT_LANDMARKS:seatbelt_landmarks"
  options {
    [mediapipe.SeatbeltLandmarkOptions.ext] {
      model_path: "/models/seatbelt_landmark.tflite"
    }
  }
}

# ========== 3. 安全带状态判断 ==========
node {
  calculator: "SeatbeltDetectionCalculator"
  input_stream: "POSE_LANDMARKS:pose_landmarks"
  input_stream: "SEATBELT_LANDMARKS:seatbelt_landmarks"
  input_stream: "IMAGE:oms_image"
  output_stream: "SEATBELT_STATUS:seatbelt_status"
  output_stream: "ALERT:alert"
  options {
    [mediapipe.SeatbeltDetectionOptions.ext] {
      shoulder_belt_threshold: 0.2
      lap_belt_threshold: 0.15
      loose_threshold: 0.1
    }
  }
}

# ========== 4. CAN 信号融合（可选）==========
node {
  calculator: "SeatbeltCANFusionCalculator"
  input_stream: "SEATBELT_STATUS:seatbelt_status"
  input_stream: "CAN_SIGNAL:can_signal"
  output_stream: "FUSED_STATUS:fused_status"
  options {
    [mediapipe.SeatbeltCANFusionOptions.ext] {
      can_weight: 0.3
      vision_weight: 0.7
    }
  }
}

---

五十七、总结

要点	说明
Euro NCAP 要求	未系 + 错误佩戴检测
检测目标	肩带、腰带、锁扣
错误佩戴类型	Under-arm、Behind-back、Loose
融合方案	视觉 + CAN 信号

---

下篇预告

MediaPipe 系列 50：IMS 数据融合——多传感器协同

深入讲解 IMS 多传感器数据融合、时间同步、置信度融合策略。

---

参考资料

1. Euro NCAP. “Seat Belt Reminder” (2026)
2. NHTSA. “Seat Belt Use in 2025”

---

系列进度： 49/55
更新时间： 2026-03-12