Python实现直播推流效果
首先给出展示结果，大体就是检测工业板子是否出现。采取检测的方法比较简单，用的OpenCV的模板检测。
大体思路
- opencv读取视频
- 将视频分割为帧
- 对每一帧进行处理(opencv模板匹配)
- 在将此帧写入pipe管道
- 利用ffmpeg进行推流直播
中间遇到的问题
在处理本地视频时，并没有延时卡顿的情况。但对实时视频流的时候，出现了卡顿延时的效果。在一顿度娘操作之后，采取了多线程的方法。
opencv读取视频
defrun_opencv_camera():
video_stream_path=0
#当video_stream_path=0会开启计算机默认摄像头也可以为本地视频文件的路径
cap=cv2.VideoCapture(video_stream_path)
whilecap.isOpened():
is_opened,frame=cap.read()
cv2.imshow('frame',frame)
cv2.waitKey(1)
cap.release()
OpenCV模板匹配
模板匹配就是在一幅图像中寻找一个特定目标的方法之一，这种方法的原理非常简单，遍历图像中每一个可能的位置，比较各处与模板是否相似，当相似度足够高时，就认为找到了目标。
deftemplate_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) iflen(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) returnimg_rgb
FFmpeg推流
在Ubuntu14上安装Nginx-RTMP流媒体服务器
https://www.nhooo.com/article/175121.htm
importsubprocessassp
rtmpUrl=""
camera_path=""
cap=cv.VideoCapture(camera_path)
#Getvideoinformation
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))
#ffmpegcommand
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)
#readwebcamera
while(cap.isOpened()):
ret,frame=cap.read()
ifnotret:
print("Openingcameraisfailed")
break
#processframe
#yourcode
#processframe
#writetopipe
p.stdin.write(frame.tostring())
说明:rtmp是要接受视频的服务器，服务器按照上面所给连接地址即可。
多线程处理
pythonmutilprocessing多进程编程https://www.nhooo.com/article/134726.htm
defimage_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)
ifcap.isOpened():
print('success')
else:
print('faild')
whileTrue:
q.put(cap.read()[1])
q.get()ifq.qsize()>1elsetime.sleep(0.01)
defimage_get(q):
whileTrue:
#start=time.time()
#flag+=1
frame=q.get()
frame=template_match(frame)
#end=time.time()
#print("thetimeis",end-start)
cv2.imshow("frame",frame)
cv2.waitKey(0)
#pipe.stdin.write(frame.tostring())
#cv2.imwrite(save_path+"%d.jpg"%flag,frame)
#多线程执行一个摄像头
defrun_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()forprocessinprocesses]
[process.join()forprocessinprocesses]
defrun():
run_single_camera()#quick,with2threads
pass
说明:使用Python3自带的多线程模块mutilprocessing模块，创建一个队列，线程A从通过rstp协议从视频流中读取出每一帧，并放入队列中，线程B从队列中将图片取出，处理后进行显示。线程A如果发现队列里有两张图片，即线程B的读取速度跟不上线程A，那么线程A主动将队列里面的旧图片删掉，换新图片。
全部代码展示
importtime
importmultiprocessingasmp
importnumpyasnp
importrandom
importsubprocessassp
importcv2
importos
#定义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="ProductYield:100%"
result="Result:perfect"
#读取模板并获取模板的高度和宽度
template=cv2.imread(template_path,0)
h,w=template.shape[:2]
#定义模板匹配函数
deftemplate_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)
iflen(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)
returnimg_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())
defimage_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)
ifcap.isOpened():
print('success')
else:
print('faild')
whileTrue:
q.put(cap.read()[1])
q.get()ifq.qsize()>1elsetime.sleep(0.01)
#采取本地视频的方式保存图片
save_path="./res_imgs"
ifos.path.exists(save_path):
os.makedir(save_path)
defimage_get(q):
whileTrue:
#start=time.time()
#flag+=1
frame=q.get()
frame=template_match(frame)
#end=time.time()
#print("thetimeis",end-start)
cv2.imshow("frame",frame)
cv2.waitKey(0)
#pipe.stdin.write(frame.tostring())
#cv2.imwrite(save_path+"%d.jpg"%flag,frame)
#多线程执行一个摄像头
defrun_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()forprocessinprocesses]
[process.join()forprocessinprocesses]
defrun():
run_single_camera()#quick,with2threads
pass
if__name__=='__main__':
run()
总结
以上所述是小编给大家介绍的Python实现直播推流效果，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对毛票票网站的支持！
如果你觉得本文对你有帮助，欢迎转载，烦请注明出处，谢谢！
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。