YOLO如何结合opencv实现视觉实时摔倒检测？

用 YOLOv8 Pose + ByteTrack 对视频中的“人”进行跟踪，同时用关键点和边界框的几何特征做简易规则，实时判定“站立/行走/摔倒/躺卧”。代码包含完整可运行脚本与参数可调，适合先跑通再迭代优化。

准备环境

Python 3.8+

安装依赖

pip install ultralytics opencv-python numpy

说明

模型：yolov8n-pose.pt（轻量），可换 yolov8s/m/l-pose 提高精度。

跟踪：内置 ByteTrack，自动给每个人分配 track id。

判定逻辑（可调阈值）：

使用躯干角度（髋-肩连线相对竖直角度）、框的长宽比、髋部向下速度、中心速度等。

突然由“竖直”到“水平”且髋部下坠速度快，判定为“摔倒”。

竖直且中心移动速度较大，判定“行走”；竖直且速度较小，判定“站立”。

长时间水平，判定“躺卧”。

代码（保存为 yolo_fall_walk.py）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
from collections import deque
import cv2
import numpy as np
from ultralytics import YOLO

# COCO关键点索引
L_SHOULDER, R_SHOULDER = 5, 6
L_HIP, R_HIP = 11, 12

STATE_COLORS = {
    "walking": (0, 200, 0),     # 绿
    "standing": (200, 200, 0),  # 黄
    "falling": (0, 0, 255),     # 红
    "lying": (0, 140, 255),     # 橙
    "unknown": (180, 180, 180), # 灰
}

def torso_angle_deg(kps_xy):
    # 返回“躯干相对竖直的角度”，0度=笔直竖立，90度=水平
    if kps_xy is None or len(kps_xy) < 17:
        return None, None
    ls, rs = kps_xy[L_SHOULDER], kps_xy[R_SHOULDER]
    lh, rh = kps_xy[L_HIP], kps_xy[R_HIP]
    # 如果关键点缺失，用None
    if (ls == 0).all() or (rs == 0).all() or (lh == 0).all() or (rh == 0).all():
        return None, None
    shoulder = (ls + rs) / 2.0
    hip = (lh + rh) / 2.0
    v = shoulder - hip
    vx, vy = float(v[0]), float(v[1])
    if abs(vx) < 1e-6 and abs(vy) < 1e-6:
        return None, hip
    # 与竖直的夹角：用atan2(|vx|, |vy|)
    ang = np.degrees(np.arctan2(abs(vx), abs(vy)))
    return ang, hip

class TrackState:
    def __init__(self, hist=15):
        self.history = deque(maxlen=hist)  # 存最近几帧的指标
        self.state = "unknown"
        self.last_state = "unknown"
        self.stable_count = 0
        self.falling_frames = 0
        self.lying_frames = 0
        self.missed = 0

class FallWalkingDetector:
    def __init__(self, fps=30.0, hist=15):
        self.fps = fps if fps and fps > 0 else 30.0
        self.dt = 1.0 / self.fps
        self.tracks = {}
        self.hist = hist

        # 参数阈值（可调）
        self.AR_VERT = 1.2     # 竖直：高宽比
        self.AR_HORI = 0.85    # 水平：高宽比
        self.ANG_VERT = 35.0   # 竖直角阈
        self.ANG_HORI = 55.0   # 水平角阈
        self.SPD_WALK = 0.35   # 步行速度阈（按框高度归一化，单位：高/秒）
        self.VHIP_FALL = 1.2   # 髋部向下速度触发摔倒（高/秒）
        self.ANG_JUMP = 20.0   # 角度突然变化阈
        self.AR_DROP = 0.3     # 高宽比快速下降比例阈
        self.MIN_FALL_FRAMES = max(2, int(self.fps * 0.1))
        self.MIN_LYING_FRAMES = max(4, int(self.fps * 0.2))

    def _get_track(self, tid):
        if tid not in self.tracks:
            self.tracks[tid] = TrackState(hist=self.hist)
        return self.tracks[tid]

    def _geom_flags(self, ar, ang):
        vertical = (ar is not None and ar > self.AR_VERT) or (ang is not None and ang < self.ANG_VERT)
        horizontal = (ar is not None and ar < self.AR_HORI) or (ang is not None and ang > self.ANG_HORI)
        return vertical, horizontal

    def update(self, tid, bbox, kps_xy):
        st = self._get_track(tid)
        x1, y1, x2, y2 = bbox
        w = max(1.0, x2 - x1)
        h = max(1.0, y2 - y1)
        cx, cy = (x1 + x2) / 2.0, (y1 + y2) / 2.0
        ar = h / w

        ang, hip = torso_angle_deg(kps_xy)
        hip_y = hip[1] if isinstance(hip, (list, tuple, np.ndarray)) else None

        # 前一帧数据
        prev = st.history[-1] if len(st.history) > 0 else None
        spd = 0.0
        vhi...

点击查看剩余70%