from threading import Thread import threading import sys import cv2 import dlib # 实测 多线程好那么一丢丢 def _help(): print("Usage:") print(" python video_face_detect_dlib.py") print(" python video_face_detect_dlib.py <path of a video>") print("For example:") print(" python video_face_detect_dlib.py video/lee.mp4") print("If the path of a video is not provided, the camera will be used as the input.Press q to quit.") def _face_detect(color_image, detector): gray_image = cv2.cvtColor(color_image, cv2.COLOR_BGR2GRAY) # img_row, img_col = gray_image.shape[:2] # scale = 0.3 # gray_image = cv2.resize(gray_image, (int(scale * img_col), int(scale * img_row))) faces = detector(gray_image, 1) for face in faces: left = face.left() top = face.top() right = face.right() bottom = face.bottom() cv2.rectangle(color_image, (left, top), (right, bottom), (0, 255, 0), 2) cv2.namedWindow("Image", cv2.WINDOW_NORMAL) #cv2.imshow("Image", color_image) cv2.imshow("Image", color_image) frame = None lock = threading.RLock() def get_frame(video_path): #video = "http://admin:admin@192.168.1.139:8081/" # 此处@后的ipv4 地址需要改为app提供的地址 cap = cv2.VideoCapture(video_path) global frame while True: _, img = cap.read() if img is None: break else: lock.acquire() frame = img lock.release() cap.release() def face_detect(video_path="http://admin:admin@192.168.1.139:8081/"): detector = dlib.get_frontal_face_detector() # 开启后台线程不断获取视频输入流,并将视频帧保存到全局变量frame中 t = Thread(target=get_frame, name="get_video_stream", args=(video_path,)) t.daemon = True t.start() while True: if frame is not None: # 对最近保存在frame中的视频帧进行人脸检测 lock.acquire() frame_copy = frame.copy() lock.release() _face_detect(frame_copy, detector) if cv2.waitKey(1) & 0xFF == ord('q'): break cv2.destroyAllWindows() if len(sys.argv) > 2 or "-h" in sys.argv or "--help" in sys.argv: _help() elif len(sys.argv) == 2: face_detect(sys.argv[1]) else: face_detect()