小小林熬夜学编程

这个屌丝很懒，什么也没留下！

热门标签

[常用工具] Python视频处理库VidGear使用指北

作者：小小林熬夜学编程 | 2024-05-29 06:55:23

踩

vidgear

VidGear是一个高性能的Python视频处理库，它在预载多个专业视频图像处理库的基础上，如OpenCV、FFmpeg、ZeroMQ、picamera、starlette、yt_dlp、pyscreenshot、aiortc和Python mss等，提供了一个易于使用、高度可扩展、彻底优化的多线程且异步的API框架。VidGear主要关注简单性，让软件开发人员只需几行代码即可轻松集成和执行复杂的视频处理任务，同时提供稳健的错误处理和实时处理性能。

以下功能框图清楚地描述了 VidGear API 的一般功能，简单来说就是在客户端的各种设备上采集视频图像数据，然后转换为视频流，通过网络传输或者云端传输返回给远端高性能服务器，进行人工智能识别。然后远端服务器再将识别结果传输给本地客户端，达到实时检测显示的效果。在这个过程中VidGear可以实时读取、写入、处理、发送和接收来自各种设备的视频文件/帧/流。

VidGear的官方仓库见vidgear，VideGear官方文档库见：vidgear_doc

1 前置知识

1.1 安装

对于VidGear，python版本需要高于3.7。VidGear支持以下系统：

2016以后的linux版本，推荐使用linux系统运行VidGear
Windows 7及以上版本
MacOS 10.12.6及以上版本

在安装VidGear需要安装依赖库：

安装opencv-python

pip install opencv-python
linux下安装uvloop，使得系统获得更好性能：

pip install uvloop
安装ffmpeg，其它系统自行搜索安装方法，ubuntu下直接输入：

sudo apt install ffmpeg

VidGear直接使用pip安装即可，具体如下：

安装核心库（对机器性能要求较低）：

pip install -U vidgear[core]
安装核心库和异步依赖项（对机器性能要求较高）：

pip install -U vidgear[asyncio]

1.2 OpenCV-Python视频读写升级

OpenCV-Python中提供了视频读写接口，但是相比OpenCV-Python，VidGear在OpenCV基础上提供了更稳定，更高效的视频读写接口。

OpenCV和VidGear视频读取接口对比见：

任务	OpenCV	VidGear
初始化	stream = cv2.VideoCapture(path)	stream = CamGear(source=path).start()
取帧	(grabbed, frame) = stream.read()	frame = stream.read()
状态	if not grabbed:	if frame is None:
终止	stream.release()	stream.stop()

以下是OpenCV读取本机摄像头的代码，读取100帧平均每帧耗时33.31ms。

# 加载库
import cv2
import time


stream = cv2.VideoCapture(0)

totalTime = 0
num = 100

# 循环读取100帧
for i in range(num):

    start = time.time()
    (grabbed, frame) = stream.read()

    end = time.time()
    totalTime += (end-start)

    if not grabbed:
        break

    cv2.imshow("Output", frame)

    # 按q退出
    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

print("平均每帧读取时间为{:.2f}ms".format(1000.0*totalTime/num))

# 关闭窗口
cv2.destroyAllWindows()

# 释放视频流
stream.release()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

以下是VidGear读取本机摄像头的代码，读取100帧平均每帧耗时30.22ms。

# 加载库
from vidgear.gears import CamGear
import cv2
import time


stream = CamGear(source=0).start()

totalTime = 0
num = 100

# 循环读取100帧
for i in range(num):

    start = time.time()
    frame = stream.read()

    end = time.time()
    totalTime += (end-start)

    if frame is None:
        break

    cv2.imshow("Output", frame)

    # 按q退出
    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

print("平均每帧读取时间为{:.2f}ms".format(1000.0*totalTime/num))

# 关闭窗口
cv2.destroyAllWindows()

# 释放视频流
stream.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

OpenCV和VidGear视频写入接口对比见：

任务	OpenCV	VidGear
初始化	cv2.VideoWriter(‘output.avi’, cv2.VideoWriter_fourcc(*‘XVID’), 20.0, (640, 480))	writer = WriteGear(output_filename=‘Output.mp4’)
写入	writer.write(frame)	writer.write(frame)
状态	if not grabbed:	if frame is None:
终止	stream.release()	writer.close()

以下是OpenCV写入视频的代码，平均每帧写入耗时4.69ms。

# 加载库
import cv2
import time

stream = cv2.VideoCapture(0)

# 指定视频编码格式
fourcc = cv2.VideoWriter_fourcc(*'XVID')
# 初始化，视频存储地址，编码格式，帧率，图像大小
writer = cv2.VideoWriter('output.avi', fourcc, 20.0, (640, 480))

totalTime = 0
num = 100

# 循环读取100帧
for i in range(num):

    (grabbed, frame) = stream.read()

    if not grabbed:
        break

    start = time.time()
    # 写入图像
    writer.write(frame)
    end = time.time()

    cv2.imshow("Output", frame)

    totalTime += (end-start)

    # 按q退出
    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

print("平均每帧写入时间为{:.2f}ms".format(1000.0*totalTime/num))

# 关闭窗口
cv2.destroyAllWindows()

# 释放视频流
stream.release()

# 关闭文件对象
writer.release()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

以下是VidGear写入视频的代码，平均每帧写入耗时2.42ms。

# 加载库
from vidgear.gears import CamGear
from vidgear.gears import WriteGear

import cv2
import time


stream = CamGear(source=0).start()

totalTime = 0
num = 100

# 初始化
writer = WriteGear(output_filename='Output.mp4')


# 循环读取100帧
for i in range(num):

    frame = stream.read()

    if frame is None:
        break

    start = time.time()
    writer.write(frame)
    end = time.time()
    totalTime += (end-start)
    
    cv2.imshow("Output", frame)
    
    # 按q退出
    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break
    
print("平均每帧写入时间为{:.2f}ms".format(1000.0*totalTime/num))

# 关闭窗口
cv2.destroyAllWindows()

# 释放视频流
stream.stop()

# 关闭写入
writer.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

1.3 VidGear模块介绍

VidGear也即video gear，VidGear包含多个模块，在VideGear中这些模块被称为gear api。每个Gears专门用于处理/控制/处理不同数据特定和设备特定的视频流、网络流和媒体编码器/解码器，并提供独立的api接口。

按照用途这些gear可分为以下几类：

视频读取：从各种来源读取图像，转换为numpy.ndarray格式
- CamGear: 针对各种USB摄像头/网络流/流媒体站点的多线程API
- PiGear：针对各种Raspberry Pi摄像机模块的多线程API
- VideoGear：针对视频稳流的API
- ScreenGear：针对截屏的多线程API
视频写入：将numpy.ndarray格式图像写入到视频文件或网络流
- WriteGear：针对图像写入视频的API
流媒体传输：转发或广播用于流媒体的文件
- StreamGear：将源视频/音频文件和实时视频帧转为各种格式的视频流
- WebGear：异步接口，用于广播实时MJPEG数据
- WebGear_RTC：异步接口，用于广播WEBRTC数据
网络传输：通过连接的网络发送或接收数据
- NetGear: 用于处理互连网络系统之间的高性能数据传输
- NetGear_Async：NetGear的异步版本

2 CamGear

CamGear支持多种视频流，几乎可以处理或控制任何IP/USB摄像机、多媒体视频文件格式（测试高达4k）、任何网络流URL，如http（s）、rtp、rstp、rtmp、mms等。此外，它还支持各种直播视频流网站，如YouTube等。
下图展示了CamGear的功能，CamGear实际就是包装了OpenCV的VideoCapture API，允许同时解析多个视频，然后进行线程优化，提高稳定性和可用性。

2.1 基础使用

以下代码为读取本地视频和摄像头示例

# 加载库
from vidgear.gears import CamGear
import cv2


# 打开视频
# stream = CamGear(source=0).start() # 打开摄像头
stream = CamGear(source="test.mp4").start()

# 循环播放
while True:

    # 读图
    frame = stream.read()

    # 读图是否成功
    if frame is None:
        break

    # 图像处理

    # 展示图片
    cv2.imshow("Output", frame)

    # q退出
    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

# 关闭窗口
cv2.destroyAllWindows()

# 关闭视频
stream.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

2.2 读取流媒体

在设置视频源的时候指定stream_mode=True，表示设置读取流媒体，可以从各种视频网站和音乐网站读取数据。该功能主要通过yt-dlp实现，yt-dlp是一个非常出名用于网络媒体网站下载的python库。yt-dlp支持读取的网络媒体网站见：yt-dlp-supportedsites，由于一些原因，其中支持的网络媒体网站可能无法读取。yt-dlp支持的媒体网站链接实例可以见yt_dlp|extractor，每个py文件代表一个网站读取实例，打开某个py文件后有调用链接示例。

# 加载库
from vidgear.gears import CamGear
import cv2


# STREAM_RESOLUTION设置清晰度：360p, 720p, best, worst
# nocheckcertificate设置不采用SSL验证
# THREAD_TIMEOUT设置超时时间为300s
options = {"STREAM_RESOLUTION": "720p", "STREAM_PARAMS": {"nocheckcertificate": True}, "THREAD_TIMEOUT": 300}

# stream_mode设置打开媒体网站
stream = CamGear(
    source="https://abcnews.go.com/US/video/climate-change-forest-management-make-wildfires-harder-73318307", 
    stream_mode=True,
    **options).start()

# 循环播放
while True:

    # 读图
    frame = stream.read()

    # 读图是否成功
    if frame is None:
        break

    # 图像处理

    # 展示图片
    cv2.imshow("Output", frame)

    # q退出
    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

# 关闭窗口
cv2.destroyAllWindows()

# 关闭视频
stream.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

2.3 与OpenCV互通

通过设置Options可以设置OpenCV中VideoCapture的CAP_PROP类别参数，但是参数是否起作用取决摄像头。具体支持的参数介绍见：CamGear视频设置参数

# 加载库
from vidgear.gears import CamGear
import cv2


# 设置OpenCV VideoCapture参数，有些参数可能不起作用，主要取决相机
options = {
    "CAP_PROP_FRAME_WIDTH": 1000, 
    "CAP_PROP_FRAME_HEIGHT": 240,
    "CAP_PROP_FPS": 10, 
}

# logging输出日志
stream = CamGear(source=0, logging=True, **options).start()

# 循环播放
while True:

    # 读图
    frame = stream.read()

    # 读图是否成功
    if frame is None:
        break

    # 图像处理

    # 展示图片
    cv2.imshow("Output", frame)

    # q退出
    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

# 关闭窗口
cv2.destroyAllWindows()

# 关闭视频
stream.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

2.4 色彩空间转换

在获取图像时可以直接进行图像色彩空间转换，实际就是vidgear在函数内部调用opencv的cvtcolor方法，好处就是速度比opencv实现快一点。

from vidgear.gears import CamGear
import cv2

# 直接读取图像得到的是OpenCV的BGR图像，设置colorspace即可转换图像空间。
stream = CamGear(source=0, colorspace="COLOR_BGR2HSV", logging=True).start()


while True:

    frame = stream.read()

    if frame is None:
        break

    cv2.imshow("Output", frame)

    key = cv2.waitKey(1) & 0xFF

    # 按w将图像空间转为gray
    if key == ord("w"):
        stream.color_space = cv2.COLOR_BGR2GRAY  
    # 按e转换图像空间为lab
    if key == ord("e"):
        stream.color_space = cv2.COLOR_BGR2LAB  

    # 按s不转换图像空间
    if key == ord("s"):
        stream.color_space = None  

    if key == ord("q"):
        break

cv2.destroyAllWindows()

stream.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

2.5 同时播放视频

vidgear可以同时并行播放多个视频，至于播放多少个视频取决于机器性能。


from vidgear.gears import CamGear
import cv2
import time

# 视频流1
stream1 = CamGear(source=0, logging=True).start() 

# 视频流2
stream2 = CamGear(source="test.mp4", logging=True).start() 


while True:

    # 分别读取视频
    frameA = stream1.read()

    frameB = stream2.read()

    # 检测视频
    if frameA is None or frameB is None:
        break

    # 分别显示视频
    cv2.imshow("Output Frame1", frameA)
    cv2.imshow("Output Frame2", frameB)

    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

    # 按w设置不同功能
    if key == ord("w"):
        # 保存图像到本地
        cv2.imwrite("Image-1.jpg", frameA)
        cv2.imwrite("Image-2.jpg", frameB)
        # break   # 退出

# 关闭窗口
cv2.destroyAllWindows()

# 关闭视频
stream1.stop()
stream2.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

2.6 视频流读取

在这篇文章中OpenCV获取网络摄像头实时视频流描述了如何用OpenCV获取实时视频流，但是这种方式不稳定。vidgear提供了更加稳定的视频流读取方案。具体的demo如下，在下面的demo中其实用CamGear就可以直接读取视频流，但是视频流经常中断，所以增加了重连代码。视频流可以是rtmp，rtsp，http等主流协议。相关公开测试视频流地址为：

rtsp流：rtsp://wowzaec2demo.streamlock.net/vod/mp4:BigBuckBunny_115k.mp4，来源于https://www.wowza.com/developer/rtsp-stream-test
http流：http://devimages.apple.com.edgekey.net/streaming/examples/bipbop_4x3/gear2/prog_index.m3u8，来源于苹果提供的测试源

from vidgear.gears import CamGear
import cv2
import datetime
import time


class Reconnecting_CamGear:
    def __init__(self, cam_address, reset_attempts=50, reset_delay=5):
        self.cam_address = cam_address # 视频流地址
        self.reset_attempts = reset_attempts
        self.reset_delay = reset_delay
        self.source = CamGear(source=self.cam_address).start()
        self.running = True # 是否运行
        self.frame = None # 读取的帧

    def read(self):
        # 如果没有读到图
        if self.source is None:
            return None
        if self.running and self.reset_attempts > 0:
            frame = self.source.read() # 读图
            if frame is None:
                self.source.stop() # 没有读到图就停止
                self.reset_attempts -= 1 # 重新连接
                # 打印重新连接信息
                print(
                    "Re-connection Attempt-{} occured at time:{}".format(
                        str(self.reset_attempts),
                        datetime.datetime.now().strftime("%m-%d-%Y %I:%M:%S%p"),
                    )
                )
                time.sleep(self.reset_delay)
                self.source = CamGear(source=self.cam_address).start()
                # 返回前一帧
                return self.frame
            else:
                self.frame = frame # 返回结果
                return frame
        else:
            return None

    def stop(self):
        self.running = False
        self.reset_attempts = 0
        self.frame = None
        if not self.source is None:
            self.source.stop()


if __name__ == "__main__":
    # 打开可用视频流
    stream = Reconnecting_CamGear(
        cam_address="rtsp://wowzaec2demo.streamlock.net/vod/mp4:BigBuckBunny_115k.mp4", # 视频流地址
        reset_attempts= 5, # 重连次数
        reset_delay=3, # 重连后等待时间s
    )

    # 循环读取
    while True:

        # 读图
        frame = stream.read()

        # 判断图像是否存在
        if frame is None:
            break


        cv2.imshow("Output", frame)

        key = cv2.waitKey(1) & 0xFF
        if key == ord("q"):
            break

    cv2.destroyAllWindows()

    stream.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

3 VideoGear

3.1 VideoGear的使用

下图展示了VideoGear的功能，即读取视频，然后进行视频稳流。想要知道视频稳流具体如何实现可以阅读：基于特征点匹配的视频稳像

VideoGear内部封装了CamGear和视频稳像的代码，其函数调用和CamGear差不多，可以设置的参数也是一样的。

# 调用库
from vidgear.gears import VideoGear
import cv2

# VideoGear和CamGear调用方法一样，参数设置也是一样
stream = VideoGear(source=0, colorspace="COLOR_BGR2HSV", logging=True).start()
1
2
3
4
5
6

此外VideoGear还支持和ROS机器人操作系统一起使用，可以处理rtsp等视频流。但是VideoGear速度很慢，如果没有视频稳流的需求，可以仅使用CamGear。

3.2 Stabilizer的使用

可以不使用VideoGear直接调用VidGear中的Stabilizer类实现视频稳像。Stabilizer类能够以最小延迟实现vidgear的视频稳定，并且几乎不需要额外的计算需求。其基本思想是跟踪并保存给定帧数的显著特征阵列，然后使用这些定位点来抵消队列中传入帧相对于它的所有扰动。处理低频抖动，Stabilizer类效果还不错，高频的话效果很差。

基础使用

from vidgear.gears.stabilizer import Stabilizer
import cv2

# 打开摄像头
stream = cv2.VideoCapture(0)

# 打开stabilizer稳定器
# smoothing_radius默认为25，用于平滑帧间距离，数值越大，平滑效果越明显，但是延迟越大
# crop_n_zoom用于稳定数据
# border_size扩展边框大小的值，以补偿稳定期间黑色边框的减少
stab = Stabilizer(smoothing_radius=30, crop_n_zoom=True, border_size=5, logging=True)

while True:

    # 读图
    (grabbed, frame) = stream.read()

    if not grabbed:
        break

    # 稳定图像，初始几帧可能不会输出结果，因为要统计数据
    stabilized_frame = stab.stabilize(frame)

    if stabilized_frame is None:
        continue

    cv2.imshow("Stabilized Frame", stabilized_frame)

    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

cv2.destroyAllWindows()

stab.clean()

stream.release()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

4 ScreenGear

ScreenGear专为超快屏幕播放而设计，这意味着它在尽可能缩小时间延迟的情况下，通过在计算机屏幕上定义一个区域或全屏来实时抓取监视器中的图像。ScreenGear主要是基于pyscreenshot和python-mss实现的。

4.1 基础使用

以下代码为ScreenGear截取本机全屏屏幕的示例，延迟还是有点高。

from vidgear.gears import ScreenGear
import cv2
# 开始截屏
stream = ScreenGear().start()

# 循环播放
while True:

    # 读图
    frame = stream.read()

    # 读图是否成功
    if frame is None:
        break

    # 图像处理

    # 展示图片
    cv2.imshow("Output", frame)

    # q退出
    key = cv2.waitKey(100) & 0xFF
    if key == ord("q"):
        break

# 关闭窗口
cv2.destroyAllWindows()

# 关闭视频
stream.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

4.2 参数设置

我们可以通过不同参数的设置来实现不同的功能。

设置截屏区域

options = {"top": 40, "left": 0, "width": 100, "height": 100}
stream = ScreenGear(**options).start()
1
2

其他参数设置

# monitor表示屏幕编号，backend表示选择不同的截屏后端，colorspace表示选择颜色空间
stream = ScreenGear(monitor=1,backend="mss",colorspace="COLOR_BGR2HSV").start()
1
2

5 WriteGear

WriteGear API围绕领先的多媒体框架FFmpeg提供了一个完整、灵活和健壮的包装器。WriteGear可以将实时帧处理为具有任何合适要求如比特率、编解码器、帧率、分辨率、字幕等）的无损压缩视频文件。除此之外，WriteGear还提供了对OpenCV的VideoWriter API工具的灵活访问，用于无压缩的视频帧编码。因此WriterGear提供了两种不同的模式以供使用：

压缩模式：在这种模式下，WriteGear利用功能强大的FFmpeg内置编码器对无损多媒体文件进行编码。这种模式使我们能够轻松灵活地利用FFmpeg中几乎任何可用的参数，并且在这样做的同时，它能够稳健地安静地处理所有错误/警告。
无压缩模式：在这种模式下，WriteGear利用了基本的OpenCV内置的VideoWriter API工具。该模式还支持OpenCV的VideoWriter API中可用的所有参数转换，但它缺乏操作编码参数和其他重要功能（如视频压缩、音频编码等）的能力。

WriteGear的运行模式如下图所示：

5.1 压缩模式

WriteGear的压缩模式基于FFmpeg内置编码器对无损多媒体文件进行编码，通过execute_ffmpeg_cmd可以设置不同的FFmpeg参数。但是如果运行环境没有安装ffempeg，那么即使设置为压缩模型，也会切换为无压缩模式。关于的使用见execute_ffmpeg_cmd。

对于压缩模型，要注意不要提供任何具有不同尺寸或通道的帧给 WriteGear，此外当视频持续时间太短（<60 秒）时使用-disable_force_termination标志，否则WriteGear将不会产生任何有效输出。

5.1.1 基础使用

最基础的WriteGear调用方法，如下所示。

from vidgear.gears import CamGear
from vidgear.gears import WriteGear
import cv2

# 打开视频
stream = CamGear(source="test.mp4").start()

# 初始化视频编写器，compression_mode默认为True
writer = WriteGear(output_filename="Output.mp4",logging=True,compression_mode=True)


while True:

    # 读图
    frame = stream.read()

    if frame is None:
        break

    # 保存图片
    writer.write(frame)

    cv2.imshow("Output Frame", frame)

    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

cv2.destroyAllWindows()

stream.stop()

writer.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

5.1.2 RGB模式

如果要写入的图像通道顺序是RGB，而不是OpenCV默认的BGR格式，在写视频时可以设置为rgb_mode=True)，指定传入帧为RGB格式。

from vidgear.gears import VideoGear
from vidgear.gears import WriteGear
import cv2

stream = VideoGear(source=0).start()

writer = WriteGear(output_filename="Output.mp4")

while True:

    frame = stream.read()

    if frame is None:
        break

    # 模拟rgb图
    frame_rgb = frame[:, :, ::-1]

    # rgb图模式保存
    writer.write(frame_rgb, rgb_mode=True)  

    cv2.imshow("Output Frame", frame)

    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

cv2.destroyAllWindows()

stream.stop()

writer.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

5.1.3 指定参数

通过设置-input_framerate参数指定视频写入器的帧率，如stream.framerate表示为输入流的帧率。

# import required libraries
from vidgear.gears import CamGear
from vidgear.gears import WriteGear
import cv2

stream = CamGear(source=0).start()
# 读取视频流的帧率
print(stream.framerate)

# 设置视频写入器的帧率
output_params = {"-input_framerate": stream.framerate}

writer = WriteGear(output_filename="Output.mp4", **output_params)
1
2
3
4
5
6
7
8
9
10
11
12
13

此外可以设置参数与硬件编码器同时使用，但是这个需要比较了解FFmpeg。

# 指定编码器
output_params = {
    "-vcodec": "h264_vaapi",
    "-vaapi_device": "/dev/dri/renderD128",
    "-vf": "format=nv12,hwupload",
}

writer = WriteGear(output_filename="Output.mp4", **output_params)
1
2
3
4
5
6
7
8

如果装了声卡及其驱动，也可以通过WriteGear在保存视频的时候保存音频。

stream = VideoGear(source=0).start()

# 设置ffmpeg参数
output_params = {
    "-input_framerate": stream.framerate,
    "-thread_queue_size": "512",
    "-ac": "2",
    "-ar": "48000",
    "-f": "alsa", # 这个参数必须要放在-i前面
    "-i": "hw:1",
}
writer = WriteGear(output_filename="Output.mp4", logging=True, **output_params)
1
2
3
4
5
6
7
8
9
10
11
12

也可以设置ffmpeg参数保存视频片段。

# ffmpeg视频分割参数
output_params = {
    "-c:v": "libx264",
    "-crf": 22,
    "-map": 0,
    "-segment_time": 9,
    "-g": 9,
    "-sc_threshold": 0,
    "-force_key_frames": "expr:gte(t,n_forced*9)",
    "-clones": ["-f", "segment"],
}

# output_filename命名规则
writer = WriteGear(output_filename="output%03d.mp4", logging=True, **output_params)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

5.1.4 发送数据给流媒体服务器

在vidgear的0.2.6以上版本，可以通过WriteGear推流给流服务器如rtsp服务器。但是首先需要搭建一个rtsp服务器，可以通过rtsp-simple-server搭建一个非常简单的流媒体服务器。如何搭建rtsp-simple-server，可以看看windows环境下python使用ffmpeg rtsp推流。当然也可以通过ffmpeg直接推流，vidgear只是提供了快捷接口，如果真心想从事音视频流媒体开发，ffmpeg必学。此外rtsp-simple-server是基于go语言开发的，好处就是开发效率高，有垃圾回收。但是主流的rtsp服务器搭建，都是基于C++语言，好处在于C++相关的流媒体现成库多，而且C++调用ffmpeg非常方便。

stream = CamGear(source="test.mp4").start()

# 设置ffmpeg推流参数
output_params = {"-f": "rtsp", "-rtsp_transport": "tcp"}

# 将视频推给rtsp服务器
writer = WriteGear(
    output_filename="rtsp://localhost:8554/mystream", logging=True, **output_params
)
1
2
3
4
5
6
7
8
9

5.2 无压缩模式

无压缩模式直接调用OpenCV的写视频接口，但是缺乏控制输出质量、压缩和其他重要功能。因为没有用到ffmpeg内部能力，因此，与压缩模式相比，生成的输出视频文件大小将大很多。具体示例如下：

# 设置OpenCV中的参数
output_params = {"-fourcc": "MJPG", "-fps": 30}

# compression_mode设置为False
writer = WriteGear(output_filename="Output.mp4", compression_mode=False, logging=True, **output_params)
1
2
3
4
5

6 NetGear

NetGear用于通过网络实时发送和接受视频帧，并同时提供JPEG帧压缩功能。

6.1 基础使用

在本机环境，可以通过服务端代码传递数据，客户端收取数据。

客户端

from vidgear.gears import NetGear
import cv2

# receive_mode = True表示为接收数据
client = NetGear(receive_mode=True)


while True:

    # 从网络读图
    frame = client.recv()

    if frame is None:
        break

    cv2.imshow("Output Frame", frame)

    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

cv2.destroyAllWindows()

client.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

服务端


from vidgear.gears import VideoGear
from vidgear.gears import NetGear

stream = VideoGear(source="test.mp4").start()

# 开启Netgear服务
server = NetGear()
while True:

    try:

        frame = stream.read()

        if frame is None:
            break

        # 传递图像
        server.send(frame)

    except KeyboardInterrupt:
        break

stream.stop()

server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

6.2 参数设置

设置不同参数以应对不同环境。

客户端


from vidgear.gears import NetGear
import cv2

# 设置，这里都是消息接收的方式，直接默认就行了，不用管。
options = {"flag": 0, "copy": False, "track": False}

# address设置为客户端的ip，port设置为客户端的端口
client = NetGear(
    address="0.0.0.0",
    port="5454",
    protocol="tcp",
    pattern=1,
    receive_mode=True,
    logging=True,
    **options
)

while True:
    # 收取数据
    frame = client.recv()

    if frame is None:
        break

    cv2.imshow("Output Frame", frame)

    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

cv2.destroyAllWindows()

client.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

服务端

# import required libraries
from vidgear.gears import VideoGear
from vidgear.gears import NetGear

# 定义参数
options = {"flag": 0, "copy": False, "track": False}

stream = VideoGear(source=0).start()

# address设置为客户端的ip，port设置为客户端的端口
# pattern表示设置服务器/客户端之间支持的消息传递模式
server = NetGear(
    address="0.0.0.0",
    port="5454",
    protocol="tcp",
    pattern=1,
    logging=True,
    **options
)

while True:

    try:
        frame = stream.read()

        if frame is None:
            break

        server.send(frame)

    except KeyboardInterrupt:
        break

stream.stop()

server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

6.3 多服务端模式

NetGear可以同时通过多个服务端向一个客户端传递数据。

6.3.1 单向传输

客户端

from vidgear.gears import NetGear
# 一个整合opencv，numpy，matplotlib基本操作的库
from imutils import build_montages
import cv2

# 多服务器模式
options = {"multiserver_mode": True}

# address客户端ip
# port双端口读取
client = NetGear(
    address="0.0.0.0",
    port=(5566, 5567),
    protocol="tcp",
    pattern=1,
    receive_mode=True,
    **options
)

# 设置收取的图像为固定大小
imgSize = (576, 704)

# 保存数据
frame_dict = {}

# 这里读取不同端口图像是按照顺序依次读取
while True:

    try:
        # 每次从一个端口读取一张图
        data = client.recv()

        if data is None:
            break

        # 提取端口地址，文本和图像
        unique_address, extracted_data, frame = data

        # 统一图像大小，因为不同源的图像大小可能不一样，使得展示统一
        frame = cv2.resize(frame, imgSize)

        # 绘制文字
        cv2.putText(frame, extracted_data, (20, 50),
                    cv2.FONT_HERSHEY_SIMPLEX, 2, (0, 0, 255), 2)

        (h, w) = frame.shape[:2]

        # 保存数据用于展示
        frame_dict[unique_address] = frame

        montages = build_montages(frame_dict.values(), (w, h), (2, 1))

        for (i, montage) in enumerate(montages):

            cv2.imshow("Montage Footage {}".format(i), montage)

        key = cv2.waitKey(1) & 0xFF
        if key == ord("q"):
            break

    except KeyboardInterrupt:
        break

cv2.destroyAllWindows()

client.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

服务端1

from vidgear.gears import NetGear
from vidgear.gears import CamGear

# 读取视频
stream = CamGear(source=0).start()

# 设置多服务器模式
options = {"multiserver_mode": True}

# address设置为客户端的ip，port设置为客户端的端口
server = NetGear(
    address="0.0.0.0", port="5566", protocol="tcp", pattern=1, **options
)

while True:

    try:
        frame = stream.read()

        if frame is None:
            break

        text = "Port: 5566"

        # 发送数据
        server.send(frame, message=text)

    except KeyboardInterrupt:
        break

stream.stop()

server.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

服务端2

from vidgear.gears import NetGear
from vidgear.gears import CamGear

# 读取视频
stream = CamGear(source='test.mp4').start()

# 设置多服务器模式
options = {"multiserver_mode": True}

# address设置为客户端的ip，port设置为客户端的端口
server = NetGear(
    address="0.0.0.0", port="5567", protocol="tcp", pattern=1, **options
)

while True:

    try:
        frame = stream.read()

        if frame is None:
            break

        text = "Port: 5567"

        # 发送数据
        server.send(frame, message=text)

    except KeyboardInterrupt:
        break

stream.stop()

server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

6.3.2 双向传输

客户端

from vidgear.gears import NetGear
# 一个整合opencv，numpy，matplotlib基本操作的库
from imutils import build_montages
import cv2

# 多服务器和双向连接模式，但是延迟可能较大
options = {"multiserver_mode": True, "bidirectional_mode": True}


client = NetGear(
    address="0.0.0.0",
    port=(5566, 5567),
    protocol="tcp",
    pattern=1,
    receive_mode=True,
    logging=True,
    **options
)

# 设置收取的图像为固定大小
imgSize = (576, 704)

frame_dict = {}


while True:

    try:
        target_data = "已经收到数据"
        data = client.recv(return_data=target_data)
        # 每次从一个端口读取一张图，并返回数据
        data = client.recv(return_data=target_data)

        if data is None:
            break

        # 提取端口地址，文本和图像
        unique_address, extracted_data, frame = data

        # 统一图像大小，因为不同源的图像大小可能不一样，使得展示统一
        frame = cv2.resize(frame, imgSize)

        # 绘制文字
        cv2.putText(frame, extracted_data, (20, 50),
                    cv2.FONT_HERSHEY_SIMPLEX, 2, (0, 0, 255), 2)

        (h, w) = frame.shape[:2]

        # 保存数据用于展示
        frame_dict[unique_address] = frame

        montages = build_montages(frame_dict.values(), (w, h), (2, 1))

        for (i, montage) in enumerate(montages):

            cv2.imshow("Montage Footage {}".format(i), montage)

        key = cv2.waitKey(1) & 0xFF
        if key == ord("q"):
            break

    except KeyboardInterrupt:
        break

cv2.destroyAllWindows()

client.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

服务端1

from vidgear.gears import NetGear
from vidgear.gears import CamGear

# 读取视频
stream = CamGear(source=0).start()

# 多服务器和双向连接模式，但是延迟可能较大
options = {"multiserver_mode": True,"bidirectional_mode": True}

# address设置为客户端的ip，port设置为客户端的端口
server = NetGear(
    address="0.0.0.0", port="5566", protocol="tcp", pattern=1, **options
)

while True:

    try:
        frame = stream.read()

        if frame is None:
            break

        text = "Port: 5566"

        # 发送数据
        recv_data = server.send(frame, message=text)
        
        if recv_data is not None:
            print(recv_data)

    except KeyboardInterrupt:
        break

stream.stop()

server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

服务端2

from vidgear.gears import NetGear
from vidgear.gears import CamGear

# 读取视频
stream = CamGear(source="test.mp4").start()

# 多服务器和双向连接模式，但是延迟可能较大
options = {"multiserver_mode": True, "bidirectional_mode": True}

# address设置为客户端的ip，port设置为客户端的端口
server = NetGear(
    address="0.0.0.0", port="5567", protocol="tcp", pattern=1, **options
)

while True:

    try:
        frame = stream.read()

        if frame is None:
            break

        text = "Port: 5567"

        # 发送数据
        recv_data = server.send(frame, message=text)

        if recv_data is not None:
            print(recv_data)

    except KeyboardInterrupt:
        break

stream.stop()

server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

6.4 多客户端模式

用VidGear驱动多客户端模式有点不太稳定，具体的使用见netgear_multi_client，这里不进行介绍。

6.5 双向模式

该模式下，服务端和客户端都可以相互传递数据。

服务端

# import required libraries
from vidgear.gears import NetGear
from vidgear.gears import CamGear
stream = CamGear(source="test.mp4").start()

options = {"bidirectional_mode": True}

server = NetGear(
    address="0.0.0.0",
    port="5454",
    protocol="tcp",
    pattern=1,
    logging=True,
    **options
)

while True:

    try:
        frame = stream.read()

        if frame is None:
            break

        target_data = "Hello, I am a Server."
        # 发送数据，并接收数据
        recv_data = server.send(frame, message=target_data)

        if not (recv_data is None):
            print(recv_data)

    except KeyboardInterrupt:
        break

stream.stop()

server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

客户端

# import required libraries
from vidgear.gears import NetGear
import cv2

options = {"bidirectional_mode": True}

client = NetGear(
    address="0.0.0.0",
    port="5454",
    protocol="tcp",
    pattern=1,
    receive_mode=True,
    logging=True,
    **options
)

while True:

    target_data = "Hi, I am a Client here."

    data = client.recv(return_data=target_data)

    if data is None:
        break

    server_data, frame = data

    if frame is None:
        break

    if not (server_data is None):
        print(server_data)

    cv2.imshow("Output Frame", frame)

    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

cv2.destroyAllWindows()

client.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

6.6 其他模式设置

安全模式

# secure_mode可选值为0，1，2。数字越大越安全，但是越慢，0为默认值
options = {
    "secure_mode": 2,
} 
1
2
3
4

帧压缩

# 色彩空间转换
options = {"jpeg_compression": "GRAY"}
# 压缩jgp图像质量，jpeg_compression_quality越大图像越清晰
options = {"jpeg_compression": True, "jpeg_compression_quality": 95}
# jpeg_compression_fastdct加快解码速度，默认为True
options = {"jpeg_compression": True, "jpeg_compression_fastdct": True}
# jpeg_compression_fastupsample加快采样速度，默认为False
options = {"jpeg_compression": True, "jpeg_compression_fastupsample": True}
1
2
3
4
5
6
7
8

6.7 NetGear_Async

NetGear_Async通过异步的方式以大约三分之一的内存消耗产生与NetGear API相同的性能，并且还提供完整的服务器-客户端处理，以及使用类似于NetGear的可变协议/模式的各种选项，但缺乏灵活性，因为它仅支持部分NetGear的模式。但是要使用Async模式，比如安装vidgear[asyncio]。安装代码如下：

pip install vidgear[asyncio]

客户端

# import libraries
from vidgear.gears.asyncio import NetGear_Async
from vidgear.gears import WriteGear
import cv2
import asyncio

# receive_mode=True接收数据
client = NetGear_Async(
    address="0.0.0.0",
    port="5454",
    protocol="tcp",
    pattern=2,
    receive_mode=True,
    logging=True,
).launch()
# 写视频
writer = WriteGear(output_filename="Output.mp4", logging=True)


# 异步函数，处理数据
async def main():
    async for frame in client.recv_generator():
        writer.write(frame)

        cv2.imshow("Output Frame", frame)
        key = cv2.waitKey(1) & 0xFF

        # 等待数据
        await asyncio.sleep(0)


if __name__ == "__main__":
    asyncio.set_event_loop(client.loop)
    try:
        client.loop.run_until_complete(main())
    except (KeyboardInterrupt, SystemExit):
        pass

    cv2.destroyAllWindows()
    client.close()
    writer.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

服务端

# import libraries
from vidgear.gears.asyncio import NetGear_Async
import asyncio

# address客户端IP
server = NetGear_Async(
    source='test.mp4',
    address="0.0.0.0",
    port="5454",
    protocol="tcp",
    pattern=2,
    stabilize=True,
    logging=True,
).launch()

if __name__ == "__main__":
    asyncio.set_event_loop(server.loop)
    try:
        # 循环运行
        server.loop.run_until_complete(server.task)
    except (KeyboardInterrupt, SystemExit):
        pass
    finally:
        server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

7 StreamGear

SteamGear 是一种开箱即用的解决方案，用于对源视频/音频文件和实时视频帧进行转码，并将它们分解成一系列具有适当长度的多个片段。这些片段可以以不同的质量级别（不同的比特率或空间分辨率）流式传输视频。

StreamGear主要在以下独立的转码模式下运行：

单源模式：将整个视频文件（而不是逐帧）转码为多个较小的序列以进行流式传输。
实时帧模式：直接逐帧转码（而不是整个视频文件），转成多个小块/小段的序列用于流媒体。

StreamGear的基本流程如下图所示，基于ffmpeg切分视频和音频为若干片段，通过http协议传输给不同应用端。StreamGear的使用必须要基于ffmpeg，如果本地环境没有安装ffmpeg将会报错。

7.1 单源模式

基础使用

下面的代码就是将一个视频直接切分为若干片段。

from vidgear.gears import StreamGear

# 使用有效的视频输入激活单源模式
stream_params = {"-video_source": "test.mp4"}
# livestream为True表示启用直播
# stream_params = {"-video_source": 0, "-livestream": True}
# 设置输出模式，会在目录下生成若干片段
streamer = StreamGear(output="dash_out.mpd", **stream_params)
# 切分整个视频
streamer.transcode_source()
# 结束
streamer.terminate()
1
2
3
4
5
6
7
8
9
10
11
12

设置不同参数的流

from vidgear.gears import StreamGear

# 定义各种流
stream_params = {
    "-video_source": "test.mp4",
    "-streams": [
        {"-resolution": "1920x1080", "-video_bitrate": "4000k"}, 
        {"-resolution": "1280x720", "-framerate": 30.0}, 
        {"-resolution": "640x360", "-framerate": 60.0},  
        {"-resolution": "320x240", "-video_bitrate": "500k"}, 
    ],
}

streamer = StreamGear(output="dash_out.mpd", **stream_params)
# 设置不同参数格式
# streamer = StreamGear(output="hls_out.m3u8", format = "hls", **stream_params)
streamer.transcode_source()
streamer.terminate()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

ffmpeg参数设置

stream_params = {
    "-video_source": "test.mp4",
    "-vcodec": "libx265", 
    "-x265-params": "lossless=1", 
    "-crf": 25,
    "-bpp": "0.15",
    "-streams": [
        {"-resolution": "1280x720", "-video_bitrate": "4000k"},
        {"-resolution": "640x360", "-framerate": 60.0},  
    ],
    "-audio": "/home/foo/foo1.aac",  # 设置音频
    "-acodec": "libfdk_aac", 
    "-vbr": 4,
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

7.2 实时帧模式

实时帧模式并不是直接开启视频直播，因为不流畅。如果想开启直播模式，设置livestream为True。

基础使用

下面代码实现逐帧切分图像。


from vidgear.gears import CamGear
from vidgear.gears import StreamGear
import cv2

stream = CamGear(source='test.mp4').start() 
streamer = StreamGear(output="dash_out.mpd")

# 设置格式参数
# streamer = StreamGear(output="hls_out.m3u8", format = "hls")

# 设置为直播模式
# stream = CamGear(source=0).start()
# stream_params = {"-input_framerate": stream.framerate, "-livestream": True}

while True:

    frame = stream.read()

    if frame is None:
        break

    # 发送图像给streamer，进行转换
    streamer.stream(frame)
    # 设置传入帧的RGB通道格式
    # streamer.stream(frame_rgb, rgb_mode = True) 
    cv2.imshow("Output Frame", frame)

    key = cv2.waitKey(1) & 0xFF
    if key == ord("q"):
        break

cv2.destroyAllWindows()

stream.stop()

streamer.terminate()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

参数设置

# 设置帧率，stream.framerate表示从网络摄像头中读取帧率，如果没有设置帧率则默认为25
# stream_params = {"-input_framerate":stream.framerate}

# 设置不同流
# stream_params = {
#     "-streams": [
#         {"-resolution": "1280x720", "-framerate": 30.0}, 
#         {"-resolution": "640x360", "-framerate": 60.0}, 
#         {"-resolution": "320x240", "-video_bitrate": "500k"},
#     ],
# }
1
2
3
4
5
6
7
8
9
10
11

8 WebGear

WebGear简单来说用于将实时视频帧传输到网络中的任何网络浏览器。不过这个只是普通的demo展示，实际工程不会这样使用。此外WebGear的代码都应该在命令行下运行。

8.1 WebGear的使用

WebGear依赖于带有asyncio支持的VidGear，安装方式如下：

pip install vidgear[asyncio]

基础使用

以下代码通过uvicorn创建web服务器播放视频（以图片形式展示），打开地址http://localhost:8000/即可看到结果。uvicorn是非常轻量快速的Python异步web框架，详情使用见uvicorn。

import uvicorn
from vidgear.gears.asyncio import WebGear

# 参数设置
options = {
    "frame_size_reduction": 40, # frame尺寸减少40%
    "jpeg_compression_quality": 80, # jpeg图质量
    "jpeg_compression_fastdct": True, # 使用fastdct快速编码
    "jpeg_compression_fastupsample": False,
}

# 初始化
web = WebGear(source="test.mp4", logging=True, **options)

# 利用Uvicorn创建服务器播放视频，地址：http://localhost:8000/
uvicorn.run(web(), host="localhost", port=8000)

# 关闭
web.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

与OpenCV一同使用

WebGear自定义视频源，如设置从OpenCV获得图像。

import uvicorn, asyncio, cv2
from vidgear.gears.asyncio import WebGear
from vidgear.gears.asyncio.helper import reducer

# 初始化，自定义输入源无法使用配置参数
web = WebGear(logging=True)


async def my_frame_producer():
    
    stream = cv2.VideoCapture(0)
    while True:
        (grabbed, frame) = stream.read()
        if not grabbed:
            break

        # 图像尺寸减少比例为30%
        frame = await reducer(frame, percentage=30, interpolation=cv2.INTER_AREA)  
        # 图像编码
        encodedImage = cv2.imencode(".jpg", frame)[1].tobytes()
        # 发送图片
        yield (b"--frame\r\nContent-Type:image/jpeg\r\n\r\n" + encodedImage + b"\r\n")
        await asyncio.sleep(0)
    stream.release()


# 自定义图像生成器
web.config["generator"] = my_frame_producer

uvicorn.run(web(), host="localhost", port=8000)

web.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

添加网页路由

添加网页路由就是在指定网站中添加一个新的网页，比如hello.html中的内容如下。

<html>
   <header>
      <title>This is Hello world page</title>
   </header>
   <body>
      <h1>Hello World</h1>
      <p>你好，世界！</p>
   </body>
</html>
1
2
3
4
5
6
7
8
9

以下代码实现当输入http://localhost:8000/hello可打开hello.html。

import uvicorn, asyncio
from starlette.templating import Jinja2Templates
from starlette.routing import Route
from vidgear.gears.asyncio import WebGear

# 设置网页文件根目录
template = Jinja2Templates(directory="./myweb")

# 设置另外一个网页，hello.html位于./myweb目录下
async def hello_world(request):
    page = "hello.html"
    context = {"request": request}
    return template.TemplateResponse(page, context)


# 配置参数
options = {
    "frame_size_reduction": 40,
    "jpeg_compression_quality": 80,
    "jpeg_compression_fastdct": True,
    "jpeg_compression_fastupsample": False,
}

web = WebGear(
    source="test.mp4", logging=True, **options
) 

# 添加路由，即打开http://localhost:8000/hello访问hello.html中的内容
web.routes.append(Route("/hello", endpoint=hello_world))

uvicorn.run(web(), host="localhost", port=8000)

web.shutdown()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

将NetGear_Async与WebGear一起使用

以下代码通过NetGear_Async传输服务端的数据给客户端，然后客户端展示数据。

服务端代码如下：

from vidgear.gears.asyncio import NetGear_Async
import cv2, asyncio

# 初始化服务端
# address写客户端的ip
server = NetGear_Async(
    source=None,
    address="0.0.0.0",
    port="5454",
    protocol="tcp",
    pattern=1,
    logging=True,
)

# 自定义数据
async def my_frame_generator():

    # 打开视频
    stream = cv2.VideoCapture(0)

    # 读取视频
    while True:

        (grabbed, frame) = stream.read()

        if not grabbed:
            break

        yield frame
        await asyncio.sleep(0)

    stream.release()


if __name__ == "__main__":
    asyncio.set_event_loop(server.loop)
    # 设置数据来源
    server.config["generator"] = my_frame_generator()
    # 启动服务
    server.launch()
    try:
        server.loop.run_until_complete(server.task)
    except (KeyboardInterrupt, SystemExit):
        pass
    finally:
        server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

客户端代码如下：

from vidgear.gears.asyncio import NetGear_Async
from vidgear.gears.asyncio import WebGear
from vidgear.gears.asyncio.helper import reducer
import uvicorn, asyncio, cv2


client = NetGear_Async(
    address="0.0.0.0",
    port="5454",
    receive_mode=True,
    pattern=1,
    logging=True,
).launch()


async def my_frame_producer():

    async for frame in client.recv_generator():
        # 压缩得到的图片
        frame = await reducer(
            frame, percentage=30, interpolation=cv2.INTER_AREA
        )  

        encodedImage = cv2.imencode(".jpg", frame)[1].tobytes()
        yield (b"--frame\r\nContent-Type:image/jpeg\r\n\r\n" + encodedImage + b"\r\n")
        await asyncio.sleep(0)


if __name__ == "__main__":
    asyncio.set_event_loop(client.loop)

    web = WebGear(logging=True)

    web.config["generator"] = my_frame_producer

    # http://localhost:8000/展示数据
    uvicorn.run(web(), host="localhost", port=8000)

    client.close()

    web.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

8.2 WebGear_RTC

WebGear_RTC在许多方面与WeGear相似，但在底层使用WebRTC技术，这使其适用性更好。WebGear_RTC必须使用vidgear[asyncio]安装代码如下：

pip install vidgear[asyncio]

此外WebGear_RTC也需要aiortc，aiortc需要使用Microsoft Visual C++ 14.0。安装如下：

pip install aiortc

基础使用

打开网站后会多一个视频控制条。

import uvicorn
from vidgear.gears.asyncio import WebGear_RTC

# 设置视频参数
options = {
    "frame_size_reduction": 25,
}

web = WebGear_RTC(source="test.mp4", logging=True, **options)
# 打开http://localhost:8000访问内容
uvicorn.run(web(), host="localhost", port=8000)

web.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13

启用实时播放

import uvicorn
from vidgear.gears.asyncio import WebGear_RTC

# enable_live_broadcast设置直播流
options = {
    "frame_size_reduction": 30,
    "enable_live_broadcast": True,
}


web = WebGear_RTC(source="test.mp4", logging=True, **options)

# 打开http://localhost:8000/播放视频
uvicorn.run(web(), host="0.0.0.0", port=8000)


web.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

9 help方法

下面代码展示了vidgear常用工具方法的使用。

# asyncio版本
# from vidgear.gears.asyncio.helper import *

from vidgear.gears.helper import *
import cv2
# 打印opencv大版本号
print(check_CV_version())
# 检查OpenCV是否使用Gstreamer
print(check_gstreamer_support())
# 安全创建文件夹
mkdir_safe(dir_path="build")
# 安全删除文件夹build中的demo.py文件
delete_ext_safe(dir_path="build", extensions="demo.py")

# 将opencv图像的宽高缩小百分之percentage
img = cv2.imread("test.jpg")
img_ = reducer(img, percentage=30)
print('w ratio is:{}, h ratio is:{}'.format(img_.shape[1]/img.shape[1],
      img_.shape[0]/img.shape[0]))

# 创建一个和img一样尺寸的纯黑图像（所有像素值为0）
img_ = create_blank_frame(img)
# 创建一个和img一样尺寸的纯黑图像，然后在图中绘制文字text
img_ = create_blank_frame(img, text="hello")

# 把dict类型转换为arg列表
param = {"first": 1, "second": 2}
param_arg = dict2Args(param)
print(param_arg)

# 判断指定路径是否有读写权限,is_windows表示当前为windows系统
print(check_WriteAccess("./", is_windows=True))

# 查看指定address的port是否被占用
print(check_open_port('127.0.0.1', 135))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

10 参考

VidGear基础

CamGear

VideoGear

基于特征点匹配的视频稳像

WriteGear

NetGear

netgear_multi_client

WebGear

uvicorn

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/641061