DepthAI Fatigue detection with mono camera

fatigue-detection-mono.py

import blobconverter
import cv2
import depthai as dai
import face_landmarks

openvinoVersion = "2020.3"
pipeline = dai.Pipeline()

# Define sources and outputs
monoRight = pipeline.create(dai.node.MonoCamera)
monoRight.setBoardSocket(dai.CameraBoardSocket.RIGHT)
monoRight.setResolution(dai.MonoCameraProperties.SensorResolution.THE_720_P)

# Crop 1280x720 -> 720x720. Convert grayscale to BGR
rgbManip = pipeline.create(dai.node.ImageManip)
rgbManip.initialConfig.setResize(720, 720)
rgbManip.setMaxOutputFrameSize(720*720*3)
rgbManip.initialConfig.setFrameType(dai.ImgFrame.Type.BGR888p)
monoRight.out.link(rgbManip.inputImage)

resizeManip = pipeline.create(dai.node.ImageManip)
resizeManip.initialConfig.setResize(300, 300)
rgbManip.out.link(resizeManip.inputImage)

manipOut = pipeline.create(dai.node.XLinkOut)
manipOut.setStreamName("right")
rgbManip.out.link(manipOut.input)

# NN that detects faces in the image
face_nn = pipeline.create(dai.node.MobileNetDetectionNetwork)
face_nn.setConfidenceThreshold(0.3)
face_nn.setBlobPath(blobconverter.from_zoo("face-detection-retail-0004", shaves=6, version=openvinoVersion))
resizeManip.out.link(face_nn.input)

# # Send ImageManipConfig to host so it can visualize the landmarks
config_xout = pipeline.create(dai.node.XLinkOut)
config_xout.setStreamName("face_det")
face_nn.out.link(config_xout.input)

# Script node will take the output from the NN as an input, get the first bounding box
# and send ImageManipConfig to the manip_crop
image_manip_script = pipeline.create(dai.node.Script)
face_nn.out.link(image_manip_script.inputs['nn_in'])
#cam.preview.link(image_manip_script.inputs['frame'])
rgbManip.out.link(image_manip_script.inputs['frame'])
image_manip_script.setScript("""
import time
def enlrage_roi(det): # For better face landmarks NN results
    det.xmin -= 0.05
    det.ymin -= 0.02
    det.xmax += 0.05
    det.ymax += 0.02
def limit_roi(det):
    if det.xmin <= 0: det.xmin = 0.001
    if det.ymin <= 0: det.ymin = 0.001
    if det.xmax >= 1: det.xmax = 0.999
    if det.ymax >= 1: det.ymax = 0.999

while True:
    frame = node.io['frame'].get()
    face_dets = node.io['nn_in'].get().detections

    #node.warn(f"Faces detected: {len(face_dets)}")
    for det in face_dets:
        enlrage_roi(det)
        limit_roi(det)
        # node.warn(f"Detection rect: {det.xmin}, {det.ymin}, {det.xmax}, {det.ymax}")
        cfg = ImageManipConfig()
        cfg.setCropRect(det.xmin, det.ymin, det.xmax, det.ymax)
        cfg.setResize(160, 160)
        cfg.setKeepAspectRatio(False)
        node.io['manip_cfg'].send(cfg)
        node.io['manip_img'].send(frame)
        # node.warn(f"1 from nn_in: {det.xmin}, {det.ymin}, {det.xmax}, {det.ymax}")
""")

# This ImageManip will crop the mono frame based on the NN detections. Resulting image will be the cropped
# face that was detected by the face-detection NN.
manip_crop = pipeline.create(dai.node.ImageManip)
image_manip_script.outputs['manip_img'].link(manip_crop.inputImage)
image_manip_script.outputs['manip_cfg'].link(manip_crop.inputConfig)
manip_crop.initialConfig.setResize(160, 160)
manip_crop.setWaitForConfigInput(True)

# Second NN that detcts emotions from the cropped 64x64 face
landmarks_nn = pipeline.createNeuralNetwork()
# landmarks_nn.setBlobPath(blobconverter.from_zoo(name="facial_landmarks_68_160x160", zoo_type="depthai", version=openvinoVersion))
landmarks_nn.setBlobPath("models/face_landmark_160x160_openvino_2020_1_4shave.blob")
manip_crop.out.link(landmarks_nn.input)

landmarks_nn_xout = pipeline.createXLinkOut()
landmarks_nn_xout.setStreamName("nn")
landmarks_nn.out.link(landmarks_nn_xout.input)

# Pipeline is defined, now we can connect to the device
with dai.Device(pipeline) as device:
    videoQ = device.getOutputQueue(name="right", maxSize=1, blocking=False)
    faceDetQ = device.getOutputQueue(name="face_det", maxSize=1, blocking=False)
    nnQ = device.getOutputQueue(name="nn", maxSize=4, blocking=False)

    decode = face_landmarks.FaceLandmarks()

    while True:
        frame = videoQ.get().getCvFrame()

        if nnQ.has():
            nndata = nnQ.get()
            dets = faceDetQ.get().detections
            if 0 < len(dets):
                faceDet = dets[0]
                face_coords = [faceDet.xmin, faceDet.ymin,faceDet.xmax,faceDet.ymax]
                decode.run_land68(frame, nndata, face_coords)

        cv2.imshow("frame", frame)

        if cv2.waitKey(1) == ord('q'):
            break