Python实现推流直播
发布日期:2021-06-29 15:32:07
浏览次数:4
分类:技术文章
本文共 7746 字,大约阅读时间需要 25 分钟。
Python实现推流直播
首先给出展示结果,大体就是检测工业板子是否出现。采取检测的方法比较简单,用的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.cnblogs.com/cocoajin/p/4353767.html
import subprocess as sprtmpUrl = ""camera_path = ""cap = cv.VideoCapture(camera_path)# Get video informationfps = 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 commandcommand = ['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 webcamerawhile(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://blog.csdn.net/jeffery0207/article/details/82958520
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 timeimport multiprocessing as mpimport numpy as npimport randomimport subprocess as spimport cv2import os# 定义opencv所需的模板template_path = "./high_img_template.jpg"# 定义矩形框所要展示的变量category = "Category: board"var_confidence = (np.random.randint(86, 98)) / 100Confidence = "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 = 11hz = 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 passif __name__ == '__main__': run()
参考文章
-
- 在Ubuntu 14 上安装 Nginx-RTMP 流媒体服务器:https://www.cnblogs.com/cocoajin/p/4353767.html
-
- python mutilprocessing多进程编程:https://blog.csdn.net/jeffery0207/article/details/82958520
-
- ffmpeg 将视频和图片互转化:https://blog.csdn.net/TingiBanDeQu/article/details/53896944
-
- 基于python2.7的opencv3.3-ffmpeg-rtmp视频处理并推送流直播:https://blog.csdn.net/u014303844/article/details/80394101
-
- 读取多个(海康\大华)网络摄像头的视频流 (使用opencv-python),解决实时读取延迟问题:https://zhuanlan.zhihu.com/p/38136322
-
- python利用ffmpeg进行rtmp推流直播:https://zhuanlan.zhihu.com/p/74260950
转载地址:https://codingchaozhang.blog.csdn.net/article/details/102732555 如侵犯您的版权,请留言回复原文章的地址,我们会给您删除此文章,给您带来不便请您谅解!
发表评论
最新留言
留言是一种美德,欢迎回访!
[***.207.175.100]2024年04月24日 20时47分26秒
关于作者
喝酒易醉,品茶养心,人生如梦,品茶悟道,何以解忧?唯有杜康!
-- 愿君每日到此一游!
推荐文章
数据可视化中的格式塔心理学
2019-04-29
电动汽车的“专属危险”:网络威胁问题不容小觑
2019-04-29
短暂的告别,马上再回来
2019-04-29
统治50年:为什么SQL在如今仍然很重要?
2019-04-29
测试是一场竞争,而数据每次都会获得胜利
2019-04-29
读心的测谎系统:究竟是骗子还是个天才?
2019-04-29
最大规模技术重建:数据库连接从15000个到100个以下
2019-04-29
复工之后:员工如何改善网络安全?
2019-04-29
70%求职者因此被拒,你还不避开这些“雷区”?!
2019-04-29
办法不在多,有用就行!用Dropout解决过度拟合问题
2019-04-29
色情演员识别?绝对是人脸识别最糟糕的应用……
2019-04-29
让强化学习逃离“乏味区域陷阱”,试着加点噪音吧!
2019-04-29
超详细Spring Boot面试问题集锦,死角一个不留!
2019-04-29
10个业余时间可完成的项目,助你飞速提升编码能力!
2019-04-29
网络爬虫初涉——用python爬取网络小说
2019-04-29
Pycharm+tensorflow dropout 学习(三)
2019-04-29
Pycharm+tensorflow CNN 学习(四)
2019-04-29
用python暴力破解压缩包密码
2019-04-29
基于OpenCV 将图片进行预处理,转变为MNIST图片格式
2019-04-29
PyCharm+Tensorflow CNN调用训练好的模型进行预测 (五)
2019-04-29