首先给出展示结果，大体就是检测工业板子是否出现。采取检测的方法比较简单，用的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 流媒体服务器

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://mand, 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()

总结

以上所述是小编给大家介绍的Python实现直播推流效果，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对网站的支持！
如果你觉得本文对你有帮助，欢迎转载，烦请注明出处，谢谢！