Social_Distancing_Monitor.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Social_Distancing_Monitor.ipynb",
      "provenance": [],
      "collapsed_sections": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "accelerator": "GPU"
  },
  "cells": [
    {
      "cell_type": "code",
      "metadata": {
        "id": "Ygq4ZQabrDEE"
      },
      "source": [
        "# Mount Google Drive (If using Colab)\n",
        "\n",
        "from google.colab import drive\n",
        "drive.mount(\"drive\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "tvGhWHSHfU8B"
      },
      "source": [
        "# Install Required Libraries from PyPI\n",
        "\n",
        "!pip install git+https://github.com/hukkelas/DSFD-Pytorch-Inference.git\n",
        "!pip install tqdm"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "KgpQJHNSfsK6"
      },
      "source": [
        "# Import Required Libraries\n",
        "\n",
        "import os\n",
        "import numpy as np\n",
        "import cv2\n",
        "import face_detection \n",
        "from sklearn.cluster import DBSCAN\n",
        "from keras.models import load_model\n",
        "from keras.applications.resnet50 import preprocess_input\n",
        "import tqdm\n",
        "from google.colab.patches import cv2_imshow"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ygki7lR_sFvf"
      },
      "source": [
        "# Path to the Working Environment\n",
        "\n",
        "# If using Google Colab (If on a Local Environment, no path required => set BASE_PATH  = \"\")\n",
        "BASE_PATH = \"drive/My Drive/Social_Distancing_with_AI/\"\n",
        "\n",
        "# Path to Input Video File in the BASE_PATH\n",
        "FILE_PATH = \"test_video.mp4\""
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2xrRSzoT9EBa"
      },
      "source": [
        "# Initialize a Face Detector \n",
        "# Confidence Threshold can be Adjusted, Greater values would Detect only Clear Faces\n",
        "\n",
        "detector = face_detection.build_detector(\"DSFDDetector\", confidence_threshold=.5, nms_iou_threshold=.3)\n",
        "\n",
        "# Other Availaible Detectors are RetinaNetResNet50 & RetinaNetMobileNetV1"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Kv-5woacC0C5"
      },
      "source": [
        "# Load Pretrained Face Mask Classfier (Keras Model)\n",
        "\n",
        "mask_classifier = load_model(\"drive/My Drive/Social_Distancing_with_AI/Models/ResNet50_Classifier.h5\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "8O6mF_orCxN4"
      },
      "source": [
        "# Set the Safe Distance in Pixel Units (Minimum Distance Expected to be Maintained between People)\n",
        "# This Parameter would Affect the Results, Adjust according to the Footage captured by CCTV Camera \n",
        "\n",
        "threshold_distance = 150 # Try with different Values before Finalizing"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ejHxuy-9iZwL"
      },
      "source": [
        "##################################### Analyze the Video ################################################\n",
        "\n",
        "# Load YOLOv3\n",
        "net = cv2.dnn.readNet(BASE_PATH+\"Models/\"+\"yolov3.weights\", BASE_PATH+\"Models/\"+\"yolov3.cfg\")\n",
        "\n",
        "# Load COCO Classes\n",
        "classes = []\n",
        "with open(BASE_PATH+\"Models/\"+\"coco.names\", \"r\") as f:\n",
        "    classes = [line.strip() for line in f.readlines()]\n",
        "    \n",
        "layer_names = net.getLayerNames()\n",
        "output_layers = [layer_names[i[0] - 1] for i in net.getUnconnectedOutLayers()]\n",
        "\n",
        "# Fetch Video Properties\n",
        "cap = cv2.VideoCapture(BASE_PATH + FILE_PATH )\n",
        "fps = cap.get(cv2.CAP_PROP_FPS)\n",
        "width = cap.get(cv2.CAP_PROP_FRAME_WIDTH)\n",
        "height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT)\n",
        "n_frames = cap.get(cv2.CAP_PROP_FRAME_COUNT)\n",
        "\n",
        "# Create Directory for Storing Results (Make sure it doesn't already exists !)\n",
        "os.mkdir(BASE_PATH+\"Results\")\n",
        "os.mkdir(BASE_PATH+\"Results/Extracted_Faces\")\n",
        "os.mkdir(BASE_PATH+\"Results/Extracted_Persons\")\n",
        "os.mkdir(BASE_PATH+\"Results/Frames\")\n",
        "\n",
        "# Initialize Output Video Stream\n",
        "out_stream = cv2.VideoWriter(\n",
        "    BASE_PATH+'Results/Output.mp4',\n",
        "    cv2.VideoWriter_fourcc('X','V','I','D'),\n",
        "    fps,\n",
        "    (int(width),int(height)))\n",
        "\n",
        "print(\"Processing Frames :\")\n",
        "for frame in tqdm.notebook.tqdm(range(int(n_frames))):\n",
        "    \n",
        "    # Capture Frame-by-Frame\n",
        "    ret, img = cap.read()\n",
        "\n",
        "    # Check EOF\n",
        "    if ret == False:\n",
        "        break;\n",
        "\n",
        "    # Get Frame Dimentions\n",
        "    height, width, channels = img.shape\n",
        "\n",
        "    # Detect Objects in the Frame with YOLOv3\n",
        "    blob = cv2.dnn.blobFromImage(img, 0.00392, (416,416), (0, 0, 0), True, crop=False)\n",
        "    net.setInput(blob)\n",
        "    outs = net.forward(output_layers)\n",
        "\n",
        "    class_ids = []\n",
        "    confidences = []\n",
        "    boxes = []\n",
        "    \n",
        "    # Store Detected Objects with Labels, Bounding_Boxes and their Confidences\n",
        "    for out in outs:\n",
        "        for detection in out:\n",
        "            scores = detection[5:]\n",
        "            class_id = np.argmax(scores)\n",
        "            confidence = scores[class_id]\n",
        "            if confidence > 0.5:\n",
        "                \n",
        "                # Get Center, Height and Width of the Box\n",
        "                center_x = int(detection[0] * width)\n",
        "                center_y = int(detection[1] * height)\n",
        "                w = int(detection[2] * width)\n",
        "                h = int(detection[3] * height)\n",
        "\n",
        "                # Topleft Co-ordinates\n",
        "                x = int(center_x - w / 2)\n",
        "                y = int(center_y - h / 2)\n",
        "\n",
        "                boxes.append([x, y, w, h])\n",
        "                confidences.append(float(confidence))\n",
        "                class_ids.append(class_id)\n",
        "\n",
        "    indexes = cv2.dnn.NMSBoxes(boxes, confidences, 0.5, 0.4)\n",
        "\n",
        "    # Initialize empty lists for storing Bounding Boxes of People and their Faces\n",
        "    persons = []\n",
        "    masked_faces = []\n",
        "    unmasked_faces = []\n",
        "\n",
        "    # Work on Detected Persons in the Frame\n",
        "    for i in range(len(boxes)):\n",
        "        if i in indexes:\n",
        "\n",
        "            box = np.array(boxes[i])\n",
        "            box = np.where(box<0,0,box)\n",
        "            (x, y, w, h) = box\n",
        "\n",
        "            label = str(classes[class_ids[i]])\n",
        "\n",
        "            if label=='person':\n",
        "\n",
        "                persons.append([x,y,w,h])\n",
        "                \n",
        "                # Save Image of Cropped Person (If not required, comment the command below)\n",
        "                cv2.imwrite(BASE_PATH + \"Results/Extracted_Persons/\"+str(frame)\n",
        "                            +\"_\"+str(len(persons))+\".jpg\",\n",
        "                            img[y:y+h,x:x+w])\n",
        "\n",
        "                # Detect Face in the Person\n",
        "                person_rgb = img[y:y+h,x:x+w,::-1]   # Crop & BGR to RGB\n",
        "                detections = detector.detect(person_rgb)\n",
        "\n",
        "                # If a Face is Detected\n",
        "                if detections.shape[0] > 0:\n",
        "\n",
        "                  detection = np.array(detections[0])\n",
        "                  detection = np.where(detection<0,0,detection)\n",
        "\n",
        "                  # Calculating Co-ordinates of the Detected Face\n",
        "                  x1 = x + int(detection[0])\n",
        "                  x2 = x + int(detection[2])\n",
        "                  y1 = y + int(detection[1])\n",
        "                  y2 = y + int(detection[3])\n",
        "\n",
        "                  try :\n",
        "\n",
        "                    # Crop & BGR to RGB\n",
        "                    face_rgb = img[y1:y2,x1:x2,::-1]   \n",
        "\n",
        "                    # Preprocess the Image\n",
        "                    face_arr = cv2.resize(face_rgb, (224, 224), interpolation=cv2.INTER_NEAREST)\n",
        "                    face_arr = np.expand_dims(face_arr, axis=0)\n",
        "                    face_arr = preprocess_input(face_arr)\n",
        "\n",
        "                    # Predict if the Face is Masked or Not\n",
        "                    score = mask_classifier.predict(face_arr)\n",
        "\n",
        "                    # Determine and store Results\n",
        "                    if score[0][0]<0.5:\n",
        "                      masked_faces.append([x1,y1,x2,y2])\n",
        "                    else:\n",
        "                      unmasked_faces.append([x1,y1,x2,y2])\n",
        "\n",
        "                    # Save Image of Cropped Face (If not required, comment the command below)\n",
        "                    cv2.imwrite(BASE_PATH + \"Results/Extracted_Faces/\"+str(frame)\n",
        "                                +\"_\"+str(len(persons))+\".jpg\",\n",
        "                                img[y1:y2,x1:x2])\n",
        "\n",
        "                  except:\n",
        "                    continue\n",
        "    \n",
        "    # Calculate Coordinates of People Detected and find Clusters using DBSCAN\n",
        "    person_coordinates = []\n",
        "\n",
        "    for p in range(len(persons)):\n",
        "      person_coordinates.append((persons[p][0]+int(persons[p][2]/2),persons[p][1]+int(persons[p][3]/2)))\n",
        "\n",
        "    if len(persons)>0:\n",
        "      clustering = DBSCAN(eps=threshold_distance,min_samples=2).fit(person_coordinates)\n",
        "      isSafe = clustering.labels_\n",
        "    else:\n",
        "      isSafe = []\n",
        "\n",
        "    # Count \n",
        "    person_count = len(persons)\n",
        "    masked_face_count = len(masked_faces)\n",
        "    unmasked_face_count = len(unmasked_faces)\n",
        "    safe_count = np.sum((isSafe==-1)*1)\n",
        "    unsafe_count = person_count - safe_count\n",
        "\n",
        "    # Show Clusters using Red Lines\n",
        "    arg_sorted = np.argsort(isSafe)\n",
        "\n",
        "    for i in range(1,person_count):\n",
        "\n",
        "      if isSafe[arg_sorted[i]]!=-1 and isSafe[arg_sorted[i]]==isSafe[arg_sorted[i-1]]:\n",
        "        cv2.line(img,person_coordinates[arg_sorted[i]],person_coordinates[arg_sorted[i-1]],(0,0,255),2)\n",
        "\n",
        "    # Put Bounding Boxes on People in the Frame\n",
        "    for p in range(person_count):\n",
        "\n",
        "      a,b,c,d = persons[p]\n",
        "\n",
        "      # Green if Safe, Red if UnSafe\n",
        "      if isSafe[p]==-1:\n",
        "        cv2.rectangle(img, (a, b), (a + c, b + d), (0,255,0), 2)\n",
        "      else:\n",
        "        cv2.rectangle(img, (a, b), (a + c, b + d), (0,0,255), 2)\n",
        "\n",
        "    # Put Bounding Boxes on Faces in the Frame\n",
        "    # Green if Safe, Red if UnSafe\n",
        "    for f in range(masked_face_count):\n",
        "\n",
        "      a,b,c,d = masked_faces[f]\n",
        "      cv2.rectangle(img, (a, b), (c,d), (0,255,0), 2)\n",
        "\n",
        "    for f in range(unmasked_face_count):\n",
        "\n",
        "      a,b,c,d = unmasked_faces[f]\n",
        "      cv2.rectangle(img, (a, b), (c,d), (0,0,255), 2)\n",
        "\n",
        "    # Show Monitoring Status in a Black Box at the Top\n",
        "    cv2.rectangle(img,(0,0),(width,50),(0,0,0),-1)\n",
        "    cv2.rectangle(img,(1,1),(width-1,50),(255,255,255),2)\n",
        "\n",
        "    xpos = 15\n",
        "\n",
        "    string = \"Total People = \"+str(person_count)\n",
        "    cv2.putText(img,string,(xpos,35),cv2.FONT_HERSHEY_SIMPLEX,1,(255,255,255),2)\n",
        "    xpos += cv2.getTextSize(string,cv2.FONT_HERSHEY_SIMPLEX,1,2)[0][0]\n",
        "\n",
        "    string = \" ( \"+str(safe_count) + \" Safe \"\n",
        "    cv2.putText(img,string,(xpos,35),cv2.FONT_HERSHEY_SIMPLEX,1,(0,255,0),2)\n",
        "    xpos += cv2.getTextSize(string,cv2.FONT_HERSHEY_SIMPLEX,1,2)[0][0]\n",
        "\n",
        "    string = str(unsafe_count)+ \" Unsafe ) \"\n",
        "    cv2.putText(img,string,(xpos,35),cv2.FONT_HERSHEY_SIMPLEX,1,(0,0,255),2)\n",
        "    xpos += cv2.getTextSize(string,cv2.FONT_HERSHEY_SIMPLEX,1,2)[0][0]\n",
        "    \n",
        "    string = \"( \" +str(masked_face_count)+\" Masked \"+str(unmasked_face_count)+\" Unmasked \"+\\\n",
        "             str(person_count-masked_face_count-unmasked_face_count)+\" Unknown )\"\n",
        "    cv2.putText(img,string,(xpos,35),cv2.FONT_HERSHEY_SIMPLEX,1,(0,255,255),2)\n",
        "\n",
        "    # Write Frame to the Output File\n",
        "    out_stream.write(img)\n",
        "\n",
        "    # Save the Frame in frame_no.png format (If not required, comment the command below)\n",
        "    cv2.imwrite(BASE_PATH+\"Results/Frames/\"+str(frame)+\".jpg\",img)\n",
        "\n",
        "    # Use if you want to see Results Frame by Frame\n",
        "    # cv2_imshow('results',img)\n",
        "\n",
        "    # Exit on Pressing Q Key\n",
        "    if cv2.waitKey(25) & 0xFF == ord('q'):\n",
        "        break\n",
        "\n",
        "# Release Streams\n",
        "out_stream.release()\n",
        "cap.release()\n",
        "cv2.destroyAllWindows()\n",
        "\n",
        "# Good to Go!\n",
        "print(\"Done !\")"
      ],
      "execution_count": null,
      "outputs": []
    }
  ]
}