458 lines
16 KiB
Python
458 lines
16 KiB
Python
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))
|