"""在 person_tracker 已确认"有人"的时间段内跑 MediaPipe Pose 关键点估计， 用确定性规则引擎判断是否摔倒，写入 events 表（event_type='fall'）。 不用VLM做摔倒检测：这是安全信号，需要可解释、可调阈值、宁可多报不漏报的确定性 规则，而不是模型的"感觉"。规则：短时间内从"直立"骤变为"横躺"，且"横躺"状态 持续一段时间——单独判断"是否躺平"分不清摔倒和弯腰/坐下，必须叠加"骤变"和 "持续"两个时间条件。 """ import json import math from datetime import datetime, timedelta import cv2 import mediapipe as mp import numpy as np from mediapipe.tasks import python as mp_python from mediapipe.tasks.python import vision from analysis.config import ( FALL_MODEL_PATH, FALL_SAMPLE_STRIDE, FALL_MIN_VISIBILITY, FALL_ASPECT_RATIO_THRESH, FALL_TORSO_ANGLE_UPRIGHT, FALL_TORSO_ANGLE_PRONE, FALL_TRANSITION_MAX_SEC, FALL_SUSTAIN_MIN_SEC, FALL_GAP_TOLERANCE_SEC, PROJECT_ROOT, ) from analysis.db import get_all_videos, get_events, insert_event, update_video_status LEFT_SHOULDER, RIGHT_SHOULDER = 11, 12 LEFT_HIP, RIGHT_HIP = 23, 24 def _make_landmarker(): base_options = mp_python.BaseOptions(model_asset_path=str(FALL_MODEL_PATH)) options = vision.PoseLandmarkerOptions( base_options=base_options, running_mode=vision.RunningMode.VIDEO, num_poses=1, ) return vision.PoseLandmarker.create_from_options(options) def _classify_pose(landmarks, width, height): """返回 'upright' / 'prone' / 'unknown'（关键点不足或角度介于两者之间）。""" visible = [lm for lm in landmarks if lm.visibility >= FALL_MIN_VISIBILITY] if len(visible) < 4: return "unknown" xs = [lm.x * width for lm in visible] ys = [lm.y * height for lm in visible] bbox_w = max(xs) - min(xs) bbox_h = max(ys) - min(ys) if bbox_h <= 0: return "unknown" aspect_ratio = bbox_w / bbox_h shoulders = [landmarks[i] for i in (LEFT_SHOULDER, RIGHT_SHOULDER) if landmarks[i].visibility >= FALL_MIN_VISIBILITY] hips = [landmarks[i] for i in (LEFT_HIP, RIGHT_HIP) if landmarks[i].visibility >= FALL_MIN_VISIBILITY] if not shoulders or not hips: return "unknown" shoulder_mid = (sum(lm.x for lm in shoulders) / len(shoulders) * width, sum(lm.y for lm in shoulders) / len(shoulders) * height) hip_mid = (sum(lm.x for lm in hips) / len(hips) * width, sum(lm.y for lm in hips) / len(hips) * height) dx = abs(hip_mid[0] - shoulder_mid[0]) dy = abs(hip_mid[1] - shoulder_mid[1]) torso_angle = math.degrees(math.atan2(dx, dy + 1e-6)) # 0°=垂直直立，90°=水平躺平 if aspect_ratio >= FALL_ASPECT_RATIO_THRESH and torso_angle >= FALL_TORSO_ANGLE_PRONE: return "prone" if aspect_ratio < 1.0 and torso_angle <= FALL_TORSO_ANGLE_UPRIGHT: return "upright" return "unknown" def _extract_pose_states(landmarker, cap, fps, width, height, start_sec, end_sec): """在给定时间段内采样跑姿态估计，返回 [(t, state), ...]。""" start_frame = int(start_sec * fps) end_frame = int(end_sec * fps) cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame) states = [] frame_idx = start_frame while frame_idx <= end_frame: ret, frame = cap.read() if not ret: break if (frame_idx - start_frame) % FALL_SAMPLE_STRIDE == 0: t = frame_idx / fps rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=np.ascontiguousarray(rgb)) result = landmarker.detect_for_video(mp_image, int(t * 1000)) if result.pose_landmarks: state = _classify_pose(result.pose_landmarks[0], width, height) else: state = "unknown" states.append((t, state)) frame_idx += 1 return states def _find_fall_windows(states): """从时序姿态状态里找出"直立→骤变→横躺并持续"的片段，返回 [(start_sec, end_sec), ...]。 用"疑似摔倒事件"（episode）建模：只有真正观察到"直立"样本才会结束当前事件的计时； 事件进行期间即使长时间检测不到姿态（真实摔倒后人体角度/遮挡经常导致关键点大面积 漏检，不代表已经爬起来），也继续算同一次事件，直到确认恢复直立，或者长时间 （超过 FALL_GAP_TOLERANCE_SEC）连"横躺"样本都等不到才判定事件已经结束/跟丢。 """ windows = [] last_upright_t = None episode_start = None episode_last_prone = None def close_episode(): nonlocal episode_start, episode_last_prone if episode_start is not None and (episode_last_prone - episode_start) >= FALL_SUSTAIN_MIN_SEC: windows.append((episode_start, episode_last_prone)) episode_start = None episode_last_prone = None for t, state in states: if state == "upright": close_episode() last_upright_t = t elif state == "prone": if episode_start is None: # 只有紧接在"直立"之后、且间隔够短，才算一次骤变 if last_upright_t is not None and (t - last_upright_t) <= FALL_TRANSITION_MAX_SEC: episode_start = t episode_last_prone = t # 否则忽略：不是从直立状态骤变而来（比如视频开头就是横躺，或者是缓慢躺下） elif (t - episode_last_prone) <= FALL_GAP_TOLERANCE_SEC: episode_last_prone = t else: # 距离上一次确认"横躺"太久，且中途也没等到"直立"信号，视为跟丢，结算当前事件 close_episode() # state == "unknown" 不改变状态机，只是跳过这一帧（容忍短暂遮挡/漏检） close_episode() return windows def _merge_windows(events): """合并人物track的时间区间（有重叠或紧邻就合并成一段）。 同一个人有时会被 ByteTrack 拆成多个track_id（跟踪中途丢失又重新编号）， 这些track的时间区间往往高度重叠——如果按track各自跑一遍姿态估计， 同一次摔倒会被重复检测、重复插入，所以要先合并再处理，保证同一段画面只处理一次。 """ spans = sorted((e["start_offset_sec"], e["end_offset_sec"]) for e in events) merged = [] for start, end in spans: if merged and start <= merged[-1][1]: merged[-1] = (merged[-1][0], max(merged[-1][1], end)) else: merged.append((start, end)) return merged def process_video(video, person_events): if not person_events: update_video_status(video["id"], "fall_done") return 0 cap = cv2.VideoCapture(str(PROJECT_ROOT / video["file_path"])) if not cap.isOpened(): update_video_status(video["id"], "fall_done") return 0 fps = video["fps"] or 15.0 width, height = video["width"], video["height"] start_time = None if video["start_time"]: try: start_time = datetime.strptime(video["start_time"], "%Y-%m-%d %H:%M:%S") except ValueError: pass fall_count = 0 # landmarker 实例按需复用：VIDEO 模式要求时间戳在同一实例内严格递增，只有当下一个 # 事件的起始时间会比当前实例已处理到的时间更早（多人同时/交替出现导致事件时间重叠） # 时才换新实例。之前"每个事件都新建一个实例"的写法在批量跑千级视频时会持续新建 # GPU/Metal 上下文，资源迟迟未完全释放，最终在某个看似无关的帧读取操作上引发底层崩溃 landmarker = _make_landmarker() last_t = -1.0 try: # 先合并所有track的时间区间再处理：同一个人有时会被拆成多个重叠的track_id， # 按原始track各自跑一遍会导致同一次摔倒被重复检测、重复插入 for merged_start, merged_end in _merge_windows(person_events): start_frame_t = int(merged_start * fps) / fps if start_frame_t <= last_t: landmarker.close() landmarker = _make_landmarker() states = _extract_pose_states( landmarker, cap, fps, width, height, merged_start, merged_end, ) if states: last_t = states[-1][0] for start_sec, end_sec in _find_fall_windows(states): timestamp_abs = (start_time + timedelta(seconds=start_sec)).isoformat() if start_time else None insert_event({ "video_id": video["id"], "event_type": "fall", "label": None, "start_offset_sec": start_sec, "end_offset_sec": end_sec, "timestamp_abs": timestamp_abs, "confidence": None, "track_id": None, "thumbnail_path": None, "extra_json": None, "created_at": datetime.now().isoformat(), }) fall_count += 1 except Exception as e: # 兜底：单条视频处理异常不应中断整个批次，标记失败并继续下一条。 # 异常本身可能就是底层解码/GPU上下文已经出问题，close()也可能再报错，不能让它盖过原始异常 try: landmarker.close() except Exception: pass cap.release() update_video_status(video["id"], "fall_failed", error_message=str(e)) return fall_count landmarker.close() cap.release() update_video_status(video["id"], "fall_done") return fall_count def process_all(limit=None): videos = get_all_videos(status="person_done") if limit: videos = videos[:limit] events_by_video = {} for e in get_events("person_presence"): events_by_video.setdefault(e["video_id"], []).append(e) processed = 0 total_falls = 0 for video in videos: total_falls += process_video(video, events_by_video.get(video["id"], [])) processed += 1 return {"processed": processed, "total_falls": total_falls} if __name__ == "__main__": print("===== 开始检测摔倒事件 =====") result = process_all() print(f"→ 处理 {result['processed']} 个视频，检测到 {result['total_falls']} 次疑似摔倒") print("===== 检测完成 =====")