opencv-python计算视频光流

光流基本概念

光流表示的是相邻两帧图像中每个像素的运动速度和运动方向。具体：光流是空间运动物体在观察成像平面上的像素运动的瞬时速度，是利用图像序列中像素在时间域上的变化以及相邻帧之间的相关性来找到上一帧跟当前帧之间存在的对应关系，从而计算出相邻帧之间物体的运动信息的一种方法。一般而言，光流是由于场景中前景目标本身的移动、相机的运动，或者两者的共同运动所产生的。

opencv光流算子

参考链接

光流法大全(DeepFlow、DenseFlow、DisFlow、FbFlow、PCAFlow、SimpleFlow、TV_L1)

视频光流计算demo

import os
import cv2
import flow_vis
import numpy as np
from tqdm import tqdm


def compute_flow(prev, curr, bound=15):

    '''Farneback optical flow'''
    # flow = cv2.calcOpticalFlowFarneback(prev=prev, next=curr, flow=None, pyr_scale=0.5, levels=5,winsize=15, iterations=3, poly_n=5, poly_sigma=1.1, flags=cv2.OPTFLOW_FARNEBACK_GAUSSIAN)

    ''' TVL1 optical flow（需安装opencv_contrib）'''
    TVL1 = cv2.optflow.DualTVL1OpticalFlow_create()
    flow = TVL1.calc(prev, curr, None)

    ''' deepflow optical flow （需安装opencv_contrib）'''
    # inst = cv2.optflow.createOptFlow_DeepFlow()
    # flow = inst.calc(prev, curr, None)

    ''' sparse to dense flow optical flow （需安装opencv_contrib）'''
    # flow = cv2.optflow.calcOpticalFlowSparseToDense(prev, curr)

    ''' pca flow optical flow （需安装opencv_contrib）'''
    # inst = cv2.optflow.createOptFlow_PCAFlow()
    # flow = inst.calc(prev, curr, None)

    ''' DIS optical flow '''
    # dis = cv2.DISOpticalFlow_create(2)
    # flow = dis.calc(prev, curr, None)

    return flow


def video2flow(video_path:str, flow_path:str):
    ''' 读取视频，获取视频基本信息 '''
    videoCapture = cv2.VideoCapture(video_path)
    if not videoCapture.isOpened():  # 若视频文件读取失败，读取下一段视频
        print('视频打开失败!!!')
        print(video_path)
        return False
    total_frames = int(videoCapture.get(cv2.CAP_PROP_FRAME_COUNT))  # 获取视频总帧数
    # fourcc = int(videoCapture.get(cv2.CAP_PROP_FOURCC))  # 原生不支持h264编码
    fps = videoCapture.get(cv2.CAP_PROP_FPS)  # 获取视频帧率
    w = int(videoCapture.get(cv2.CAP_PROP_FRAME_WIDTH))  # 获取图像宽度
    h = int(videoCapture.get(cv2.CAP_PROP_FRAME_HEIGHT))  # 获取图像高度

    '''光流视频 写入设置 '''
    # fourcc = cv2.VideoWriter_fourcc(*'MJPG')  # avi格式
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')  # MP4格式
    video_flow = cv2.VideoWriter(flow_path, fourcc, fps, frameSize=(w, h), isColor=True)

    ''' 计算光流并保存 '''
    success, prev = videoCapture.read()
    total_frames -= 1
    if not success:
        print('视频首帧读取失败')
        return False

    pbar = tqdm(total=total_frames)
    while total_frames:
        success, curr = videoCapture.read()
        if success:
            prev_gray = cv2.cvtColor(prev, cv2.COLOR_BGR2GRAY)
            curr_gray = cv2.cvtColor(curr, cv2.COLOR_BGR2GRAY)
            tmp_flow = compute_flow(prev_gray, curr_gray)
            rgb = flow_vis.flow_to_color(tmp_flow, convert_to_bgr=False)
            video_flow.write(rgb.astype(np.uint8))
            prev = curr

            # cv2.imshow('frame', flow_xy.astype(np.uint8))
            # cv2.waitKey(0)=='q'
        else:
            print('某中间帧读取失败，光流视频生成失败')
            return False
        pbar.update(1)
        total_frames -= 1

    videoCapture.release()
    video_flow.release()
    cv2.destroyAllWindows()

    return True


if __name__=='__main__':
    video_path = './forest.mp4'
    flow_path = './forest_flow.mp4'
    print(video2flow(video_path, flow_path))