背景

最近实验室有个需求， 即将 多个摄像头采集到时视频流推送给VLM大模型，由VLM大模型对摄像头采集到的环境和用户状态信息进展实时感知和分析， 并将分析结果来和预设的标签信息进行映射，最终实现对训练数据自动打标签的功能验证， 因此用python 简单写了个实现多摄像头视频录制及间隔抓拍的脚本

脚本

import time
from concurrent.futures import ThreadPoolExecutor
import cv2
import yaml

config = {
    "interval": 20,  # 每个分片视频的录制长度， 单位s
    "camera_ids": ["cam01", "cam02", "cam03"],
    "res_path": "./imgs",
    "fps": [30, 30, 30]
}


def test_real_cam_fps(index):
    # 打开默认摄像头（摄像头索引通常是0）
    cap = cv2.VideoCapture(index)
    if not cap.isOpened():
        print("无法打开摄像头")
        exit()
    # 设置要计算的帧数（测试的帧数越多结果越稳定）
    num_frames = 60
    print(f"正在计算帧率，将采集 {num_frames} 帧...")
    # 初始读取几帧（某些摄像头需要预热）
    for _ in range(5):
        cap.read()
    # 记录开始时间
    start = time.time()
    # 循环读取指定帧数
    for _ in range(num_frames):
        ret, frame = cap.read()
        if not ret:
            break
    # 计算时间差和帧率
    end = time.time()
    seconds = end - start
    fps = num_frames / seconds
    print(f"实际帧率(FPS): {fps:.2f}")
    # 释放摄像头
    cap.release()
    return int(fps)



def record_and_capture(cap, device_name, config, fps):
    print(f"开始录制视频：{device_name}")

    try:
        # 获取摄像头默认分辨率
        frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        image_path = config["res_path"]

        # 设置视频编码器（推荐使用 MP4V 或 XVID）
        fourcc = cv2.VideoWriter_fourcc(*'XVID')  # 也可以使用 'mp4v' 保存为 mp4 格式

        output_filename = f"{image_path}/{device_name}/{device_name}_{time.strftime('%Y%m%d%H%M%S')}.avi"
        out = cv2.VideoWriter(output_filename, fourcc, fps, (frame_width, frame_height))
        start_time = time.time()
        interval_rate = config["interval"]
        while cap.isOpened():
            ret, frame = cap.read()
            if ret:
                # 写入帧到视频文件
                out.write(frame)
            else:
                print("无法获取视频帧")
                break
            # 图片抓拍
            time_now = time.time()
            if time_now - start_time > interval_rate:
                filename = f"{image_path}/{device_name}/{device_name}_{time.strftime('%Y%m%d%H%M%S')}.jpg"
                cv2.putText(frame, time.strftime("%Y-%m-%d %H:%M:%S"),
                            (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)
                cv2.imwrite(filename, frame)
                start_time = time_now
    except Exception as e:
        print(f"摄像头采集异常: {e}")


if __name__ == '__main__':
    print(cv2.__version__)
    try:
        with open("./config.yaml", 'r', encoding='utf-8') as f:
            user_config = yaml.load(f, Loader=yaml.FullLoader) or {}
            config = user_config
    except Exception as e:
        print("读取配置文件异常")

    camera_ids = config["camera_ids"]
    camera_fps = config["fps"]
    for index, element in enumerate(camera_ids):
        test_real_cam_fps(index)

    # 注意视频录制时fps设置需要按实际fps设置，  否则会存在录制的视频和实际时长有偏差
    with ThreadPoolExecutor(3) as t:
        for index, element in enumerate(camera_ids):
            cap = cv2.VideoCapture(index)
            if not cap.isOpened():
                print("无法打开摄像头")
                continue
            else:
                t.submit(record_and_capture, cap, element, config, camera_fps[index])



# pyinstaller --onefile --windowed --icon=app.ico main.py

问题及注意事项：

1： 一开始录制的帧率设置为20帧， 录制时发现录制的时长超过时间的时长， 定位排查后发现是设置的帧率和摄像头的实际帧率不一致导致， 实测帧率为30帧，修改后该问题解决