保姆级教程：用Python+One-Class SVM给监控视频做‘体检’，快速识别打架、偷窃等异常行为

实战指南：基于Python与One-Class SVM的监控视频异常行为识别系统

监控摄像头已经遍布城市的各个角落，但如何让这些"电子眼"真正具备智能识别能力？想象一下，当监控系统能够自动识别打架、偷窃等异常行为并实时报警，这将为安保工作带来革命性的改变。本文将带你从零开始，用Python和机器学习构建这样一个智能监控原型系统。

1. 环境准备与数据获取

在开始之前，我们需要准备好开发环境。推荐使用Python 3.8+版本，并创建一个独立的虚拟环境：

python -m venv anomaly_detection source anomaly_detection/bin/activate # Linux/Mac # 或者 anomaly_detection\Scripts\activate # Windows

接下来安装必要的依赖库：

pip install opencv-python tensorflow scikit-learn matplotlib numpy

UCF-Crime数据集是异常检测领域的基准数据集，包含1900个长短不一的监控视频片段，总时长超过128小时。数据集中的异常行为被细分为13类，包括：

• 打架斗殴
• 偷窃行为
• 交通事故
• 纵火
• 抢劫等

重要提示：由于数据集较大（约35GB），建议使用稳定的网络连接下载。官方下载地址可通过邮件申请获取，下载后建议按以下结构组织文件：

data/ ├── train/ │ ├── Normal/ │ └── Anomaly/ └── test/ ├── Normal/ └── Anomaly/

2. 视频预处理与特征提取

监控视频本质上是连续的图像序列，我们需要将其转换为适合机器学习处理的格式。首先，使用OpenCV提取视频帧：

import cv2 import os def extract_frames(video_path, output_dir, frame_interval=10): """提取视频关键帧""" if not os.path.exists(output_dir): os.makedirs(output_dir) cap = cv2.VideoCapture(video_path) frame_count = 0 saved_count = 0 while True: ret, frame = cap.read() if not ret: break if frame_count % frame_interval == 0: save_path = os.path.join(output_dir, f"frame_{saved_count:04d}.jpg") cv2.imwrite(save_path, frame) saved_count += 1 frame_count += 1 cap.release() return saved_count

提示：设置frame_interval参数可以控制帧采样率，值越大处理速度越快但可能丢失关键信息，建议根据视频帧率调整。

接下来是特征提取环节。我们使用VGG16的预训练模型，但需要注意几个常见陷阱：

1. 输入尺寸问题：VGG16要求输入为224x224的RGB图像
2. 预处理规范：必须使用与训练时相同的预处理方式
3. 特征层选择：通常使用最后一个卷积层的输出

from tensorflow.keras.applications.vgg16 import VGG16, preprocess_input from tensorflow.keras.preprocessing import image import numpy as np # 加载预训练模型（不包括全连接层） model = VGG16(weights='imagenet', include_top=False, pooling='avg') def extract_features(img_path): """提取单张图像的特征向量""" img = image.load_img(img_path, target_size=(224, 224)) x = image.img_to_array(img) x = np.expand_dims(x, axis=0) x = preprocess_input(x) features = model.predict(x) return features.flatten()

性能优化技巧：在实际应用中，可以使用批量预测来提升效率：

def batch_extract_features(img_paths, batch_size=32): """批量提取特征""" all_features = [] for i in range(0, len(img_paths), batch_size): batch_paths = img_paths[i:i+batch_size] batch_images = [] for path in batch_paths: img = image.load_img(path, target_size=(224, 224)) x = image.img_to_array(img) x = preprocess_input(x) batch_images.append(x) batch_images = np.array(batch_images) batch_features = model.predict(batch_images) batch_features = batch_features.reshape(batch_features.shape[0], -1) all_features.extend(batch_features) return np.array(all_features)

3. One-Class SVM模型构建与训练

One-Class SVM是一种专门用于异常检测的无监督学习算法。它的核心思想是：在特征空间中找到一个超球面，使得大多数正常数据点都位于球内，而异常点则位于球外。

3.1 模型训练流程

from sklearn.svm import OneClassSVM from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler # 构建处理管道 model = Pipeline([ ('scaler', StandardScaler()), ('ocsvm', OneClassSVM(nu=0.1, kernel='rbf', gamma='auto')) ]) # 假设train_features是从正常视频中提取的特征矩阵 model.fit(train_features)

参数说明：

3.2 处理类别不平衡问题

监控视频中异常帧占比通常不足5%，这种极端不平衡会影响模型性能。我们可以采用以下策略：

1. 数据重采样：

   • 对少数类（异常）进行过采样
   • 对多数类（正常）进行欠采样

2. 集成方法：

   • 训练多个子模型，每个使用不同的正常数据子集
   • 通过投票机制综合判断

from sklearn.ensemble import IsolationForest from sklearn.neighbors import LocalOutlierFactor # 创建模型集合 models = { 'ocsvm': OneClassSVM(nu=0.05), 'isolation_forest': IsolationForest(contamination=0.05), 'lof': LocalOutlierFactor(n_neighbors=20, novelty=True) } # 训练每个模型 for name, model in models.items(): model.fit(train_features)

4. 实时检测与报警系统实现

将训练好的模型部署到实际应用中，需要考虑实时性和可视化反馈。以下是关键实现步骤：

4.1 实时视频处理框架

import cv2 import time from queue import Queue from threading import Thread class VideoProcessor: def __init__(self, model, alert_callback=None): self.model = model self.alert_callback = alert_callback self.frame_queue = Queue(maxsize=30) self.running = False def start(self, video_source=0): self.running = True self.capture_thread = Thread(target=self._capture_frames, args=(video_source,)) self.process_thread = Thread(target=self._process_frames) self.capture_thread.start() self.process_thread.start() def stop(self): self.running = False self.capture_thread.join() self.process_thread.join() def _capture_frames(self, video_source): cap = cv2.VideoCapture(video_source) while self.running: ret, frame = cap.read() if not ret: break if not self.frame_queue.full(): self.frame_queue.put(frame) cap.release() def _process_frames(self): while self.running or not self.frame_queue.empty(): if self.frame_queue.empty(): time.sleep(0.1) continue frame = self.frame_queue.get() # 特征提取和预测 features = extract_features_from_frame(frame) # 实现略 prediction = self.model.predict([features]) if prediction == -1: # 异常 annotated_frame = self._mark_anomaly(frame) if self.alert_callback: self.alert_callback(annotated_frame) def _mark_anomaly(self, frame): """在帧上标记异常区域""" # 简单实现：添加红色边框和文字 marked = cv2.copyMakeBorder(frame, 10, 10, 10, 10, cv2.BORDER_CONSTANT, value=(0, 0, 255)) cv2.putText(marked, "ALERT: Anomaly Detected!", (20, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2) return marked

4.2 性能优化策略

1. 多线程处理：

   • 独立线程负责视频采集
   • 独立线程负责模型推理

2. 帧采样策略：

   • 动态调整处理频率
   • 异常时提高采样率

3. 模型轻量化：

   • 使用MobileNet替代VGG16
   • 量化模型权重

# 使用轻量级MobileNetV2 from tensorflow.keras.applications.mobilenet_v2 import MobileNetV2, preprocess_input mobile_net = MobileNetV2(weights='imagenet', include_top=False, pooling='avg')

5. 系统评估与调优

构建好系统后，我们需要科学评估其性能。常用的评估指标包括：

from sklearn.metrics import classification_report, confusion_matrix # 假设y_true是真实标签，y_pred是预测结果 print(classification_report(y_true, y_pred)) print("\nConfusion Matrix:") print(confusion_matrix(y_true, y_pred))

常见问题排查指南：

1. 高误报率：

   • 检查训练数据是否纯净
   • 调整模型nu参数
   • 增加正常场景的多样性

2. 漏检严重：

   • 检查异常样本是否具有代表性
   • 尝试不同的特征提取方法
   • 考虑使用半监督学习

3. 处理延迟高：

   • 优化特征提取流程
   • 降低输入分辨率
   • 使用更高效的模型架构

在实际项目中，我们还需要考虑系统的可扩展性。当监控点位增多时，可以采用分布式处理架构：

[摄像头] -> [边缘节点：初步分析] -> [中心服务器：深度分析] -> [报警终端]

这种架构既能减轻中心服务器的压力，又能保证关键分析的准确性。边缘节点可以运行轻量级模型进行初步筛选，只有可疑视频才会被上传到中心服务器进行深度分析。