import argparse import time import numpy as np import cv2 as cv # ------------------------Service operations------------------------ def weight_path(model_path): """ Get path of weights based on path to IR Params: model_path: the string contains path to IR file Return: Path to weights file """ assert model_path.endswith('.xml'), "Wrong topology path was provided" return model_path[:-3] + 'bin' def build_argparser(): """ Parse arguments from command line Return: Pack of arguments from command line """ parser = argparse.ArgumentParser(description='This is an OpenCV-based version of Gaze Estimation example') parser.add_argument('--input', help='Path to the input video file') parser.add_argument('--out', help='Path to the output video file') parser.add_argument('--facem', default='face-detection-retail-0005.xml', help='Path to OpenVINO face detection model (.xml)') parser.add_argument('--faced', default='CPU', help='Target device for the face detection' + '(e.g. CPU, GPU, VPU, ...)') parser.add_argument('--headm', default='head-pose-estimation-adas-0001.xml', help='Path to OpenVINO head pose estimation model (.xml)') parser.add_argument('--headd', default='CPU', help='Target device for the head pose estimation inference ' + '(e.g. CPU, GPU, VPU, ...)') parser.add_argument('--landm', default='facial-landmarks-35-adas-0002.xml', help='Path to OpenVINO landmarks detector model (.xml)') parser.add_argument('--landd', default='CPU', help='Target device for the landmarks detector (e.g. CPU, GPU, VPU, ...)') parser.add_argument('--gazem', default='gaze-estimation-adas-0002.xml', help='Path to OpenVINO gaze vector estimaiton model (.xml)') parser.add_argument('--gazed', default='CPU', help='Target device for the gaze vector estimation inference ' + '(e.g. CPU, GPU, VPU, ...)') parser.add_argument('--eyem', default='open-closed-eye-0001.xml', help='Path to OpenVINO open closed eye model (.xml)') parser.add_argument('--eyed', default='CPU', help='Target device for the eyes state inference (e.g. CPU, GPU, VPU, ...)') return parser # ------------------------Support functions for custom kernels------------------------ def intersection(surface, rect): """ Remove zone of out of bound from ROI Params: surface: image bounds is rect representation (top left coordinates and width and height) rect: region of interest is also has rect representation Return: Modified ROI with correct bounds """ l_x = max(surface[0], rect[0]) l_y = max(surface[1], rect[1]) width = min(surface[0] + surface[2], rect[0] + rect[2]) - l_x height = min(surface[1] + surface[3], rect[1] + rect[3]) - l_y if width < 0 or height < 0: return (0, 0, 0, 0) return (l_x, l_y, width, height) def process_landmarks(r_x, r_y, r_w, r_h, landmarks): """ Create points from result of inference of facial-landmarks network and size of input image Params: r_x: x coordinate of top left corner of input image r_y: y coordinate of top left corner of input image r_w: width of input image r_h: height of input image landmarks: result of inference of facial-landmarks network Return: Array of landmarks points for one face """ lmrks = landmarks[0] raw_x = lmrks[::2] * r_w + r_x raw_y = lmrks[1::2] * r_h + r_y return np.array([[int(x), int(y)] for x, y in zip(raw_x, raw_y)]) def eye_box(p_1, p_2, scale=1.8): """ Get bounding box of eye Params: p_1: point of left edge of eye p_2: point of right edge of eye scale: change size of box with this value Return: Bounding box of eye and its midpoint """ size = np.linalg.norm(p_1 - p_2) midpoint = (p_1 + p_2) / 2 width = scale * size height = width p_x = midpoint[0] - (width / 2) p_y = midpoint[1] - (height / 2) return (int(p_x), int(p_y), int(width), int(height)), list(map(int, midpoint)) # ------------------------Custom graph operations------------------------ @cv.gapi.op('custom.GProcessPoses', in_types=[cv.GArray.GMat, cv.GArray.GMat, cv.GArray.GMat], out_types=[cv.GArray.GMat]) class GProcessPoses: @staticmethod def outMeta(arr_desc0, arr_desc1, arr_desc2): return cv.empty_array_desc() @cv.gapi.op('custom.GParseEyes', in_types=[cv.GArray.GMat, cv.GArray.Rect, cv.GOpaque.Size], out_types=[cv.GArray.Rect, cv.GArray.Rect, cv.GArray.Point, cv.GArray.Point]) class GParseEyes: @staticmethod def outMeta(arr_desc0, arr_desc1, arr_desc2): return cv.empty_array_desc(), cv.empty_array_desc(), \ cv.empty_array_desc(), cv.empty_array_desc() @cv.gapi.op('custom.GGetStates', in_types=[cv.GArray.GMat, cv.GArray.GMat], out_types=[cv.GArray.Int, cv.GArray.Int]) class GGetStates: @staticmethod def outMeta(arr_desc0, arr_desc1): return cv.empty_array_desc(), cv.empty_array_desc() # ------------------------Custom kernels------------------------ @cv.gapi.kernel(GProcessPoses) class GProcessPosesImpl: """ Custom kernel. Processed poses of heads """ @staticmethod def run(in_ys, in_ps, in_rs): """ Сustom kernel executable code Params: in_ys: yaw angle of head in_ps: pitch angle of head in_rs: roll angle of head Return: Arrays with heads poses """ return [np.array([ys[0], ps[0], rs[0]]).T for ys, ps, rs in zip(in_ys, in_ps, in_rs)] @cv.gapi.kernel(GParseEyes) class GParseEyesImpl: """ Custom kernel. Get information about eyes """ @staticmethod def run(in_landm_per_face, in_face_rcs, frame_size): """ Сustom kernel executable code Params: in_landm_per_face: landmarks from inference of facial-landmarks network for each face in_face_rcs: bounding boxes for each face frame_size: size of input image Return: Arrays of ROI for left and right eyes, array of midpoints and array of landmarks points """ left_eyes = [] right_eyes = [] midpoints = [] lmarks = [] surface = (0, 0, *frame_size) for landm_face, rect in zip(in_landm_per_face, in_face_rcs): points = process_landmarks(*rect, landm_face) lmarks.extend(points) rect, midpoint_l = eye_box(points[0], points[1]) left_eyes.append(intersection(surface, rect)) rect, midpoint_r = eye_box(points[2], points[3]) right_eyes.append(intersection(surface, rect)) midpoints.append(midpoint_l) midpoints.append(midpoint_r) return left_eyes, right_eyes, midpoints, lmarks @cv.gapi.kernel(GGetStates) class GGetStatesImpl: """ Custom kernel. Get state of eye - open or closed """ @staticmethod def run(eyesl, eyesr): """ Сustom kernel executable code Params: eyesl: result of inference of open-closed-eye network for left eye eyesr: result of inference of open-closed-eye network for right eye Return: States of left eyes and states of right eyes """ out_l_st = [int(st) for eye_l in eyesl for st in (eye_l[:, 0] < eye_l[:, 1]).ravel()] out_r_st = [int(st) for eye_r in eyesr for st in (eye_r[:, 0] < eye_r[:, 1]).ravel()] return out_l_st, out_r_st if __name__ == '__main__': ARGUMENTS = build_argparser().parse_args() # ------------------------Demo's graph------------------------ g_in = cv.GMat() # Detect faces face_inputs = cv.GInferInputs() face_inputs.setInput('data', g_in) face_outputs = cv.gapi.infer('face-detection', face_inputs) faces = face_outputs.at('detection_out') # Parse faces sz = cv.gapi.streaming.size(g_in) faces_rc = cv.gapi.parseSSD(faces, sz, 0.5, False, False) # Detect poses head_inputs = cv.GInferInputs() head_inputs.setInput('data', g_in) face_outputs = cv.gapi.infer('head-pose', faces_rc, head_inputs) angles_y = face_outputs.at('angle_y_fc') angles_p = face_outputs.at('angle_p_fc') angles_r = face_outputs.at('angle_r_fc') # Parse poses heads_pos = GProcessPoses.on(angles_y, angles_p, angles_r) # Detect landmarks landmark_inputs = cv.GInferInputs() landmark_inputs.setInput('data', g_in) landmark_outputs = cv.gapi.infer('facial-landmarks', faces_rc, landmark_inputs) landmark = landmark_outputs.at('align_fc3') # Parse landmarks left_eyes, right_eyes, mids, lmarks = GParseEyes.on(landmark, faces_rc, sz) # Detect eyes eyes_inputs = cv.GInferInputs() eyes_inputs.setInput('input.1', g_in) eyesl_outputs = cv.gapi.infer('open-closed-eye', left_eyes, eyes_inputs) eyesr_outputs = cv.gapi.infer('open-closed-eye', right_eyes, eyes_inputs) eyesl = eyesl_outputs.at('19') eyesr = eyesr_outputs.at('19') # Process eyes states l_eye_st, r_eye_st = GGetStates.on(eyesl, eyesr) # Gaze estimation gaze_inputs = cv.GInferListInputs() gaze_inputs.setInput('left_eye_image', left_eyes) gaze_inputs.setInput('right_eye_image', right_eyes) gaze_inputs.setInput('head_pose_angles', heads_pos) gaze_outputs = cv.gapi.infer2('gaze-estimation', g_in, gaze_inputs) gaze_vectors = gaze_outputs.at('gaze_vector') out = cv.gapi.copy(g_in) # ------------------------End of graph------------------------ comp = cv.GComputation(cv.GIn(g_in), cv.GOut(out, faces_rc, left_eyes, right_eyes, gaze_vectors, angles_y, angles_p, angles_r, l_eye_st, r_eye_st, mids, lmarks)) # Networks face_net = cv.gapi.ie.params('face-detection', ARGUMENTS.facem, weight_path(ARGUMENTS.facem), ARGUMENTS.faced) head_pose_net = cv.gapi.ie.params('head-pose', ARGUMENTS.headm, weight_path(ARGUMENTS.headm), ARGUMENTS.headd) landmarks_net = cv.gapi.ie.params('facial-landmarks', ARGUMENTS.landm, weight_path(ARGUMENTS.landm), ARGUMENTS.landd) gaze_net = cv.gapi.ie.params('gaze-estimation', ARGUMENTS.gazem, weight_path(ARGUMENTS.gazem), ARGUMENTS.gazed) eye_net = cv.gapi.ie.params('open-closed-eye', ARGUMENTS.eyem, weight_path(ARGUMENTS.eyem), ARGUMENTS.eyed) nets = cv.gapi.networks(face_net, head_pose_net, landmarks_net, gaze_net, eye_net) # Kernels pack kernels = cv.gapi.kernels(GParseEyesImpl, GProcessPosesImpl, GGetStatesImpl) # ------------------------Execution part------------------------ ccomp = comp.compileStreaming(args=cv.gapi.compile_args(kernels, nets)) source = cv.gapi.wip.make_capture_src(ARGUMENTS.input) ccomp.setSource(cv.gin(source)) ccomp.start() frames = 0 fps = 0 print('Processing') START_TIME = time.time() while True: start_time_cycle = time.time() has_frame, (oimg, outr, l_eyes, r_eyes, outg, out_y, out_p, out_r, out_st_l, out_st_r, out_mids, outl) = ccomp.pull() if not has_frame: break # Draw GREEN = (0, 255, 0) RED = (0, 0, 255) WHITE = (255, 255, 255) BLUE = (255, 0, 0) PINK = (255, 0, 255) YELLOW = (0, 255, 255) M_PI_180 = np.pi / 180 M_PI_2 = np.pi / 2 M_PI = np.pi FACES_SIZE = len(outr) for i, out_rect in enumerate(outr): # Face box cv.rectangle(oimg, out_rect, WHITE, 1) rx, ry, rwidth, rheight = out_rect # Landmarks lm_radius = int(0.01 * rwidth + 1) lmsize = int(len(outl) / FACES_SIZE) for j in range(lmsize): cv.circle(oimg, outl[j + i * lmsize], lm_radius, YELLOW, -1) # Headposes yaw = out_y[i] pitch = out_p[i] roll = out_r[i] sin_y = np.sin(yaw[:] * M_PI_180) sin_p = np.sin(pitch[:] * M_PI_180) sin_r = np.sin(roll[:] * M_PI_180) cos_y = np.cos(yaw[:] * M_PI_180) cos_p = np.cos(pitch[:] * M_PI_180) cos_r = np.cos(roll[:] * M_PI_180) axis_length = 0.4 * rwidth x_center = int(rx + rwidth / 2) y_center = int(ry + rheight / 2) # center to right cv.line(oimg, [x_center, y_center], [int(x_center + axis_length * (cos_r * cos_y + sin_y * sin_p * sin_r)), int(y_center + axis_length * cos_p * sin_r)], RED, 2) # center to top cv.line(oimg, [x_center, y_center], [int(x_center + axis_length * (cos_r * sin_y * sin_p + cos_y * sin_r)), int(y_center - axis_length * cos_p * cos_r)], GREEN, 2) # center to forward cv.line(oimg, [x_center, y_center], [int(x_center + axis_length * sin_y * cos_p), int(y_center + axis_length * sin_p)], PINK, 2) scale_box = 0.002 * rwidth cv.putText(oimg, "head pose: (y=%0.0f, p=%0.0f, r=%0.0f)" % (np.round(yaw), np.round(pitch), np.round(roll)), [int(rx), int(ry + rheight + 5 * rwidth / 100)], cv.FONT_HERSHEY_PLAIN, scale_box * 2, WHITE, 1) # Eyes boxes color_l = GREEN if out_st_l[i] else RED cv.rectangle(oimg, l_eyes[i], color_l, 1) color_r = GREEN if out_st_r[i] else RED cv.rectangle(oimg, r_eyes[i], color_r, 1) # Gaze vectors norm_gazes = np.linalg.norm(outg[i][0]) gaze_vector = outg[i][0] / norm_gazes arrow_length = 0.4 * rwidth gaze_arrow = [arrow_length * gaze_vector[0], -arrow_length * gaze_vector[1]] left_arrow = [int(a+b) for a, b in zip(out_mids[0 + i * 2], gaze_arrow)] right_arrow = [int(a+b) for a, b in zip(out_mids[1 + i * 2], gaze_arrow)] if out_st_l[i]: cv.arrowedLine(oimg, out_mids[0 + i * 2], left_arrow, BLUE, 2) if out_st_r[i]: cv.arrowedLine(oimg, out_mids[1 + i * 2], right_arrow, BLUE, 2) v0, v1, v2 = outg[i][0] gaze_angles = [180 / M_PI * (M_PI_2 + np.arctan2(v2, v0)), 180 / M_PI * (M_PI_2 - np.arccos(v1 / norm_gazes))] cv.putText(oimg, "gaze angles: (h=%0.0f, v=%0.0f)" % (np.round(gaze_angles[0]), np.round(gaze_angles[1])), [int(rx), int(ry + rheight + 12 * rwidth / 100)], cv.FONT_HERSHEY_PLAIN, scale_box * 2, WHITE, 1) # Add FPS value to frame cv.putText(oimg, "FPS: %0i" % (fps), [int(20), int(40)], cv.FONT_HERSHEY_PLAIN, 2, RED, 2) # Show result cv.imshow('Gaze Estimation', oimg) cv.waitKey(1) fps = int(1. / (time.time() - start_time_cycle)) frames += 1 EXECUTION_TIME = time.time() - START_TIME print('Execution successful') print('Mean FPS is ', int(frames / EXECUTION_TIME))