PythonPython-OpenCV实时处理视频

Posted 2023-04-06 @苏丶

利用Python的opencv包实时处理视频（也可以处理视频文件），并输出到虚拟摄像头（类似直播时实时处理视频）。

由于视频处理即帧处理需要一定的时间，全部放在一个进程中会导致进程阻塞、视频卡顿，于是这里采用两个进程分别进行视频处理和推流到虚拟摄像头并通过队列的方式进行通信。

步骤

• 类与初始化
• 视频捕获预处理
• 视频帧输出到虚拟摄像头
• 进程设置
• 完整代码
• Unity插件
• 本文内容分享仅供学习，切勿用于商业或违法用途

类与初始化

这里构建一个LIVE类用于实现各种功能，不同的方法可能共用一些相同的参数，因此使用类的实例属性来保存这些参数。

class Live(object):
    # 初始化模板与视频（摄像头）参数
    def __init__(self, template, file, target):
        # 定义opencv所需的模板
        self.template = template
        # 读取模板并获取模板的高度和宽度
        self.template = cv2.imread(self.template, 1)
        # 视频路径，0时调用摄像头
        self.file = file
        # 视频对象，0时调用摄像头
        cap = cv2.VideoCapture(self.file)
        # 视频高度
        self.height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        # 视频宽度
        self.width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        # 视频尺寸
        self.size = f'self.widthxself.height'
        # 视频帧率
        self.fps = int(cap.get(cv2.CAP_PROP_FPS))
        del cap

        self.target = target  # 裁剪坐标和尺寸（x,y,width,height）
        self.targetSize = f'self.target[2]xself.target[3]'

这里主要以实时裁剪视频的例子来讲解，所以self.target保存了视频裁剪的坐标和尺寸。

视频捕获预处理

这里使用opencv捕获摄像头的画面（或读取视频文件）为一个个帧，将帧进行处理（裁剪到目标尺寸）后，最后put进队列中。

    def image_put(self, queue, filepath):
        while 1:
            cap = cv2.VideoCapture(filepath)
            if filepath == 0:  # 摄像头时设置分辨率
                cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.width)
                cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.height)
            print(f'Source Video Parameters: size: self.size fps: self.fps.')

            while cap.isOpened():
                ret, frame = cap.read()
                if not ret:
                    if filepath == 0:
                        print("Opening camera is failed!!!")
                    else:
                        print("Video is over.")
                    queue.put(cv2.resize(self.template, (self.target[0], self.target[1]), cv2.INTER_AREA))  # 放一张模板
                    queue.put('stop')
                    break
                queue.put(frame[self.target[1]:self.target[1] + self.target[3],
                          self.target[0]:self.target[0] + self.target[2]])

此方法后面会作为一个进程的方法来调用。

视频帧输出到虚拟摄像头

这里使用Unity Video Capture作为虚拟摄像头的插件，插件下载链接我会放在文末，官网也能下载到。

    def image_get(self, queue):
        cam = pyvirtualcam.Camera(width=self.target[2], height=self.target[3], fps=self.fps)
        print(f'Virtual Camera: cam.device.')
        print(f'Virtual Camera Parameters: size: self.targetSize fps: self.fps.')
        while True:
            frame = queue.get()
            if type(frame) == type('stop'):
                break
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)  # BGR TO RGB
            frame = cv2.flip(frame, 1)  # 水平翻转
            cam.send(frame)
            cam.sleep_until_next_frame()

利用pyvirtualcam包初始化虚拟摄像头，并使用该包的send方法将get到的视频帧推到虚拟摄像头上。

进程设置

首先创建一个队列用于保存视频帧，然后将视频捕获预处理方法和视频帧输出到虚拟摄像头方法分别用一个进程来实现。

    def run_single_camera(self):
        queue = mp.Queue(maxsize=2)  # 队列
        processes = [mp.Process(target=self.image_put, args=(queue, self.file)),
                     mp.Process(target=self.image_get, args=(queue,))]
        [process.start() for process in processes]
        [process.join() for process in processes]

完整代码

import cv2
import pyvirtualcam
import multiprocessing as mp


class Live(object):
    # 初始化模板与视频（摄像头）参数
    def __init__(self, template, file, target):
        # 定义opencv所需的模板
        self.template = template
        # 读取模板并获取模板的高度和宽度
        self.template = cv2.imread(self.template, 1)
        # 视频路径，0时调用摄像头
        self.file = file
        # 视频对象，0时调用摄像头
        cap = cv2.VideoCapture(self.file)
        # 视频高度
        self.height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        # 视频宽度
        self.width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        # 视频尺寸
        self.size = f'self.widthxself.height'
        # 视频帧率
        self.fps = int(cap.get(cv2.CAP_PROP_FPS))
        del cap

        self.target = target  # 截图尺寸
        self.targetSize = f'self.target[2]xself.target[3]'

    # 此线程用于视频捕获处理
    def image_put(self, queue, filepath):
        while 1:
            cap = cv2.VideoCapture(filepath)
            if filepath == 0:  # 摄像头时设置分辨率
                cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.width)
                cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.height)
            print(f'Source Video Parameters: size: self.size fps: self.fps.')

            while cap.isOpened():
                ret, frame = cap.read()
                if not ret:
                    if filepath == 0:
                        print("Opening camera is failed!!!")
                    else:
                        print("Video is over.")
                    queue.put(cv2.resize(self.template, (self.target[0], self.target[1]), cv2.INTER_AREA))  # 放一张模板
                    queue.put('stop')
                    break
                queue.put(frame[self.target[1]:self.target[1] + self.target[3],
                          self.target[0]:self.target[0] + self.target[2]])

    # 此线程用于推流
    def image_get(self, queue):
        cam = pyvirtualcam.Camera(width=self.target[2], height=self.target[3], fps=self.fps)
        print(f'Virtual Camera: cam.device.')
        print(f'Virtual Camera Parameters: size: self.targetSize fps: self.fps.')
        while True:
            frame = queue.get()
            if type(frame) == type('stop'):
                break
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)  # BGR TO RGB
            frame = cv2.flip(frame, 1)  # 水平翻转
            cam.send(frame)
            cam.sleep_until_next_frame()

    # 多线程执行一个摄像头
    def run_single_camera(self):
        queue = mp.Queue(maxsize=2)  # 队列
        processes = [mp.Process(target=self.image_put, args=(queue, self.file)),
                     mp.Process(target=self.image_get, args=(queue,))]
        [process.start() for process in processes]
        [process.join() for process in processes]


if __name__ == '__main__':
    mp.freeze_support()
    template_path = 'template.jpg'
    # file_path = "xxx.mp4"
    file_path = 0  # 摄像头
    real_camera_size = (640, 480)
    target_size = (380, 300)
    target = ((real_camera_size[0] - target_size[0]) // 2, (real_camera_size[1] - target_size[1]) // 2, target_size[0],
              target_size[1])
    live = Live(template=template_path, file=file_path, target=target)
    live.run_single_camera()

代码执行时要在代码或者生成的exe目录下放置一张jpg模板（template_path = ‘template.jpg’）。

Unity插件

Unity插件下载链接

本文内容分享仅供学习，切勿用于商业或违法用途

本文内容分享仅供学习，切勿用于商业或违法用途

Flink视频教程_基于Flink流处理的动态实时电商实时分析系统

Flink视频教程_基于Flink流处理的动态实时电商实时分析系统

课程分享地址链接:https://pan.baidu.com/s/1cX7O-45y6yUPT4B-ACfliA
密码:jqmk

在开始学习前给大家说下什么是Flink？
1.Flink是一个针对流数据和批数据的分布式处理引擎，主要用Java代码实现。
2.Apache Flink作为Apache的顶级项目，Flink集众多优点于一身，包括快速、可靠可扩展、完全兼容Hadoop、使用简便、表现卓越通过以上的描述大家对Flink有了一个基本的认识，本套课程不会讲解基础内容，因此建议有Flink基础的同学进行认购。

本课程将基于真实的电商分析系统构建，通过Flink实现真正的实时分析，该系统会从无到有一步一步带大家实现，让大家在实操中快速掌握Flink技术。

课程所涵盖的知识点包括
Flink、Kafka、Flume、Sqoop、SpringMVC、Redis、HDFS、Mapreduce、Hbase、Hive、SpringBoot、SpringCloud等等
学完该课程大家会对Flink有非常深入的了解，同时可以体会到Flink的强大之处，以及可以结合自己公司的业务进行使用，减少自己研究和学习Flink的时间。