import copy import os import sys import argparse parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument("-i", "--ip", help="Set IP address for sending tracking data", default="127.0.0.1") parser.add_argument("-p", "--port", type=int, help="Set port for sending tracking data", default=11573) if os.name == 'nt': parser.add_argument("-l", "--list-cameras", type=int, help="Set this to 1 to list the available cameras and quit, set this to a number higher than 1 to output only the names", default=0) parser.add_argument("-W", "--width", type=int, help="Set camera and raw RGB width", default=640) parser.add_argument("-H", "--height", type=int, help="Set camera and raw RGB height", default=360) parser.add_argument("-F", "--fps", type=int, help="Set camera frames per second", default=24) else: parser.add_argument("-W", "--width", type=int, help="Set raw RGB width", default=640) parser.add_argument("-H", "--height", type=int, help="Set raw RGB height", default=360) parser.add_argument("-c", "--capture", help="Set camera ID (0, 1...) or video file", default="0") parser.add_argument("-m", "--max-threads", type=int, help="Set the maximum number of threads", default=1) parser.add_argument("-t", "--threshold", type=float, help="Set minimum confidence threshold for face detection", default=0.4) parser.add_argument("-v", "--visualize", type=int, help="Set this to 1 to visualize the tracking, to 2 to also show face ids or to 3 to add numbers to the point display", default=0) parser.add_argument("-P", "--pnp-points", type=int, help="Set this to 1 to add the 3D fitting points to the visualization", default=0) parser.add_argument("-s", "--silent", type=int, help="Set this to 1 to prevent text output on the console", default=0) parser.add_argument("--faces", type=int, help="Set the maximum number of faces (slow)", default=1) parser.add_argument("--scan-every", type=int, help="Set after how many frames a scan for new faces should run", default=3) parser.add_argument("--discard-after", type=int, help="Set the how long the tracker should keep looking for lost faces", default=10) parser.add_argument("--video-out", help="Set this to the filename of an AVI file to save the tracking visualization as a video", default=None) parser.add_argument("--raw-rgb", type=int, help="When this is set, raw RGB frames of the size given with \"-W\" and \"-H\" are read from standard input instead of reading a video", default=0) parser.add_argument("--log-data", help="You can set a filename to which tracking data will be logged here", default="") parser.add_argument("--model", type=int, help="This can be used to select the tracking model. Higher numbers are models with better tracking quality, but slower speed. The recommended models are 3 and 1.", default=3, choices=[0, 1, 2, 3]) parser.add_argument("--model-dir", help="This can be used to specify the path to the directory containing the .onnx model files", default=None) parser.add_argument("--high-quality-3d", type=int, help="When set to 1, more nose points are used when estimating the face pose", default=1) parser.add_argument("--gaze-tracking", type=int, help="When set to 1, experimental blink detection and gaze tracking are enabled, which makes things slightly slower", default=1) parser.add_argument("--face-id-offset", type=int, help="When set, this offset is added to all face ids, which can be useful for mixing tracking data from multiple network sources", default=0) args = parser.parse_args() os.environ["OMP_NUM_THREADS"] = "1" if os.name == 'nt' and args.list_cameras > 0: import escapi escapi.init() camera_count = escapi.count_capture_devices() if args.list_cameras == 1: print("Available cameras:") for i in range(camera_count): camera_name = escapi.device_name(i).decode() if args.list_cameras == 1: print(f"{i}: {camera_name}") else: print(camera_name) sys.exit(0) import numpy as np import time import cv2 import socket import struct from input_reader import InputReader, list_cameras from tracker import Tracker target_ip = args.ip target_port = args.port if args.faces >= 40: print("Transmission of tracking data over network is not supported with 40 or more faces.") fps = 0 if os.name == 'nt': fps = args.fps input_reader = InputReader(args.capture, args.raw_rgb, args.width, args.height, fps) log = None out = None first = True height = 0 width = 0 tracker = None sock = None tracking_time = 0.0 tracking_frames = 0 frame_count = 0 if args.log_data != "": log = open(args.log_data, "w") log.write("Frame,Time,Width,Height,FPS,Face,FaceID,RightOpen,LeftOpen,AverageConfidence,Success3D,PnPError,RotationQuat.X,RotationQuat.Y,RotationQuat.Z,RotationQuat.W,Euler.X,Euler.Y,Euler.Z,RVec.X,RVec.Y,RVec.Z,TVec.X,TVec.Y,TVec.Z") for i in range(66): log.write(f",Landmark[{i}].X,Landmark[{i}].Y,Landmark[{i}].Confidence") for i in range(66): log.write(f",Point3D[{i}].X,Point3D[{i}].Y,Point3D[{i}].Z") log.write("\r\n") log.flush() try: while input_reader.is_open(): if not input_reader.is_ready(): time.sleep(0.001) continue ret, frame = input_reader.read() if not ret: break frame_count += 1 now = time.time() if first: first = False height, width, channels = frame.shape sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) tracker = Tracker(width, height, threshold=args.threshold, max_threads=args.max_threads, max_faces=args.faces, discard_after=args.discard_after, scan_every=args.scan_every, silent=False if args.silent == 0 else True, model_type=args.model, pnp_quality=args.high_quality_3d, model_dir=args.model_dir, no_gaze=False if args.gaze_tracking != 0 else True) if not args.video_out is None: out = cv2.VideoWriter(args.video_out, cv2.VideoWriter_fourcc('F','F','V','1'), 24, (width,height)) inference_start = time.perf_counter() faces = tracker.predict(frame) if len(faces) > 0: tracking_time += (time.perf_counter() - inference_start) / len(faces) tracking_frames += 1 packet = bytearray() detected = False for face_num, f in enumerate(faces): f = copy.copy(f) f.id += args.face_id_offset right_state = "O" if f.eye_blink[0] > 0.8 else "-" left_state = "O" if f.eye_blink[1] > 0.8 else "-" if args.silent == 0: print(f"Confidence[{f.id}]: {f.conf:.4f} / 3D fitting error: {f.pnp_error:.4f} / Eyes: {left_state}, {right_state}") detected = True if not f.success: pts_3d = np.zeros((70, 3), np.float32) packet.extend(bytearray(struct.pack("d", now))) packet.extend(bytearray(struct.pack("i", f.id))) packet.extend(bytearray(struct.pack("f", width))) packet.extend(bytearray(struct.pack("f", height))) packet.extend(bytearray(struct.pack("f", f.eye_blink[0]))) packet.extend(bytearray(struct.pack("f", f.eye_blink[1]))) packet.extend(bytearray(struct.pack("B", 1 if f.success else 0))) packet.extend(bytearray(struct.pack("f", f.pnp_error))) packet.extend(bytearray(struct.pack("f", f.quaternion[0]))) packet.extend(bytearray(struct.pack("f", f.quaternion[1]))) packet.extend(bytearray(struct.pack("f", f.quaternion[2]))) packet.extend(bytearray(struct.pack("f", f.quaternion[3]))) packet.extend(bytearray(struct.pack("f", f.euler[0]))) packet.extend(bytearray(struct.pack("f", f.euler[1]))) packet.extend(bytearray(struct.pack("f", f.euler[2]))) packet.extend(bytearray(struct.pack("f", f.translation[0]))) packet.extend(bytearray(struct.pack("f", f.translation[1]))) packet.extend(bytearray(struct.pack("f", f.translation[2]))) if not log is None: log.write(f"{frame_count},{now},{width},{height},{args.fps},{face_num},{f.id},{f.eye_blink[0]},{f.eye_blink[1]},{f.conf},{f.success},{f.pnp_error},{f.quaternion[0]},{f.quaternion[1]},{f.quaternion[2]},{f.quaternion[3]},{f.euler[0]},{f.euler[1]},{f.euler[2]},{f.rotation[0]},{f.rotation[1]},{f.rotation[2]},{f.translation[0]},{f.translation[1]},{f.translation[2]}") for (x,y,c) in f.lms: packet.extend(bytearray(struct.pack("f", c))) for pt_num, (x,y,c) in enumerate(f.lms): packet.extend(bytearray(struct.pack("f", y))) packet.extend(bytearray(struct.pack("f", x))) if not log is None: log.write(f",{y},{x},{c}") if pt_num == 66 and f.eye_blink[0] < 0.5: continue if pt_num == 67 and f.eye_blink[0] < 0.5: continue x = int(x + 0.5) y = int(y + 0.5) if args.visualize != 0 or not out is None: if args.visualize > 1: frame = cv2.putText(frame, str(f.id), (int(f.bbox[0]), int(f.bbox[1])), cv2.FONT_HERSHEY_SIMPLEX, 0.75, (255,0,255)) if args.visualize > 2: frame = cv2.putText(frame, str(pt_num), (int(y), int(x)), cv2.FONT_HERSHEY_SIMPLEX, 0.25, (255,255,0)) color = (0, 0, 255) if pt_num >= 66: color = (255, 255, 0) if not (x < 0 or y < 0 or x >= height or y >= width): frame[int(x), int(y)] = color x += 1 if not (x < 0 or y < 0 or x >= height or y >= width): frame[int(x), int(y)] = color y += 1 if not (x < 0 or y < 0 or x >= height or y >= width): frame[int(x), int(y)] = color x -= 1 if not (x < 0 or y < 0 or x >= height or y >= width): frame[int(x), int(y)] = color if args.pnp_points != 0 and (args.visualize != 0 or not out is None): if args.pnp_points > 1: projected = cv2.projectPoints(f.face_3d, f.rotation, f.translation, tracker.camera, tracker.dist_coeffs) else: projected = cv2.projectPoints(f.contour, f.rotation, f.translation, tracker.camera, tracker.dist_coeffs) for [(x,y)] in projected[0]: x = int(x + 0.5) y = int(y + 0.5) if not (x < 0 or y < 0 or x >= height or y >= width): frame[int(x), int(y)] = (0, 255, 255) x += 1 if not (x < 0 or y < 0 or x >= height or y >= width): frame[int(x), int(y)] = (0, 255, 255) y += 1 if not (x < 0 or y < 0 or x >= height or y >= width): frame[int(x), int(y)] = (0, 255, 255) x -= 1 if not (x < 0 or y < 0 or x >= height or y >= width): frame[int(x), int(y)] = (0, 255, 255) for (x,y,z) in f.pts_3d: packet.extend(bytearray(struct.pack("f", x))) packet.extend(bytearray(struct.pack("f", -y))) packet.extend(bytearray(struct.pack("f", -z))) if not log is None: log.write(f",{x},{-y},{-z}") if not log is None: log.write("\r\n") log.flush() if detected and len(faces) < 40: sock.sendto(packet, (target_ip, target_port)) if not out is None: out.write(frame) if args.visualize != 0: cv2.imshow('OpenSeeFace Visualization', frame) if cv2.waitKey(1) & 0xFF == ord('q'): break except KeyboardInterrupt: if args.silent == 0: print("Quitting") input_reader.close() if not out is None: out.release() cv2.destroyAllWindows() if args.silent == 0 and tracking_frames > 0: tracking_time = 1000 * tracking_time / tracking_frames print(f"Average tracking time per detected face: {tracking_time:.2f} ms")