Python如何实现直播推流效果

这篇文章将为大家详细讲解有关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.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如何实现直播推流效果”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，使各位可以学到更多知识，如果觉得文章不错，请把它分享出去让更多的人看到。