{ "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 = \"\".format(s3VideoUrl)\n", "\n", "videoui = \"
{}
\".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 }