在设置视频源的时候指定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()
通过设置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()
在获取图像时可以直接进行图像色彩空间转换，实际就是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()
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()
在这篇文章中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()
下图展示了VideoGear的功能，即读取视频，然后进行视频稳流。想要知道视频稳流具体如何实现可以阅读：基于特征点匹配的视频稳像
VideoGear内部封装了CamGear和视频稳像的代码，其函数调用和CamGear差不多，可以设置的参数也是一样的。
# 调用库
from vidgear.gears import VideoGear
import cv2
# VideoGear和CamGear调用方法一样，参数设置也是一样
stream = VideoGear(source=0, colorspace="COLOR_BGR2HSV", logging=True).start()
此外VideoGear还支持和ROS机器人操作系统一起使用，可以处理rtsp等视频流。但是VideoGear速度很慢，如果没有视频稳流的需求，可以仅使用CamGear。
可以不使用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()
ScreenGear专为超快屏幕播放而设计，这意味着它在尽可能缩小时间延迟的情况下，通过在计算机屏幕上定义一个区域或全屏来实时抓取监视器中的图像。ScreenGear主要是基于pyscreenshot和python-mss实现的。
以下代码为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()
我们可以通过不同参数的设置来实现不同的功能。
设置截屏区域
options = {"top": 40, "left": 0, "width": 100, "height": 100}
stream = ScreenGear(**options).start()
其他参数设置
# monitor表示屏幕编号，backend表示选择不同的截屏后端，colorspace表示选择颜色空间
stream = ScreenGear(monitor=1,backend="mss",colorspace="COLOR_BGR2HSV").start()
WriteGear API围绕领先的多媒体框架FFmpeg提供了一个完整、灵活和健壮的包装器。WriteGear可以将实时帧处理为具有任何合适要求如比特率、编解码器、帧率、分辨率、字幕等）的无损压缩视频文件。除此之外，WriteGear还提供了对OpenCV的VideoWriter API工具的灵活访问，用于无压缩的视频帧编码。因此WriterGear提供了两种不同的模式以供使用：
-
压缩模式：在这种模式下，WriteGear利用功能强大的FFmpeg内置编码器对无损多媒体文件进行编码。这种模式使我们能够轻松灵活地利用FFmpeg中几乎任何可用的参数，并且在这样做的同时，它能够稳健地安静地处理所有错误/警告。
-
无压缩模式：在这种模式下，WriteGear利用了基本的OpenCV内置的VideoWriter API工具。该模式还支持OpenCV的VideoWriter API中可用的所有参数转换，但它缺乏操作编码参数和其他重要功能（如视频压缩、音频编码等）的能力。
WriteGear的运行模式如下图所示：
WriteGear的压缩模式基于FFmpeg内置编码器对无损多媒体文件进行编码，通过execute_ffmpeg_cmd可以设置不同的FFmpeg参数。但是如果运行环境没有安装ffempeg，那么即使设置为压缩模型，也会切换为无压缩模式。关于的使用见execute_ffmpeg_cmd。
对于压缩模型，要注意不要提供任何具有不同尺寸或通道的帧给 WriteGear，此外当视频持续时间太短（<60 秒）时使用-disable_force_termination标志，否则WriteGear将不会产生任何有效输出。
最基础的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()
如果要写入的图像通道顺序是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()
通过设置-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)
此外可以设置参数与硬件编码器同时使用，但是这个需要比较了解FFmpeg。
# 指定编码器
output_params = {
"-vcodec": "h264_vaapi",
"-vaapi_device": "/dev/dri/renderD128",
"-vf": "format=nv12,hwupload",
writer = WriteGear(output_filename="Output.mp4", **output_params)
如果装了声卡及其驱动，也可以通过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)
也可以设置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)
在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
无压缩模式直接调用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)
NetGear用于通过网络实时发送和接受视频帧，并同时提供JPEG帧压缩功能。
在本机环境，可以通过服务端代码传递数据，客户端收取数据。
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()
from vidgear.gears import VideoGear
from vidgear.gears import NetGear
stream = VideoGear(source="test.mp4").start()
# 开启Netgear服务
server = NetGear()
while True:
frame = stream.read()
if frame is None:
break
# 传递图像
server.send(frame)
except KeyboardInterrupt:
break
stream.stop()
server.close()
设置不同参数以应对不同环境。
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()
# 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:
frame = stream.read()
if frame is None:
break
server.send(frame)
except KeyboardInterrupt:
break
stream.stop()
server.close()
NetGear可以同时通过多个服务端向一个客户端传递数据。
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:
# 每次从一个端口读取一张图
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()
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:
frame = stream.read()
if frame is None:
break
text = "Port: 5566"
# 发送数据
server.send(frame, message=text)
except KeyboardInterrupt:
break
stream.stop()
server.close()
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:
frame = stream.read()
if frame is None:
break
text = "Port: 5567"
# 发送数据
server.send(frame, message=text)
except KeyboardInterrupt:
break
stream.stop()
server.close()
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:
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()
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:
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()
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:
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()
用VidGear驱动多客户端模式有点不太稳定，具体的使用见netgear_multi_client，这里不进行介绍。
该模式下，服务端和客户端都可以相互传递数据。
# 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:
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()
# 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()
# secure_mode可选值为0，1，2。数字越大越安全，但是越慢，0为默认值
options = {
"secure_mode": 2,
# 色彩空间转换
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}
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)
client.loop.run_until_complete(main())
except (KeyboardInterrupt, SystemExit):
cv2.destroyAllWindows()
client.close()
writer.close()
# 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)
# 循环运行
server.loop.run_until_complete(server.task)
except (KeyboardInterrupt, SystemExit):
finally:
server.close()
SteamGear 是一种开箱即用的解决方案，用于对源视频/音频文件和实时视频帧进行转码，并将它们分解成一系列具有适当长度的多个片段。这些片段可以以不同的质量级别（不同的比特率或空间分辨率）流式传输视频。
StreamGear主要在以下独立的转码模式下运行：
- 单源模式：将整个视频文件 （而不是逐帧）转码为多个较小的序列以进行流式传输。
- 实时帧模式：直接逐帧转码（而不是整个视频文件），转成多个小块/小段的序列用于流媒体。
StreamGear的基本流程如下图所示，基于ffmpeg切分视频和音频为若干片段，通过http协议传输给不同应用端。StreamGear的使用必须要基于ffmpeg，如果本地环境没有安装ffmpeg将会报错。
下面的代码就是将一个视频直接切分为若干片段。
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()
设置不同参数的流
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()
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,
实时帧模式并不是直接开启视频直播，因为不流畅。如果想开启直播模式，设置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()
# 设置帧率，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"},
# ],
WebGear简单来说用于将实时视频帧传输到网络中的任何网络浏览器。不过这个只是普通的demo展示，实际工程不会这样使用。此外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()
与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()
添加网页路由
添加网页路由就是在指定网站中添加一个新的网页，比如hello.html中的内容如下。
<header>
<title>This is Hello world page</title>
</header>
<h1>Hello World</h1>
<p>你好，世界！</p>
</body>
</html>
以下代码实现当输入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()
将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()
server.loop.run_until_complete(server.task)
except (KeyboardInterrupt, SystemExit):
finally:
server.close()
客户端代码如下：
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()
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()
启用实时播放
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()
下面代码展示了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))
本文来自博客园，作者：落痕的寒假，转载请注明原文链接：https://www.cnblogs.com/luohenyueji/p/16970164.html
