用 Python 实现智能摄像头监控：运动检测 + 自动录像 + 时间水印

关键词：Python、OpenCV、运动检测、帧差法、视频录制、时间戳、安防监控、边缘计算

在家庭安防、宠物看护、仓库监控等场景中，一个轻量、可靠、无需联网的本地摄像头监控系统非常实用。本文将带你从零实现一个基于 Python + OpenCV 的智能运动检测与自动录像系统——它能在检测到画面变化时自动开始录制视频，运动停止后继续缓冲几秒，并在保存的视频左上角添加精确的时间水印。

更重要的是：录制的视频是原始画面，不包含任何检测框或调试信息，可直接用于存档或回放！

✅ 项目亮点

• 三帧差分法：比传统两帧差分更稳定，减少“目标静止即消失”的问题
• 仅录有效片段：只在检测到运动时录像，节省存储空间
• ⏱️ 自动时间水印：每帧叠加 2026-01-12 21:07:35 格式时间戳
• 自动归档：视频按时间命名，存入 recordings/ 目录
• 即插即用：支持 USB 摄像头、树莓派 CSI 摄像头等
• 资源安全释放：异常退出也能正确关闭文件和设备

核心原理：为什么用“三帧差分”？

传统的两帧差分法（当前帧 vs 上一帧）在目标静止时会丢失轮廓，导致漏检。而三帧差分法通过计算：

highlighter- Ini

diff1 = |frame(t) - frame(t-1)|
diff2 = |frame(t+1) - frame(t)|
motion = diff1 ∩ diff2

只有连续两帧都发生变化的区域才被认为是“真实运动”，有效抑制了噪声和短暂干扰。

配合形态学开运算 + 膨胀，还能去除小噪点并填充运动区域空洞，大幅提升检测鲁棒性。

完整代码解析

以下是经过优化的完整实现（已注释关键逻辑）：

python

import cv2
import datetime
import os


CAMERA_INDEX = 1          
THRESHOLD = 25            
MIN_CONTOUR_AREA = 800    
RECORD_DIR = "recordings"
POST_MOTION_BUFFER_SEC = 3  
os.makedirs(RECORD_DIR, exist_ok=True)


cap = cv2.VideoCapture(CAMERA_INDEX)
if not cap.isOpened():
    raise IOError("无法打开摄像头，请检查设备连接")


ret, prev_frame = cap.read()
ret, curr_frame = cap.read()
ret, next_frame = cap.read()
if not all([ret, ret, ret]):
    raise RuntimeError("无法获取初始视频帧")


is_recording = False
video_writer = None
last_motion_time = datetime.datetime.now()

def three_frame_diff(prev, curr, nxt, thresh_val=25):
    """三帧差分 + 形态学优化"""
    gray_prev = cv2.cvtColor(prev, cv2.COLOR_BGR2GRAY)
    gray_curr = cv2.cvtColor(curr, cv2.COLOR_BGR2GRAY)
    gray_next = cv2.cvtColor(nxt, cv2.COLOR_BGR2GRAY)

    diff1 = cv2.absdiff(gray_curr, gray_prev)
    diff2 = cv2.absdiff(gray_next, gray_curr)

    _, bin1 = cv2.threshold(diff1, thresh_val, 255, cv2.THRESH_BINARY)
    _, bin2 = cv2.threshold(diff2, thresh_val, 255, cv2.THRESH_BINARY)

    combined = cv2.bitwise_and(bin1, bin2)
    kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
    combined = cv2.morphologyEx(combined, cv2.MORPH_OPEN, kernel)  
    combined = cv2.dilate(combined, kernel, iterations=1)         
    return combined

def add_timestamp_to_frame(frame):
    """在帧左上角添加时间水印（格式：2026-01-12 21:07:35）"""
    timestamp_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    cv2.putText(frame, timestamp_str, (10, 30),
                cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2, cv2.LINE_AA)
    cv2.putText(frame, timestamp_str, (10, 30),
                cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 0), 1, cv2.LINE_AA)
    return frame

try:
    print("\n[*] 运动检测开始")
    while True:
        ret, raw_frame = cap.read()
        if not ret: break

        display_frame = raw_frame.copy()  
        motion_mask = three_frame_diff(prev_frame, curr_frame, next_frame, THRESHOLD)

        
        contours, _ = cv2.findContours(motion_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        motion_detected = False
        for cnt in contours:
            if cv2.contourArea(cnt) > MIN_CONTOUR_AREA:
                x, y, w, h = cv2.boundingRect(cnt)
                cv2.rectangle(display_frame, (x, y), (x + w, y + h), (0, 255, 0), 2)
                cv2.putText(display_frame, "MOTION!", (x, y - 10),
                            cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
                motion_detected = True

        
        prev_frame = curr_frame.copy()
        curr_frame = next_frame.copy()
        ret, next_frame = cap.read()
        if not ret: break

        current_time = datetime.datetime.now()

        
        if motion_detected:
            last_motion_time = current_time
            if not is_recording:
                timestamp = current_time.strftime("%Y%m%d_%H%M%S")
                video_path = os.path.join(RECORD_DIR, f"motion_{timestamp}.mp4")
                fourcc = cv2.VideoWriter_fourcc(*'mp4v')
                h, w = raw_frame.shape[:2]
                video_writer = cv2.VideoWriter(video_path, fourcc, 20.0, (w, h))
                is_recording = True
                print(f"[+] 开始录制: {video_path}, {current_time.strftime('%Y-%m-%d %H:%M:%S')}")

        
        if is_recording:
            frame_to_save = raw_frame.copy()
            frame_to_save = add_timestamp_to_frame(frame_to_save)
            video_writer.write(frame_to_save)

            
            if (current_time - last_motion_time).total_seconds() > POST_MOTION_BUFFER_SEC:
                video_writer.release()
                is_recording = False
                print(f"[-] 停止录制, {current_time.strftime('%Y-%m-%d %H:%M:%S')}")

        
        
        
        

finally:
    
    if is_recording:
        video_writer.release()
    cap.release()
    cv2.destroyAllWindows()

注意：代码中已注释掉 cv2.imshow 部分，使其可在无图形界面的服务器或树莓派后台运行。如需调试，取消注释即可。

点击查看代码

highlighter- Python

import cv2
import datetime
import os


CAMERA_INDEX = 1          
THRESHOLD = 25            
MIN_CONTOUR_AREA = 800    
RECORD_DIR = "recordings"
POST_MOTION_BUFFER_SEC = 3  
os.makedirs(RECORD_DIR, exist_ok=True)


cap = cv2.VideoCapture(CAMERA_INDEX)
if not cap.isOpened():
    raise IOError("无法打开摄像头，请检查设备连接")


ret, prev_frame = cap.read()
ret, curr_frame = cap.read()
ret, next_frame = cap.read()
if not all([ret, ret, ret]):
    raise RuntimeError("无法获取初始视频帧")


is_recording = False
video_writer = None
last_motion_time = datetime.datetime.now()

def three_frame_diff(prev, curr, nxt, thresh_val=25):
    """三帧差分法 + 形态学优化"""
    gray_prev = cv2.cvtColor(prev, cv2.COLOR_BGR2GRAY)
    gray_curr = cv2.cvtColor(curr, cv2.COLOR_BGR2GRAY)
    gray_next = cv2.cvtColor(nxt, cv2.COLOR_BGR2GRAY)

    diff1 = cv2.absdiff(gray_curr, gray_prev)
    diff2 = cv2.absdiff(gray_next, gray_curr)

    _, bin1 = cv2.threshold(diff1, thresh_val, 255, cv2.THRESH_BINARY)
    _, bin2 = cv2.threshold(diff2, thresh_val, 255, cv2.THRESH_BINARY)

    combined = cv2.bitwise_and(bin1, bin2)
    kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
    combined = cv2.morphologyEx(combined, cv2.MORPH_OPEN, kernel)
    combined = cv2.dilate(combined, kernel, iterations=1)
    return combined

def add_timestamp_to_frame(frame):
    """在帧左上角添加中文格式时间水印"""
    timestamp_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    cv2.putText(
        frame,
        timestamp_str,
        (10, 30),
        cv2.FONT_HERSHEY_SIMPLEX,
        0.7,
        (255, 255, 255),   
        2,
        cv2.LINE_AA
    )
    cv2.putText(
        frame,
        timestamp_str,
        (10, 30),
        cv2.FONT_HERSHEY_SIMPLEX,
        0.7,
        (0, 0, 0),         
        1,
        cv2.LINE_AA
    )
    return frame

try:
    print("\n[*] 运动检测开始")
    while True:
        ret, raw_frame = cap.read()  
        if not ret:
            break

        
        display_frame = raw_frame.copy()

        
        motion_mask = three_frame_diff(prev_frame, curr_frame, next_frame, THRESHOLD)

        
        contours, _ = cv2.findContours(motion_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        motion_detected = False
        for cnt in contours:
            if cv2.contourArea(cnt) > MIN_CONTOUR_AREA:
                x, y, w, h = cv2.boundingRect(cnt)
                cv2.rectangle(display_frame, (x, y), (x + w, y + h), (0, 255, 0), 2)
                cv2.putText(display_frame, "MOTION!", (x, y - 10),
                            cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
                motion_detected = True

        
        prev_frame = curr_frame.copy()
        curr_frame = next_frame.copy()
        ret, next_frame = cap.read()
        if not ret:
            break

        current_time = datetime.datetime.now()

        
        if motion_detected:
            last_motion_time = current_time
            if not is_recording:
                timestamp = current_time.strftime("%Y%m%d_%H%M%S")
                video_path = os.path.join(RECORD_DIR, f"motion_{timestamp}.mp4")
                fourcc = cv2.VideoWriter_fourcc(*'mp4v')
                height, width = raw_frame.shape[:2]
                video_writer = cv2.VideoWriter(video_path, fourcc, 20.0, (width, height))
                is_recording = True
                timestamp_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                print(f"[+] 开始录制: {video_path},{timestamp_str}")

        
        if is_recording:
            frame_to_save = raw_frame.copy()
            frame_to_save = add_timestamp_to_frame(frame_to_save)
            video_writer.write(frame_to_save)

            
            if (current_time - last_motion_time).total_seconds() > POST_MOTION_BUFFER_SEC:
                video_writer.release()
                is_recording = False
                timestamp_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                print(f"[-] 停止录制,{timestamp_str}")

        
        
        
        
        

finally:
    if is_recording:
        video_writer.release()
    cap.release()
    cv2.destroyAllWindows()

️ 使用指南

1. 安装依赖

bash

pip install opencv-python

2. 修改摄像头索引

• CAMERA_INDEX = 0：笔记本内置摄像头
• CAMERA_INDEX = 1：外接 USB 摄像头
• 在 Linux 下可通过 ls /dev/video* 查看设备

3. 调整灵敏度

• 提高灵敏度：降低 THRESHOLD（如 15）或减小 MIN_CONTOUR_AREA
• 降低误报：提高 THRESHOLD（如 40）或增大 MIN_CONTOUR_AREA

4. 运行

bash

python motion_monitor.py

录制的视频将自动保存在 recordings/ 文件夹中，命名如：

highlighter-

motion_20260112_210735.mp4

应用场景扩展

• 家庭宠物监控：记录猫咪捣乱瞬间
• 无人值守店铺：夜间异常闯入报警
• 实验室/机房：设备状态异常记录
• 树莓派 + 移动电源：便携式野外监控站

若需进一步升级，可考虑：

• 添加微信/邮件通知（使用 smtplib 或企业微信 API）
• 支持 RTSP 网络摄像头
• 集成 YOLO 进行人/车分类
• 使用 FFmpeg 转 H.264 降低体积

✅ 总结

本文提供了一个轻量、高效、生产可用的 Python 视频监控方案。它不依赖深度学习模型，资源占用低，适合部署在边缘设备上。通过三帧差分与智能录像策略，既保证了检测准确性，又避免了无效视频堆积。

真正的智能，不在于复杂，而在于恰到好处的自动化。

欢迎点赞、收藏、转发！
如果你有改进想法或遇到问题，欢迎在评论区交流

__EOF__