Python实现直播推流效果
作者:mind_programmonkey 发布时间:2023-10-06 12:43:29
首先给出展示结果,大体就是检测工业板子是否出现。采取检测的方法比较简单,用的OpenCV的模板检测。
大体思路
opencv读取视频
将视频分割为帧
对每一帧进行处理(opencv模板匹配)
在将此帧写入pipe管道
利用ffmpeg进行推流直播
中间遇到的问题
在处理本地视频时,并没有延时卡顿的情况。但对实时视频流的时候,出现了卡顿延时的效果。在一顿度娘操作之后,采取了多线程的方法。
opencv读取视频
def run_opencv_camera():
video_stream_path = 0
# 当video_stream_path = 0 会开启计算机 默认摄像头 也可以为本地视频文件的路径
cap = cv2.VideoCapture(video_stream_path)
while cap.isOpened():
is_opened, frame = cap.read()
cv2.imshow('frame', frame)
cv2.waitKey(1)
cap.release()
OpenCV模板匹配
模板匹配就是在一幅图像中寻找一个特定目标的方法之一,这种方法的原理非常简单,遍历图像中每一个可能的位置,比较各处与模板是否相似,当相似度足够高时,就认为找到了目标。
def template_match(img_rgb):
# 灰度转换
img_gray = cv2.cvtColor(img_rgb, cv2.COLOR_BGR2GRAY)
# 模板匹配
res = cv2.matchTemplate(img_gray, template, cv2.TM_CCOEFF_NORMED)
# 设置阈值
threshold = 0.8
loc = np.where(res >= threshold)
if len(loc[0]):
# 这里直接固定区域
cv2.rectangle(img_rgb, (155, 515), (1810, 820), (0, 0, 255), 3)
cv2.putText(img_rgb, category, (240, 600), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.putText(img_rgb, Confidence, (240, 640), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.putText(img_rgb, Precision, (240, 680), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.putText(img_rgb, product_yield, (240, 720), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.putText(img_rgb, result, (240, 780), cv2.FONT_HERSHEY_SIMPLEX, 2, (0, 255, 0), 5)
return img_rgb
FFmpeg推流
在Ubuntu 14 上安装 Nginx-RTMP 流媒体服务器
https://www.jb51.net/article/175121.htm
import subprocess as sp
rtmpUrl = ""
camera_path = ""
cap = cv.VideoCapture(camera_path)
# Get video information
fps = int(cap.get(cv.CAP_PROP_FPS))
width = int(cap.get(cv.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv.CAP_PROP_FRAME_HEIGHT))
# ffmpeg command
command = ['ffmpeg',
'-y',
'-f', 'rawvideo',
'-vcodec','rawvideo',
'-pix_fmt', 'bgr24',
'-s', "{}x{}".format(width, height),
'-r', str(fps),
'-i', '-',
'-c:v', 'libx264',
'-pix_fmt', 'yuv420p',
'-preset', 'ultrafast',
'-f', 'flv',
rtmpUrl]
# 管道配置
p = sp.Popen(command, stdin=sp.PIPE)
# read webcamera
while(cap.isOpened()):
ret, frame = cap.read()
if not ret:
print("Opening camera is failed")
break
# process frame
# your code
# process frame
# write to pipe
p.stdin.write(frame.tostring())
说明:rtmp是要接受视频的服务器,服务器按照上面所给连接地址即可。
多线程处理
python mutilprocessing多进程编程 https://www.jb51.net/article/134726.htm
def image_put(q):
# 采取本地视频验证
cap = cv2.VideoCapture("./new.mp4")
# 采取视频流的方式
# cap = cv2.VideoCapture(0)
# cap.set(cv2.CAP_PROP_FRAME_WIDTH,1920)
# cap.set(cv2.CAP_PROP_FRAME_HEIGHT,1080)
if cap.isOpened():
print('success')
else:
print('faild')
while True:
q.put(cap.read()[1])
q.get() if q.qsize() > 1 else time.sleep(0.01)
def image_get(q):
while True:
# start = time.time()
#flag += 1
frame = q.get()
frame = template_match(frame)
# end = time.time()
# print("the time is", end-start)
cv2.imshow("frame", frame)
cv2.waitKey(0)
# pipe.stdin.write(frame.tostring())
#cv2.imwrite(save_path + "%d.jpg"%flag,frame)
# 多线程执行一个摄像头
def run_single_camera():
# 初始化
mp.set_start_method(method='spawn') # init
# 队列
queue = mp.Queue(maxsize=2)
processes = [mp.Process(target=image_put, args=(queue, )),
mp.Process(target=image_get, args=(queue, ))]
[process.start() for process in processes]
[process.join() for process in processes]
def run():
run_single_camera() # quick, with 2 threads
pass
说明:使用Python3自带的多线程模块mutilprocessing模块,创建一个队列,线程A从通过rstp协议从视频流中读取出每一帧,并放入队列中,线程B从队列中将图片取出,处理后进行显示。线程A如果发现队列里有两张图片,即线程B的读取速度跟不上线程A,那么线程A主动将队列里面的旧图片删掉,换新图片。
全部代码展示
import time
import multiprocessing as mp
import numpy as np
import random
import subprocess as sp
import cv2
import os
# 定义opencv所需的模板
template_path = "./high_img_template.jpg"
# 定义矩形框所要展示的变量
category = "Category: board"
var_confidence = (np.random.randint(86, 98)) / 100
Confidence = "Confidence: " + str(var_confidence)
var_precision = round(random.uniform(98, 99), 2)
Precision = "Precision: " + str(var_precision) + "%"
product_yield = "Product Yield: 100%"
result = "Result: perfect"
# 读取模板并获取模板的高度和宽度
template = cv2.imread(template_path, 0)
h, w = template.shape[:2]
# 定义模板匹配函数
def template_match(img_rgb):
# 灰度转换
img_gray = cv2.cvtColor(img_rgb, cv2.COLOR_BGR2GRAY)
# 模板匹配
res = cv2.matchTemplate(img_gray, template, cv2.TM_CCOEFF_NORMED)
# 设置阈值
threshold = 0.8
loc = np.where(res >= threshold)
if len(loc[0]):
# 这里直接固定区域
cv2.rectangle(img_rgb, (155, 515), (1810, 820), (0, 0, 255), 3)
cv2.putText(img_rgb, category, (240, 600), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.putText(img_rgb, Confidence, (240, 640), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.putText(img_rgb, Precision, (240, 680), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.putText(img_rgb, product_yield, (240, 720), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.putText(img_rgb, result, (240, 780), cv2.FONT_HERSHEY_SIMPLEX, 2, (0, 255, 0), 5)
return img_rgb
# 视频属性
size = (1920, 1080)
sizeStr = str(size[0]) + 'x' + str(size[1])
# fps = cap.get(cv2.CAP_PROP_FPS) # 30p/self
# fps = int(fps)
fps = 11
hz = int(1000.0 / fps)
print ('size:'+ sizeStr + ' fps:' + str(fps) + ' hz:' + str(hz))
rtmpUrl = 'rtmp://localhost/hls/test'
# 直播管道输出
# ffmpeg推送rtmp 重点 : 通过管道 共享数据的方式
command = ['ffmpeg',
'-y',
'-f', 'rawvideo',
'-vcodec','rawvideo',
'-pix_fmt', 'bgr24',
'-s', sizeStr,
'-r', str(fps),
'-i', '-',
'-c:v', 'libx264',
'-pix_fmt', 'yuv420p',
'-preset', 'ultrafast',
'-f', 'flv',
rtmpUrl]
#管道特性配置
# pipe = sp.Popen(command, stdout = sp.PIPE, bufsize=10**8)
pipe = sp.Popen(command, stdin=sp.PIPE) #,shell=False
# pipe.stdin.write(frame.tostring())
def image_put(q):
# 采取本地视频验证
cap = cv2.VideoCapture("./new.mp4")
# 采取视频流的方式
# cap = cv2.VideoCapture(0)
# cap.set(cv2.CAP_PROP_FRAME_WIDTH,1920)
# cap.set(cv2.CAP_PROP_FRAME_HEIGHT,1080)
if cap.isOpened():
print('success')
else:
print('faild')
while True:
q.put(cap.read()[1])
q.get() if q.qsize() > 1 else time.sleep(0.01)
# 采取本地视频的方式保存图片
save_path = "./res_imgs"
if os.path.exists(save_path):
os.makedir(save_path)
def image_get(q):
while True:
# start = time.time()
#flag += 1
frame = q.get()
frame = template_match(frame)
# end = time.time()
# print("the time is", end-start)
cv2.imshow("frame", frame)
cv2.waitKey(0)
# pipe.stdin.write(frame.tostring())
#cv2.imwrite(save_path + "%d.jpg"%flag,frame)
# 多线程执行一个摄像头
def run_single_camera():
# 初始化
mp.set_start_method(method='spawn') # init
# 队列
queue = mp.Queue(maxsize=2)
processes = [mp.Process(target=image_put, args=(queue, )),
mp.Process(target=image_get, args=(queue, ))]
[process.start() for process in processes]
[process.join() for process in processes]
def run():
run_single_camera() # quick, with 2 threads
pass
if __name__ == '__main__':
run()
总结
以上所述是小编给大家介绍的Python实现直播推流效果网站的支持!
如果你觉得本文对你有帮助,欢迎转载,烦请注明出处,谢谢!
来源:https://blog.csdn.net/Mind_programmonkey/article/details/102732555
猜你喜欢
- 本文实例讲述了Python实现阿拉伯数字和罗马数字的互相转换功能。分享给大家供大家参考,具体如下:前面一篇介绍了《Java实现的求解经典罗马
- web 调试工具介绍和开发环境搭建python与selenium开发环境搭建:一、下载python软件:https://www.python
- 大量的多行段落本身就会降低可读性,同时空行分段也比空格分段有更高的可适应性...前文讨论的热烈程度远超我预期,正好还有之前查阅资料拍的几张实
- 去除数字,特殊字符,只保留汉字import res = '1123*#$ 中abc国'str = re.sub('[
- (1)标准类型操作符(所有的集合类型)成员关系 (in, not in) &nbs
- 在这篇asp之数学函数里,我们将会以表格的形式,让大家了解到关于ASP中能用到的数学函数,里面包括一个数的绝对值、一个数的平方根
- 实现效果将位于/img目录下的1000张.png图片,转换成.webp格式,并存放于img_webp文件夹内。源图片目录目标图片目录关于批量
- #!/usr/bin/env pythonimport sysfrom PyQt4 import QtGui,QtCoreimport ht
- 简单实现了一个在函数执行出现异常时自动重试的装饰器,支持控制最多重试次数,每次重试间隔,每次重试间隔时间递增。最新的代码可以访问从githu
- 写在前面的话:此篇还是asp相关的,相信玩ASP的都有这个感觉,当数据有5万多条时-------just like音乐网,要调用最新的10条
- 我就废话不多说了,大家还是直接看代码吧~import numpy as np#从scipy库中导入插值需要的方法 interpolatefr
- matplotlib简介matplotlib 是python最著名的绘图库,它提供了一整套和matlab相似的命令API,十分适合交互式地行
- 方法1:使用dataframe.loc[]函数通过这个方法,我们可以用一个条件或一个布尔数组来访问一组行或列。如果我们可以访问它,我们也可以
- 在命令行中使用 Python 时,它可以接收大约 20 个选项(option),语法格式如下:python [-bBdEhiIOqsSuvV
- 本文实例讲述了python开发之thread实现布朗运动的方法。分享给大家供大家参考,具体如下:这里我将给大家介绍有关python中thre
- 昨天有位同事说,他的网页查询过程中发现普通索引和唯一索引的效率是有差别的,普通索引比唯一索引快,今天在我的虚拟机中布置了环境,测试抓图如下:
- 利用python pyheatmap包绘制热力图,供大家参考,具体内容如下import matplotlib.pyplot as pltfr
- 存储过程的优缺点: 存储过程优点: 1.由于应用程序随着时间推移会不断更改,增删功能,T-SQL过程代码会变得更复杂,StoredProce
- 程序流程分析图:传播过程:代码展示:创建环境使用<pip install+包名>来下载torch,torchvision包准备数
- 如果是django2.0 必须下载xadmin2.0 不然很多地方不兼容xadmin2.0下载地址https://github.com/ss