rekognition-apis/7-ppe-detection.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Personal Protective Equipment Detection using Amazon Rekognition"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***\n",
    "This notebook provides a walkthrough of [PPE Detection API](https://docs.aws.amazon.com/rekognition/latest/dg/ppe-detection.html) in Amazon Rekognition to detect Personal Protective Equipment (PPE) worn by persons.\n",
    "***"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Initialize stuff"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Install necessary packages\n",
    "!pip install opencv-python-headless"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initialise Notebook\n",
    "import boto3\n",
    "from IPython.display import HTML, display, Image as IImage\n",
    "from PIL import Image, ImageDraw, ExifTags, ImageColor\n",
    "import json\n",
    "import cv2\n",
    "import math\n",
    "import os\n",
    "import io"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Curent AWS Region. Use this to choose corresponding S3 bucket with sample content\n",
    "\n",
    "mySession = boto3.session.Session()\n",
    "awsRegion = mySession.region_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Init clients\n",
    "rekognition = boto3.client('rekognition')\n",
    "s3 = boto3.client('s3')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# S3 bucket that contains sample images and videos\n",
    "\n",
    "# We are providing sample images and videos in this bucket so\n",
    "# you do not have to manually download/upload test images and videos.\n",
    "\n",
    "bucketName = \"aws-rek-immersionday-\" + awsRegion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create temporary directory\n",
    "# This directory is not needed to call Rekognition APIs.\n",
    "# We will only use this directory to download images from S3 bucket and draw bounding boxes\n",
    "\n",
    "!mkdir m1tmp\n",
    "tempFolder = 'm1tmp/'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Detect Persons wearing PPEs in image\n",
    "***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "imageName = \"media/ppe-detection/ppe-image-group.jpg\"\n",
    "#imageName = \"media/ppe-detection/ppe-image-single.jpg\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "display(IImage(url=s3.generate_presigned_url('get_object', Params={'Bucket': bucketName, 'Key': imageName})))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Call Rekognition to detect persons and PPE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Call Amazon Rekognition to detect PPE in the image\n",
    "# https://docs.aws.amazon.com/rekognition/latest/dg/ppe-procedure-image.html\n",
    "detectPPEResponse = rekognition.detect_protective_equipment(\n",
    "        Image={\n",
    "            'S3Object': {\n",
    "                'Bucket': bucketName,\n",
    "                'Name': imageName\n",
    "            }\n",
    "        },\n",
    "        SummarizationAttributes={\n",
    "            'MinConfidence': 90,\n",
    "            'RequiredEquipmentTypes': [\n",
    "                'FACE_COVER',\n",
    "                'HEAD_COVER',\n",
    "                'HAND_COVER', #you can remove required equipment type from the list based on your requirements\n",
    "            ]\n",
    "        }\n",
    "    )\n",
    "\n",
    "# print(detectPPEResponse)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Review the raw JSON reponse from Rekognition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Show JSON response returned by Rekognition PPE Detection API\n",
    "# In the JSON response below, you will see detected body parts, confidence score and additional information.\n",
    "\n",
    "display(detectPPEResponse)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Call Rekognition to detect persons and PPE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Call Amazon Rekognition to detect PPE in the image\n",
    "# https://docs.aws.amazon.com/rekognition/latest/dg/ppe-procedure-image.html\n",
    "detectPPEResponse = rekognition.detect_protective_equipment(\n",
    "        Image={\n",
    "            'S3Object': {\n",
    "                'Bucket': bucketName,\n",
    "                'Name': imageName\n",
    "            }\n",
    "        },\n",
    "        SummarizationAttributes={\n",
    "            'MinConfidence': 90,\n",
    "            'RequiredEquipmentTypes': [\n",
    "                'FACE_COVER',\n",
    "                'HEAD_COVER',\n",
    "                #'HAND_COVER', #only detecting face and head cover\n",
    "            ]\n",
    "        }\n",
    "    )\n",
    "\n",
    "# print(detectPPEResponse)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Review the raw JSON reponse from Rekognition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Show JSON response returned by Rekognition PPE Detection API\n",
    "# As we changed the required equipment type list above, \n",
    "# you will see that Rekognition detected that all persons in the image are wearing required PPEs\n",
    "\n",
    "display(detectPPEResponse)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Drawing bounding boxes around PPE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# it will draw green bounding box if specific PPE is found\n",
    "# it will draw warning yellow bounding box within face mask bounding box if confidence is lower than supplied value\n",
    "# if specific PPE (such as hand cover, face cover etc.) is not found, it will draw red bounding box around body/person\n",
    "\n",
    "def detect_ppe(sourceImage, confidence):\n",
    "\n",
    "    fill_green='#00d400'\n",
    "    fill_red='#ff0000'\n",
    "    fill_yellow='#ffff00'\n",
    "    line_width=3\n",
    "    \n",
    "    imageLocation = tempFolder+os.path.basename(sourceImage)\n",
    "    s3.download_file(bucketName, sourceImage, imageLocation)\n",
    "\n",
    "    # open image and get image data from stream.\n",
    "    image = Image.open(open(imageLocation,'rb'))\n",
    "    stream = io.BytesIO()\n",
    "    image.save(stream, format=image.format)    \n",
    "    image_binary = stream.getvalue()\n",
    "    imgWidth, imgHeight = image.size  \n",
    "    draw = ImageDraw.Draw(image)  \n",
    "    \n",
    "\n",
    "    for person in detectPPEResponse['Persons']:\n",
    "        \n",
    "        found_mask=False\n",
    "\n",
    "        for body_part in person['BodyParts']:\n",
    "            ppe_items = body_part['EquipmentDetections']\n",
    "                 \n",
    "            for ppe_item in ppe_items:\n",
    "                # found a mask\n",
    "                if ppe_item['Type'] == 'FACE_COVER': # you can change it to \"HEAD_COVER\" or \"HAND_COVER\"\n",
    "                    fill_color=fill_green\n",
    "                    found_mask=True\n",
    "                    # check if mask covers face\n",
    "                    if ppe_item['CoversBodyPart']['Value'] == False:\n",
    "                        fill_color=fill='#ff0000'\n",
    "                    # draw bounding box around mask\n",
    "                    box = ppe_item['BoundingBox']\n",
    "                    left = imgWidth * box['Left']\n",
    "                    top = imgHeight * box['Top']\n",
    "                    width = imgWidth * box['Width']\n",
    "                    height = imgHeight * box['Height']\n",
    "                    points = (\n",
    "                            (left,top),\n",
    "                            (left + width, top),\n",
    "                            (left + width, top + height),\n",
    "                            (left , top + height),\n",
    "                            (left, top)\n",
    "                        )\n",
    "                    draw.line(points, fill=fill_color, width=line_width)\n",
    "\n",
    "                     # Check if confidence is lower than supplied value       \n",
    "                    if ppe_item['CoversBodyPart']['Confidence'] < confidence:\n",
    "                        # draw warning yellow bounding box within face mask bounding box\n",
    "                        offset=line_width+ line_width \n",
    "                        points = (\n",
    "                                    (left+offset,top + offset),\n",
    "                                    (left + width-offset, top+offset),\n",
    "                                    ((left) + (width-offset), (top-offset) + (height)),\n",
    "                                    (left+ offset , (top) + (height -offset)),\n",
    "                                    (left + offset, top + offset)\n",
    "                                )\n",
    "                        draw.line(points, fill=fill_yellow, width=line_width)\n",
    "                \n",
    "        if found_mask==False:\n",
    "            # no face mask found so draw red bounding box around body\n",
    "            box = person['BoundingBox']\n",
    "            left = imgWidth * box['Left']\n",
    "            top = imgHeight * box['Top']\n",
    "            width = imgWidth * box['Width']\n",
    "            height = imgHeight * box['Height']\n",
    "            points = (\n",
    "                (left,top),\n",
    "                (left + width, top),\n",
    "                (left + width, top + height),\n",
    "                (left , top + height),\n",
    "                (left, top)\n",
    "                )\n",
    "            draw.line(points, fill=fill_red, width=line_width)\n",
    "\n",
    "    display(image)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "confidence=80 # confidence if the bodypart is covered by the particular PPE \n",
    "detect_ppe(imageName, confidence)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Recognize persons wearing PPEs in video\n",
    "\n",
    "- First we sample our video into individual image frames.\n",
    "- We can then call `detect_protective_equipment` to detect persons wearing PPEs in the image frame\n",
    "***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "videoName = \"media/ppe-detection/ppe-video.mov\"\n",
    "videoFile = tempFolder + \"ppe-video.mov\"\n",
    "\n",
    "s3.download_file(bucketName, videoName, videoFile)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Call Rekognition to start a job for object detection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start video label recognition job\n",
    "ppeLabels = []    \n",
    "cap = cv2.VideoCapture(videoFile)\n",
    "frameRate = cap.get(5) # frame rate\n",
    "while(cap.isOpened()):\n",
    "    frameId = cap.get(1) # current frame number\n",
    "    print(\"Processing frame id: {}\".format(frameId))\n",
    "    ret, frame = cap.read()\n",
    "    if (ret != True):\n",
    "        break\n",
    "    if (frameId % math.floor(frameRate) == 0):\n",
    "        hasFrame, imageBytes = cv2.imencode(\".jpg\", frame)\n",
    "\n",
    "        if(hasFrame):\n",
    "            response = rekognition. detect_protective_equipment(\n",
    "                Image={\n",
    "                    'Bytes': imageBytes.tobytes(),\n",
    "                }\n",
    "            )\n",
    "        \n",
    "        for person in response[\"Persons\"]:\n",
    "            person[\"Timestamp\"] = (frameId/frameRate)*1000\n",
    "            ppeLabels.append(person)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Show video in the player"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Show video in a player\n",
    "\n",
    "s3VideoUrl = s3.generate_presigned_url('get_object', Params={'Bucket': bucketName, 'Key': videoName})\n",
    "\n",
    "videoTag = \"<video controls='controls' autoplay width='640' height='360' name='Video' src='{0}'></video>\".format(s3VideoUrl)\n",
    "\n",
    "videoui = \"<table><tr><td style='vertical-align: top'>{}</td></tr></table>\".format(videoTag)\n",
    "\n",
    "display(HTML(videoui))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Review raw JSON reponse from Rekognition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Show JSON response returned by Rekognition PPE Detection API\n",
    "# In the JSON response below, you will see list of detected body parts and timestamp.\n",
    "# For each detected body part, you will see information like name of the body part and confidence\n",
    "# If PPE is detected, it will provide confidence if the body part is covered and bounding box\n",
    "\n",
    "display(ppeLabels)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Export the results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open(videoFile + \".json\", \"w\") as f:\n",
    "    f.write(json.dumps(ppeLabels)) \n",
    "\n",
    "cap.release()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Recognize persons wearing PPEs in a live stream\n",
    "***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This is a sample code to detect PPE in a live stream. \n",
    "# Before you run this code, you would need to set up an RTSP stream on a publicly accessible IP.\n",
    "\n",
    "\n",
    "\"\"\"\n",
    "import cv2\n",
    "import boto3\n",
    "import time\n",
    "from datetime import datetime\n",
    "import json\n",
    "\n",
    "def processFrame(videoStreamUrl):\n",
    "    cap = cv2.VideoCapture(videoStreamUrl)\n",
    "    ret, frame = cap.read()\n",
    "    if ret:\n",
    "        hasFrame, imageBytes = cv2.imencode(\".jpg\", frame)\n",
    "        if hasFrame:\n",
    "            session = boto3.session.Session()\n",
    "            rekognition = session.client('rekognition')\n",
    "            response = rekognition. detect_protective_equipment(\n",
    "                    Image={\n",
    "                        'Bytes': imageBytes.tobytes(),\n",
    "                    }\n",
    "                )\n",
    "            print(response)\n",
    "    cap.release()\n",
    "\n",
    "# Video stream\n",
    "videoStreamUrl = \"rtsp://@192.168.10.100\" # change this to your RTSP stream location\n",
    "frameCaptureThreshold = 300\n",
    "\n",
    "while (True):\n",
    "    try:\n",
    "        processFrame(videoStreamUrl)\n",
    "    except Exception as e:\n",
    "        print(\"Error: {}.\".format(e))\n",
    "\n",
    "    time.sleep(frameCaptureThreshold)\n",
    "\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***\n",
    "### References\n",
    "- https://docs.aws.amazon.com/rekognition/latest/APIReference/API_DetectProtectiveEquipment.html\n",
    "- https://docs.aws.amazon.com/rekognition/latest/dg/ppe-request-response.html\n",
    "\n",
    "***"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You have successfully used Amazon Rekognition to identify personal protective equipment in images and videos."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}