yingshiyun/analysis/motion_filter.py

"""MOG2背景建模检测每个视频里"有活动"的时间窗口，写回 videos.motion_windows_json。

长视频不能逐帧跑重模型，所以先用轻量背景建模预筛出活动窗口，后续 person_tracker /
fall_detector 只在窗口内跑，空场景直接跳过。

之前用固定阈值帧差法（相邻两帧像素差），对"人物占画面小、动作幅度小"的场景
（如老人在客厅整理东西，移动缓慢、遮挡画面比例低）容易漏检。改用 MOG2
自适应背景建模：背景是随时间缓慢更新的统计模型而非"上一帧"，前景掩码对
慢速/小幅度运动更敏感，且能自动适应光照的缓慢变化。
"""
import json
from datetime import datetime

import cv2

from analysis.config import (
    MOTION_RESIZE_WIDTH,
    MOTION_HISTORY,
    MOTION_VAR_THRESHOLD,
    MOTION_WARMUP_FRAMES,
    MOTION_SHORT_CLIP_SEC,
    MOTION_PIXEL_RATIO,
    MOTION_MERGE_GAP_SEC,
    MOTION_MIN_DURATION_SEC,
    MOTION_SAMPLE_STRIDE,
    PROJECT_ROOT,
)
from analysis.db import get_all_videos, update_motion_windows


def detect_motion_windows(file_path, duration_sec=None):
    """返回该视频里的活动时间窗口列表 [[start_sec, end_sec], ...]。

    duration_sec 由调用方传入（video_scanner用ffprobe测得，比cv2的帧数估算更准）；
    不传时退化为用 cv2 的帧数/帧率估算，供独立调用/测试使用。
    """
    cap = cv2.VideoCapture(str(file_path))
    if not cap.isOpened():
        return []

    fps = cap.get(cv2.CAP_PROP_FPS) or 15.0

    if duration_sec is None:
        frame_count = cap.get(cv2.CAP_PROP_FRAME_COUNT)
        duration_sec = frame_count / fps if frame_count > 0 else None

    # 触发即录的极短告警片段，帧数不够 MOG2 预热收敛，直接整段视为活动窗口
    if duration_sec is not None and duration_sec < MOTION_SHORT_CLIP_SEC:
        cap.release()
        return [[0.0, duration_sec]]

    subtractor = cv2.createBackgroundSubtractorMOG2(
        history=MOTION_HISTORY, varThreshold=MOTION_VAR_THRESHOLD, detectShadows=True
    )
    motion_flags = []
    frame_idx = 0
    sample_idx = 0

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        if frame_idx % MOTION_SAMPLE_STRIDE != 0:
            frame_idx += 1
            continue

        h, w = frame.shape[:2]
        scale = MOTION_RESIZE_WIDTH / w
        small = cv2.resize(frame, (MOTION_RESIZE_WIDTH, max(1, int(h * scale))))
        small = cv2.GaussianBlur(small, (5, 5), 0)

        fg_mask = subtractor.apply(small)
        # MOG2 用 127 标记阴影、255 标记前景，阴影不算真实活动，只统计255
        foreground_ratio = (fg_mask == 255).sum() / fg_mask.size

        if sample_idx >= MOTION_WARMUP_FRAMES:
            motion_flags.append((frame_idx / fps, foreground_ratio > MOTION_PIXEL_RATIO))

        frame_idx += 1
        sample_idx += 1

    cap.release()
    return merge_motion_windows(motion_flags)


def merge_motion_windows(motion_flags):
    windows = []
    window_start = None
    window_last_motion = None

    for t, is_motion in motion_flags:
        if is_motion:
            if window_start is None:
                window_start = t
            window_last_motion = t
        elif window_start is not None and (t - window_last_motion) > MOTION_MERGE_GAP_SEC:
            windows.append([window_start, window_last_motion])
            window_start = None
            window_last_motion = None

    if window_start is not None:
        windows.append([window_start, window_last_motion])

    return [w for w in windows if (w[1] - w[0]) >= MOTION_MIN_DURATION_SEC]


def process_all(limit=None):
    videos = get_all_videos(status="pending")
    if limit:
        videos = videos[:limit]

    processed = 0
    no_motion = 0
    failed = 0
    for video in videos:
        try:
            file_path = PROJECT_ROOT / video["file_path"]
            if not file_path.exists():
                update_video_status(video["id"], "motion_failed", error_message="文件不存在")
                failed += 1
                processed += 1
                continue
            windows = detect_motion_windows(file_path, duration_sec=video["duration_sec"])
            update_motion_windows(
                video["id"],
                json.dumps(windows),
                status="motion_done",
                analyzed_at=datetime.now().isoformat(),
            )
            processed += 1
            if not windows:
                no_motion += 1
        except Exception as e:
            update_video_status(video["id"], "motion_failed", error_message=str(e))
            failed += 1
            processed += 1

    return {"processed": processed, "no_motion": no_motion, "failed": failed}


if __name__ == "__main__":
    print("===== 开始检测活动窗口 =====")
    result = process_all()
    print(f"→ 处理 {result['processed']} 个视频，其中 {result['no_motion']} 个未检测到活动")
    print("===== 检测完成 =====")