# OpenCV实时视频处理：从摄像头捕获到智能分析实践

实时视频处理是计算机视觉应用的核心基础。本文将系统介绍如何使用OpenCV实现从摄像头捕获到实时分析的完整流程，涵盖视频采集、帧处理、对象检测和性能优化等关键技术。

## 1. 视频捕获与基础处理

### 1.1 多源视频捕获实现

```python

# video_capture_multi_source.py

import cv2

import numpy as np

from typing import Optional, Tuple

import threading

import queue

class VideoCaptureManager:

    """视频捕获管理器，支持多源输入"""

    

    def __init__(self, source: str = '0', api_preference: Optional[int] = None):

        """

        初始化视频捕获

        source: 摄像头ID (0,1,2...), 视频文件路径, 或RTSP流地址

        api_preference: OpenCV视频捕获API偏好设置

        """

        self.source = source

        self.cap = None

        self.is_running = False

        self.frame_queue = queue.Queue(maxsize=2)

        self.capture_thread = None

        

        # 尝试不同的API后端

        self.api_preferences = api_preference or [

            cv2.CAP_DSHOW,      # DirectShow (Windows)

            cv2.CAP_MSMF,       # Microsoft Media Foundation

            cv2.CAP_V4L2,       # Video4Linux (Linux)

            cv2.CAP_ANY         # 自动选择

        ]

    

    def initialize_capture(self) -> bool:

        """初始化视频捕获设备"""

        for api in self.api_preferences:

            try:

                print(f"尝试使用API {api} 打开视频源: {self.source}")

                

                if self.source.isdigit():

                    # 摄像头ID

                    self.cap = cv2.VideoCapture(int(self.source), api)

                elif self.source.startswith(('http://', 'rtsp://', 'rtmp://')):

                    # 网络流

                    self.cap = cv2.VideoCapture(self.source, api)

                else:

                    # 视频文件

                    self.cap = cv2.VideoCapture(self.source, api)

                

                if self.cap.isOpened():

                    print(f"成功打开视频源，使用API: {api}")

                    

                    # 获取视频参数

                    self.frame_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))

                    self.frame_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

                    self.fps = self.cap.get(cv2.CAP_PROP_FPS)

                    

                    if self.fps <= 0:

                        self.fps = 30  # 默认帧率

                    

                    print(f"分辨率: {self.frame_width}x{self.frame_height}")

                    print(f"帧率: {self.fps}")

                    return True

                    

            except Exception as e:

                print(f"API {api} 失败: {e}")

                continue

        

        print("所有API尝试均失败")

        return False

    

    def start_capture_thread(self):

        """启动捕获线程"""

        if not self.cap or not self.cap.isOpened():

            if not self.initialize_capture():

                raise RuntimeError("无法初始化视频捕获")

        

        self.is_running = True

        self.capture_thread = threading.Thread(target=self._capture_loop)

        self.capture_thread.daemon = True

        self.capture_thread.start()

    

    def _capture_loop(self):

        """捕获线程主循环"""

        while self.is_running:

            try:

                ret, frame = self.cap.read()

                if not ret:

                    print("帧读取失败，尝试重新连接...")

                    self._reconnect()

                    continue

                

                # 清理队列，只保留最新帧

                while self.frame_queue.full():

                    try:

                        self.frame_queue.get_nowait()

                    except queue.Empty:

                        break

                

                self.frame_queue.put(frame)

                

            except Exception as e:

                print(f"捕获错误: {e}")

                if not self.is_running:

                    break

                time.sleep(0.1)

    

    def read(self) -> Optional[np.ndarray]:

        """读取最新帧"""

        try:

            return self.frame_queue.get_nowait()

        except queue.Empty:

            return None

    

    def _reconnect(self):

        """重新连接视频源"""

        for _ in range(3):  # 重试3次

            try:

                self.cap.release()

                time.sleep(1)

                if self.initialize_capture():

                    return True

            except:

                continue

        return False

    

    def release(self):

        """释放资源"""

        self.is_running = False

        if self.capture_thread:

            self.capture_thread.join(timeout=2)

        if self.cap:

            self.cap.release()

# 使用示例

def multi_camera_capture():

    """多摄像头同时捕获示例"""

    cameras = []

    

    # 尝试打开多个摄像头

    for cam_id in range(3):

        cam = VideoCaptureManager(str(cam_id))

        if cam.initialize_capture():

            cam.start_capture_thread()

            cameras.append(cam)

            print(f"摄像头 {cam_id} 初始化成功")

        else:

            print(f"摄像头 {cam_id} 初始化失败")

    

    try:

        while True:

            frames = []

            for cam in cameras:

                frame = cam.read()

                if frame is not None:

                    frames.append(frame)

            

            if frames:

                # 合并显示多摄像头画面

                if len(frames) >= 2:

                    # 水平拼接

                    combined = cv2.hconcat(frames[:2])

                    if len(frames) > 2:

                        # 如果有第三摄像头，垂直拼接

                        row2 = cv2.hconcat(frames[2:])

                        combined = cv2.vconcat([combined, row2])

                else:

                    combined = frames[0]

                

                cv2.imshow('Multi-Camera View', combined)

            

            if cv2.waitKey(1) & 0xFF == ord('q'):

                break

                

    finally:

        for cam in cameras:

            cam.release()

        cv2.destroyAllWindows()

```

### 1.2 帧处理与预处理流水线

```python

# video_processing_pipeline.py

import cv2

import numpy as np

from dataclasses import dataclass

from typing import List, Callable, Optional

import time

@dataclass

class FrameProcessingConfig:

    """帧处理配置"""

    target_size: Tuple[int, int] = (640, 480)

    convert_to_gray: bool = False

    apply_histogram_eq: bool = False

    denoise_strength: int = 5

    brightness_adjust: float = 1.0

    contrast_adjust: float = 1.0

class FrameProcessingPipeline:

    """帧处理流水线"""

    

    def __init__(self, config: Optional[FrameProcessingConfig] = None):

        self.config = config or FrameProcessingConfig()

        self.processors = self._initialize_processors()

        self.processing_times = []

    

    def _initialize_processors(self) -> List[Callable]:

        """初始化处理函数列表"""

        processors = []

        

        # 1. 尺寸调整

        if self.config.target_size:

            processors.append(self._resize_frame)

        

        # 2. 颜色空间转换

        if self.config.convert_to_gray:

            processors.append(self._convert_to_grayscale)

        

        # 3. 直方图均衡化

        if self.config.apply_histogram_eq:

            processors.append(self._apply_histogram_equalization)

        

        # 4. 去噪

        if self.config.denoise_strength > 0:

            processors.append(self._apply_denoising)

        

        # 5. 亮度和对比度调整

        if self.config.brightness_adjust != 1.0 or self.config.contrast_adjust != 1.0:

            processors.append(self._adjust_brightness_contrast)

        

        return processors

    

    def process_frame(self, frame: np.ndarray) -> np.ndarray:

        """处理单帧图像"""

        start_time = time.time()

        

        processed = frame.copy()

        

        for processor in self.processors:

            processed = processor(processed)

        

        # 记录处理时间

        process_time = (time.time() - start_time) * 1000  # 转换为毫秒

        self.processing_times.append(process_time)

        

        # 保持最近100次处理时间记录

        if len(self.processing_times) > 100:

            self.processing_times.pop(0)

        

        return processed

    

    def _resize_frame(self, frame: np.ndarray) -> np.ndarray:

        """调整帧尺寸"""

        return cv2.resize(frame, self.config.target_size, 

                         interpolation=cv2.INTER_AREA)

    

    def _convert_to_grayscale(self, frame: np.ndarray) -> np.ndarray:

        """转换为灰度图"""

        if len(frame.shape) == 3:

            return cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

        return frame

    

    def _apply_histogram_equalization(self, frame: np.ndarray) -> np.ndarray:

        """应用直方图均衡化"""

        if len(frame.shape) == 2:  # 灰度图

            return cv2.equalizeHist(frame)

        else:  # 彩色图

            # 在Y通道上应用均衡化

            ycrcb = cv2.cvtColor(frame, cv2.COLOR_BGR2YCrCb)

            ycrcb[:, :, 0] = cv2.equalizeHist(ycrcb[:, :, 0])

            return cv2.cvtColor(ycrcb, cv2.COLOR_YCrCb2BGR)

    

    def _apply_denoising(self, frame: np.ndarray) -> np.ndarray:

        """应用去噪"""

        strength = self.config.denoise_strength

        if len(frame.shape) == 2:  # 灰度图

            return cv2.fastNlMeansDenoising(frame, h=strength)

        else:  # 彩色图

            return cv2.fastNlMeansDenoisingColored(frame, None, strength, strength, 7, 21)

    

    def _adjust_brightness_contrast(self, frame: np.ndarray) -> np.ndarray:

        """调整亮度和对比度"""

        alpha = self.config.contrast_adjust  # 对比度因子

        beta = self.config.brightness_adjust * 127 * (1 - alpha)  # 亮度偏移

        

        # 应用变换: output = alpha * input + beta

        result = cv2.convertScaleAbs(frame, alpha=alpha, beta=beta)

        return result

    

    def get_performance_stats(self) -> dict:

        """获取性能统计"""

        if not self.processing_times:

            return {}

        

        times = np.array(self.processing_times)

        return {

            'avg_processing_time_ms': np.mean(times),

            'min_processing_time_ms': np.min(times),

            'max_processing_time_ms': np.max(times),

            'std_processing_time_ms': np.std(times),

            'estimated_fps': 1000 / np.mean(times) if np.mean(times) > 0 else 0

        }

    

    def overlay_stats(self, frame: np.ndarray) -> np.ndarray:

        """在帧上叠加性能统计信息"""

        stats = self.get_performance_stats()

        if not stats:

            return frame

        

        display_frame = frame.copy()

        

        # 添加性能信息

        y_offset = 30

        font = cv2.FONT_HERSHEY_SIMPLEX

        font_scale = 0.6

        font_color = (0, 255, 0)  # 绿色

        thickness = 2

        

        info_lines = [

            f"处理时间: {stats['avg_processing_time_ms']:.1f}ms",

            f"帧率: {stats['estimated_fps']:.1f} FPS",

            f"最小: {stats['min_processing_time_ms']:.1f}ms",

            f"最大: {stats['max_processing_time_ms']:.1f}ms"

        ]

        

        for i, line in enumerate(info_lines):

            position = (10, y_offset + i * 25)

            cv2.putText(display_frame, line, position, 

                       font, font_scale, font_color, thickness)

        

        return display_frame

# 使用示例

def realtime_processing_demo():

    """实时视频处理演示"""

    # 初始化捕获

    cap_manager = VideoCaptureManager('0')

    if not cap_manager.initialize_capture():

        print("摄像头初始化失败")

        return

    

    cap_manager.start_capture_thread()

    

    # 配置处理流水线

    config = FrameProcessingConfig(

        target_size=(800, 600),

        convert_to_gray=False,

        apply_histogram_eq=True,

        denoise_strength=10,

        brightness_adjust=1.1,

        contrast_adjust=1.2

    )

    

    pipeline = FrameProcessingPipeline(config)

    

    print("开始实时视频处理，按'q'退出...")

    

    try:

        while True:

            frame = cap_manager.read()

            if frame is None:

                continue

            

            # 处理帧

            processed = pipeline.process_frame(frame)

            

            # 叠加性能信息

            displayed = pipeline.overlay_stats(processed)

            

            # 显示原始帧和处理后帧

            comparison = cv2.hconcat([frame, displayed])

            cv2.imshow('原始 vs 处理后', comparison)

            

            # 定期打印性能统计

            if len(pipeline.processing_times) % 50 == 0:

                stats = pipeline.get_performance_stats()

                print(f"平均处理时间: {stats['avg_processing_time_ms']:.1f}ms, "

                      f"预估帧率: {stats['estimated_fps']:.1f}FPS")

            

            if cv2.waitKey(1) & 0xFF == ord('q'):

                break

                

    finally:

        cap_manager.release()

        cv2.destroyAllWindows()

```

## 2. 实时对象检测与跟踪

### 2.1 基于深度学习的目标检测

```python

# realtime_object_detection.py

import cv2

import numpy as np

from typing import List, Tuple, Dict

import time

class ObjectDetector:

    """实时对象检测器"""

    

    def __init__(self, model_config: Dict):

        """

        初始化对象检测器

        model_config: 模型配置字典

        """

        self.config = model_config

        self.net = None

        self.classes = []

        self.colors = []

        self._initialize_model()

    

    def _initialize_model(self):

        """初始化深度学习模型"""

        model_type = self.config.get('model_type', 'yolo')

        

        if model_type == 'yolo':

            self._load_yolo_model()

        elif model_type == 'ssd':

            self._load_ssd_model()

        elif model_type == 'efficientdet':

            self._load_efficientdet_model()

        else:

            raise ValueError(f"不支持的模型类型: {model_type}")

        

        # 生成颜色

        np.random.seed(42)

        self.colors = np.random.uniform(0, 255, size=(len(self.classes), 3))

    

    def _load_yolo_model(self):

        """加载YOLO模型"""

        # 配置文件路径

        config_path = self.config['config_path']

        weights_path = self.config['weights_path']

        classes_path = self.config['classes_path']

        

        # 加载类别

        with open(classes_path, 'r') as f:

            self.classes = [line.strip() for line in f.readlines()]

        

        # 加载网络

        self.net = cv2.dnn.readNet(weights_path, config_path)

        

        # 设置后端（GPU加速）

        if cv2.cuda.getCudaEnabledDeviceCount() > 0:

            print("使用CUDA加速")

            self.net.setPreferableBackend(cv2.dnn.DNN_BACKEND_CUDA)

            self.net.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA)

        else:

            print("使用CPU")

            self.net.setPreferableBackend(cv2.dnn.DNN_BACKEND_OPENCV)

            self.net.setPreferableTarget(cv2.dnn.DNN_TARGET_CPU)

    

    def detect(self, frame: np.ndarray, confidence_threshold: float = 0.5, 

               nms_threshold: float = 0.4) -> List[Dict]:

        """

        检测图像中的对象

        返回: [{'class_id': int, 'class_name': str, 'confidence': float, 

                'box': (x, y, w, h), 'center': (cx, cy)}]

        """

        # 获取图像尺寸

        height, width = frame.shape[:2]

        

        # 准备输入blob

        blob = cv2.dnn.blobFromImage(

            frame, 1/255.0, (416, 416), swapRB=True, crop=False

        )

        

        # 前向传播

        self.net.setInput(blob)

        layer_names = self.net.getLayerNames()

        output_layers = [

            layer_names[i - 1] for i in self.net.getUnconnectedOutLayers()

        ]

        outputs = self.net.forward(output_layers)

        

        # 解析检测结果

        boxes = []

        confidences = []

        class_ids = []

        

        for output in outputs:

            for detection in output:

                scores = detection[5:]

                class_id = np.argmax(scores)

                confidence = scores[class_id]

                

                if confidence > confidence_threshold:

                    # 计算边界框坐标

                    center_x = int(detection[0] * width)

                    center_y = int(detection[1] * height)

                    w = int(detection[2] * width)

                    h = int(detection[3] * height)

                    

                    # 左上角坐标

                    x = int(center_x - w / 2)

                    y = int(center_y - h / 2)

                    

                    boxes.append([x, y, w, h])

                    confidences.append(float(confidence))

                    class_ids.append(class_id)

        

        # 应用非极大值抑制

        indices = cv2.dnn.NMSBoxes(

            boxes, confidences, confidence_threshold, nms_threshold

        )

        

        # 构建结果列表

        results = []

        if len(indices) > 0:

            for i in indices.flatten():

                box = boxes[i]

                results.append({

                    'class_id': class_ids[i],

                    'class_name': self.classes[class_ids[i]],

                    'confidence': confidences[i],

                    'box': (box[0], box[1], box[2], box[3]),

                    'center': (box[0] + box[2]//2, box[1] + box[3]//2)

                })

        

        return results

    

    def draw_detections(self, frame: np.ndarray, detections: List[Dict]) -> np.ndarray:

        """在图像上绘制检测结果"""

        result_frame = frame.copy()

        

        for det in detections:

            x, y, w, h = det['box']

            class_name = det['class_name']

            confidence = det['confidence']

            

            # 绘制边界框

            color = self.colors[det['class_id']].tolist()

            cv2.rectangle(result_frame, (x, y), (x+w, y+h), color, 2)

            

            # 绘制标签背景

            label = f"{class_name}: {confidence:.2f}"

            label_size = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 2)[0]

            cv2.rectangle(result_frame, (x, y-25), (x+label_size[0], y), color, -1)

            

            # 绘制标签文本

            cv2.putText(result_frame, label, (x, y-5),

                       cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 2)

            

            # 绘制中心点

            cx, cy = det['center']

            cv2.circle(result_frame, (cx, cy), 5, color, -1)

        

        return result_frame

class MultiObjectTracker:

    """多对象跟踪器"""

    

    def __init__(self, max_disappeared: int = 10):

        self.next_object_id = 0

        self.objects = {}  # object_id -> {'box': (x,y,w,h), 'centroids': [], 'disappeared': 0}

        self.disappeared = {}

        self.max_disappeared = max_disappeared

    

    def update(self, detections: List[Dict]):

        """更新跟踪器状态"""

        # 如果没有检测到对象，增加所有现有对象的消失计数

        if len(detections) == 0:

            for object_id in list(self.objects.keys()):

                self.disappeared[object_id] += 1

                if self.disappeared[object_id] > self.max_disappeared:

                    self._deregister_object(object_id)

<"dz.fsjdpx.com"><"kf.fsjdpx.com"><"pt.fsjdpx.com">

            return self.objects

        

        # 如果有检测但没有现有对象，注册新对象

        if len(self.objects) == 0:

            for det in detections:

                self._register_object(det)

        else:

            # 匹配现有对象和检测

            object_ids = list(self.objects.keys())

            object_boxes = [self.objects[obj_id]['box'] for obj_id in object_ids]

            

            # 计算IOU矩阵

            iou_matrix = self._compute_iou_matrix(object_boxes, detections)

            

            # 匈牙利算法匹配

            matched_indices = self._hungarian_match(iou_matrix)

            

            # 更新匹配的对象

            matched_object_ids = set()

            matched_detection_indices = set()

            

            for obj_idx, det_idx in matched_indices:

                if iou_matrix[obj_idx][det_idx] > 0.3:  # IOU阈值

                    object_id = object_ids[obj_idx]

                    det = detections[det_idx]

                    

                    # 更新对象位置

                    self.objects[object_id]['box'] = det['box']

                    self.objects[object_id]['centroids'].append(det['center'])

                    self.disappeared[object_id] = 0

                    

                    matched_object_ids.add(object_id)

                    matched_detection_indices.add(det_idx)

            

            # 处理未匹配的现有对象

            for obj_idx, object_id in enumerate(object_ids):

                if object_id not in matched_object_ids:

                    self.disappeared[object_id] += 1

                    if self.disappeared[object_id] > self.max_disappeared:

                        self._deregister_object(object_id)

            

            # 注册未匹配的检测作为新对象

            for det_idx, det in enumerate(detections):

                if det_idx not in matched_detection_indices:

                    self._register_object(det)

        

        return self.objects

    

    def _register_object(self, detection: Dict):

        """注册新对象"""

        object_id = self.next_object_id

        self.objects[object_id] = {

            'box': detection['box'],

            'centroids': [detection['center']],

            'class_name': detection['class_name'],

            'confidence': detection['confidence']

        }

        self.disappeared[object_id] = 0

        self.next_object_id += 1

    

    def _deregister_object(self, object_id: int):

        """注销对象"""

        del self.objects[object_id]

        del self.disappeared[object_id]

    

    def _compute_iou_matrix(self, boxes1, detections):

        """计算IOU矩阵"""

        matrix = np.zeros((len(boxes1), len(detections)))

        

        for i, box1 in enumerate(boxes1):

            for j, det in enumerate(detections):

                box2 = det['box']

                iou = self._compute_iou(box1, box2)

                matrix[i][j] = iou

        

        return matrix

    

    def _compute_iou(self, box1, box2):

        """计算两个边界框的IOU"""

        x1, y1, w1, h1 = box1

        x2, y2, w2, h2 = box2

        

        # 计算交集

        xi1 = max(x1, x2)

        yi1 = max(y1, y2)

        xi2 = min(x1 + w1, x2 + w2)

        yi2 = min(y1 + h1, y2 + h2)

        

        inter_area = max(0, xi2 - xi1) * max(0, yi2 - yi1)

        

        # 计算并集

        box1_area = w1 * h1

        box2_area = w2 * h2

        union_area = box1_area + box2_area - inter_area

        

        return inter_area / union_area if union_area > 0 else 0

    

    def _hungarian_match(self, cost_matrix):

        """使用匈牙利算法进行匹配"""

        from scipy.optimize import linear_sum_assignment

        

        # 转换为负值用于最小化

        cost_matrix = -cost_matrix

        row_ind, col_ind = linear_sum_assignment(cost_matrix)

        

        return list(zip(row_ind, col_ind))

# 使用示例

def realtime_object_tracking():

    """实时对象检测与跟踪演示"""

    

    # 配置模型

    model_config = {

        'model_type': 'yolo',

        'config_path': 'models/yolov4-tiny.cfg',

        'weights_path': 'models/yolov4-tiny.weights',

        'classes_path': 'models/coco.names'

    }

    

    # 初始化检测器和跟踪器

    detector = ObjectDetector(model_config)

    tracker = MultiObjectTracker(max_disappeared=20)

    

    # 初始化视频捕获

    cap = VideoCaptureManager('0')

    if not cap.initialize_capture():

        print("摄像头初始化失败")

        return

    

    cap.start_capture_thread()

    

    print("开始实时对象检测与跟踪，按'q'退出...")

    

    frame_count = 0

    processing_times = []

    

    try:

        while True:

            start_time = time.time()

            

            # 读取帧

            frame = cap.read()

            if frame is None:

                continue

            

            frame_count += 1

            

            # 对象检测

            detections = detector.detect(frame, confidence_threshold=0.5)

            

            # 对象跟踪

            tracked_objects = tracker.update(detections)

            

            # 绘制检测和跟踪结果

            result_frame = detector.draw_detections(frame, detections)

            

            # 绘制跟踪信息

            for object_id, obj_info in tracked_objects.items():

                x, y, w, h = obj_info['box']

                

                # 绘制跟踪ID

                cv2.putText(result_frame, f"ID: {object_id}", (x, y-30),

                           cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2)

                

                # 绘制轨迹

                centroids = obj_info['centroids']

                if len(centroids) > 1:

                    for i in range(1, len(centroids)):

                        cv2.line(result_frame, centroids[i-1], centroids[i],

                                (0, 255, 255), 2)

            

            # 计算并显示处理时间

            processing_time = (time.time() - start_time) * 1000

            processing_times.append(processing_time)

            

            if len(processing_times) > 100:

                processing_times.pop(0)

            

            avg_time = np.mean(processing_times) if processing_times else 0

            fps = 1000 / avg_time if avg_time > 0 else 0

            

            # 显示性能信息

            cv2.putText(result_frame, f"FPS: {fps:.1f}", (10, 30),

                       cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)

            cv2.putText(result_frame, f"Objects: {len(tracked_objects)}", (10, 60),

                       cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)

            

            cv2.imshow('Object Detection & Tracking', result_frame)

            

            if frame_count % 50 == 0:

                print(f"帧率: {fps:.1f}FPS, 检测到对象: {len(tracked_objects)}")

            

            if cv2.waitKey(1) & 0xFF == ord('q'):

                break

                

    finally:

        cap.release()

        cv2.destroyAllWindows()

```

### 2.2 运动检测与异常识别

```python

# motion_detection.py

import cv2

import numpy as np

from collections import deque

import time

class MotionDetector:

    """运动检测器"""

    

    def __init__(self, min_area: int = 500, history: int = 50, 

                 var_threshold: int = 16, detect_shadows: bool = True):

<"vs.fsjdpx.com"><"rb.fsjdpx.com"><"mo.fsjdpx.com">

        """

        初始化运动检测器

        min_area: 最小运动区域面积

        history: 背景减除历史长度

        var_threshold: 方差阈值

        """

        self.min_area = min_area

        self.bg_subtractor = cv2.createBackgroundSubtractorMOG2(

            history=history, 

            varThreshold=var_threshold,

            detectShadows=detect_shadows

        )

        

        # 运动历史

        self.motion_history = deque(maxlen=100)

        self.motion_alert_threshold = 10  # 连续检测到运动的帧数阈值

    

    def detect(self, frame: np.ndarray) -> dict:

        """检测运动"""

        # 应用背景减除

        fg_mask = self.bg_subtractor.apply(frame)

        

        # 去除阴影（设为黑色）

        if self.bg_subtractor.getDetectShadows():

            fg_mask[fg_mask == 127] = 0

        

        # 二值化

        _, thresh = cv2.threshold(fg_mask, 25, 255, cv2.THRESH_BINARY)

        

        # 形态学操作去除噪声

        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))

        thresh = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel)

        thresh = cv2.morphologyEx(thresh, cv2.MORPH_CLOSE, kernel)

        thresh = cv2.dilate(thresh, kernel, iterations=2)

        

        # 查找轮廓

        contours, _ = cv2.findContours(

            thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE

        )

        

        # 过滤小轮廓

        motion_regions = []

        motion_detected = False

        

        for contour in contours:

            area = cv2.contourArea(contour)

            if area > self.min_area:

                motion_detected = True

                

                # 获取边界框

                x, y, w, h = cv2.boundingRect(contour)

                motion_regions.append({

                    'bbox': (x, y, w, h),

                    'area': area,

                    'center': (x + w//2, y + h//2)

                })

        

        # 更新运动历史

        self.motion_history.append(motion_detected)

        

        # 检查是否需要警报

        alert_triggered = False

        if len(self.motion_history) >= self.motion_alert_threshold:

            recent_motion = list(self.motion_history)[-self.motion_alert_threshold:]

            if all(recent_motion):

                alert_triggered = True

        

        return {

            'motion_detected': motion_detected,

            'motion_regions': motion_regions,

            'mask': thresh,

            'alert_triggered': alert_triggered

        }

    

    def visualize_detection(self, frame: np.ndarray, detection_result: dict) -> np.ndarray:

        """可视化检测结果"""

        result_frame = frame.copy()

        

        # 绘制运动区域

        for region in detection_result['motion_regions']:

            x, y, w, h = region['bbox']

            

            # 绘制边界框

            cv2.rectangle(result_frame, (x, y), (x+w, y+h), (0, 0, 255), 2)

            

            # 绘制中心点

            cx, cy = region['center']

            cv2.circle(result_frame, (cx, cy), 5, (0, 0, 255), -1)

            

            # 显示区域信息

            cv2.putText(result_frame, f"Area: {region['area']}", 

                       (x, y-10), cv2.FONT_HERSHEY_SIMPLEX, 

                       0.5, (0, 0, 255), 2)

        

        # 显示状态信息

        status_color = (0, 0, 255) if detection_result['motion_detected'] else (0, 255, 0)

        status_text = "MOTION DETECTED" if detection_result['motion_detected'] else "No Motion"

        

        cv2.putText(result_frame, status_text, (10, 30),

                   cv2.FONT_HERSHEY_SIMPLEX, 1, status_color, 2)

        

        if detection_result['alert_triggered']:

            cv2.putText(result_frame, "ALERT: Continuous Motion!", (10, 60),

                       cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)

        

        return result_frame

class AnomalyDetector:

    """异常行为检测器"""

    

    def __init__(self, motion_detector: MotionDetector):

        self.motion_detector = motion_detector

        self.anomaly_history = []

        

        # 异常检测参数

        self.speed_threshold = 100  # 像素/帧

        self.direction_change_threshold = 60  # 度

        self.stagnant_threshold = 30  # 帧

        

        # 跟踪状态

        self.tracked_objects = {}

        self.next_track_id = 0

    

    def analyze_motion_patterns(self, current_regions: List[dict]) -> List[dict]:

        """分析运动模式，检测异常行为"""

        anomalies = []

        

        # 更新跟踪

        current_centers = [r['center'] for r in current_regions]

        

        # 简单的最近邻匹配

        matched_indices = set()

        for obj_id, obj_data in list(self.tracked_objects.items()):

            last_center = obj_data['last_center']

            last_time = obj_data['last_time']

            

            if current_centers:

                # 找到最近的当前点

                distances = [

                    np.linalg.norm(np.array(center) - np.array(last_center))

                    for center in current_centers

                ]

                min_idx = np.argmin(distances)

                

                if distances[min_idx] < 50:  # 匹配阈值

                    current_center = current_centers[min_idx]

                    matched_indices.add(min_idx)

                    

                    # 计算速度

                    time_delta = time.time() - last_time

                    if time_delta > 0:

                        speed = distances[min_idx] / time_delta

                        

                        # 检测高速移动

                        if speed > self.speed_threshold:

                            anomalies.append({

                                'type': 'high_speed',

                                'object_id': obj_id,

                                'speed': speed,

                                'position': current_center

                            })

                    

                    # 计算方向变化

                    if len(obj_data['trajectory']) >= 2:

                        vec1 = np.array(obj_data['trajectory'][-1]) - np.array(obj_data['trajectory'][-2])

                        vec2 = np.array(current_center) - np.array(obj_data['trajectory'][-1])

                        

                        if np.linalg.norm(vec1) > 0 and np.linalg.norm(vec2) > 0:

                            # 计算角度

                            dot_product = np.dot(vec1, vec2)

                            magnitude_product = np.linalg.norm(vec1) * np.linal