1-rekognition-image-apis.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Object and scene detection using Amazon Rekognition"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***\n",
    "This notebook provides a walkthrough of [object detection API](https://docs.aws.amazon.com/rekognition/latest/dg/labels.html) in Amazon Rekognition to identify objects.\n",
    "***"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initialization"
   ]
  },
  {
   "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, ImageFont\n",
    "import time\n",
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client = boto3.client(\"sts\")\n",
    "account = client.get_caller_identity()[\"Account\"]\n",
    "print(f\"AWS account:{account}\")\n",
    "\n",
    "session = boto3.session.Session()\n",
    "region = session.region_name\n",
    "print(f\"AWS region:{region}\")"
   ]
  },
  {
   "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 will be the location from where we will read images/videos\n",
    "\n",
    "bucket = 'dml102-20221003-sagemaker-bucket-{region}-{account}'.format(region = region, account = account)"
   ]
  },
  {
   "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 objects in image\n",
    "***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "imageName = \"img/rekognition-apis/cars.jpg\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!aws s3 cp $imageName s3://$bucket/$imageName"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "display(IImage(url=s3.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': imageName})))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Call Rekognition to detect objects in the image"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Call Amazon Rekognition to detect objects in the image\n",
    "# https://docs.aws.amazon.com/rekognition/latest/dg/API_DetectLabels.html\n",
    "\n",
    "detectLabelsResponse = rekognition.detect_labels(\n",
    "    Image={\n",
    "        'S3Object': {\n",
    "            'Bucket': bucket,\n",
    "            'Name': imageName,\n",
    "        }\n",
    "    }\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Review the raw JSON reponse from Rekognition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Show JSON response returned by Rekognition Labels API (Object Detection)\n",
    "# In the JSON response below, you will see Label, detected instances, confidence score and additional information.\n",
    "\n",
    "display(detectLabelsResponse)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a function that will display image with bounded boxes around recognized celebrites\n",
    "# We will call this function in next step\n",
    "  \n",
    "def drawBoundingBoxes (sourceImage, boxes):\n",
    "    # blue, green, red, grey\n",
    "    colors = ((255,255,255),(255,255,255),(76,182,252),(52,194,123))\n",
    "    \n",
    "    # Download image locally\n",
    "    imageLocation = sourceImage\n",
    "    s3.download_file(bucket, sourceImage, imageLocation)\n",
    "\n",
    "    # Draws BB on Image\n",
    "    bbImage = Image.open(imageLocation)\n",
    "    draw = ImageDraw.Draw(bbImage)\n",
    "    width, height = bbImage.size\n",
    "    col = 0\n",
    "    maxcol = len(colors)\n",
    "    line= 3\n",
    "    for box in boxes:\n",
    "        x1 = int(box[1]['Left'] * width)\n",
    "        y1 = int(box[1]['Top'] * height)\n",
    "        x2 = int(box[1]['Left'] * width + box[1]['Width'] * width)\n",
    "        y2 = int(box[1]['Top'] * height + box[1]['Height']  * height)\n",
    "        \n",
    "        draw.text((x1,y1),box[0],colors[col])\n",
    "        for l in range(line):\n",
    "            draw.rectangle((x1-l,y1-l,x2+l,y2+l),outline=colors[col])\n",
    "        col = (col+1)%maxcol\n",
    "    \n",
    "    imageFormat = \"JPG\"\n",
    "    ext = sourceImage.lower()\n",
    "    if(ext.endswith('jpg') or ext.endswith('jpeg')):\n",
    "        imageFormat = 'JPEG'\n",
    "\n",
    "#     bbImage.save(imageLocation,format=imageFormat)\n",
    "\n",
    "    display(bbImage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Extract bounding box information from JSON response above and display image with bounding boxes around celebrites.\n",
    "\n",
    "boxes = []\n",
    "objects = detectLabelsResponse['Labels']\n",
    "for obj in objects:\n",
    "    for instance in obj[\"Instances\"]:\n",
    "        boxes.append((obj['Name'], instance['BoundingBox']))\n",
    "    \n",
    "drawBoundingBoxes(imageName, boxes)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Worker Safety with Amazon Rekognition\n",
    "***\n",
    "You can use Amazon Rekognition to detect if certain objects are not present in the image or video. For example you can perform worker safety audit by revieweing images/video of a construction site  and detecting if there are any workers without safety hat."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "imageName = \"img/rekognition-apis/hat-detection.jpg\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!aws s3 cp $imageName s3://$bucket/$imageName"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "display(IImage(url=s3.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': imageName})))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Call Amazon Rekognition to detect objects in the image\n",
    "\n",
    "detectLabelsResponse = rekognition.detect_labels(\n",
    "    Image={\n",
    "        'S3Object': {\n",
    "            'Bucket': bucket,\n",
    "            'Name': imageName,\n",
    "        }\n",
    "    }\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Display Rekognition response\n",
    "display(detectLabelsResponse)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a function that will display image with bounded boxes around recognized objects\n",
    "# We will call this function in next step\n",
    "  \n",
    "def drawBoundingBoxes (sourceImage, boxes):\n",
    "    # blue, green, red, grey\n",
    "    colors = ((255,255,255),(255,255,255),(76,182,252),(52,194,123))\n",
    "    \n",
    "    # Download image locally\n",
    "    imageLocation = tempFolder+os.path.basename(sourceImage)\n",
    "    s3.download_file(bucket, sourceImage, imageLocation)\n",
    "\n",
    "    # Draws BB on Image\n",
    "    bbImage = Image.open(imageLocation)\n",
    "    draw = ImageDraw.Draw(bbImage)\n",
    "    width, height = bbImage.size\n",
    "    col = 0\n",
    "    maxcol = len(colors)\n",
    "    line= 3\n",
    "    for box in boxes:\n",
    "        x1 = int(box[1]['Left'] * width)\n",
    "        y1 = int(box[1]['Top'] * height)\n",
    "        x2 = int(box[1]['Left'] * width + box[1]['Width'] * width)\n",
    "        y2 = int(box[1]['Top'] * height + box[1]['Height']  * height)\n",
    "        \n",
    "        draw.text((x1,y1),box[0],colors[col])\n",
    "        for l in range(line):\n",
    "            draw.rectangle((x1-l,y1-l,x2+l,y2+l),outline=colors[col])\n",
    "        col = (col+1)%maxcol\n",
    "    \n",
    "    imageFormat = \"PNG\"\n",
    "    ext = sourceImage.lower()\n",
    "    if(ext.endswith('jpg') or ext.endswith('jpeg')):\n",
    "        imageFormat = 'JPEG'\n",
    "\n",
    "    bbImage.save(imageLocation,format=imageFormat)\n",
    "\n",
    "    display(bbImage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Show image and bounded boxes around detected objects\n",
    "boxes = []\n",
    "objects = detectLabelsResponse['Labels']\n",
    "for obj in objects:\n",
    "    for einstance in obj[\"Instances\"]:\n",
    "        boxes.append ((obj['Name'], einstance['BoundingBox']))\n",
    "    \n",
    "drawBoundingBoxes(imageName, boxes)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def matchPersonsAndHats(personsList, hardhatsList):\n",
    "\n",
    "    persons = []\n",
    "    hardhats = []\n",
    "    personsWithHats = []\n",
    "\n",
    "    for person in personsList:\n",
    "        persons.append(person)\n",
    "    for hardhat in hardhatsList:\n",
    "        hardhats.append(hardhat)\n",
    "\n",
    "    h = 0\n",
    "    matched = 0\n",
    "    totalHats = len(hardhats)\n",
    "    while(h < totalHats):\n",
    "        hardhat = hardhats[h-matched]\n",
    "        totalPersons = len(persons)\n",
    "        p = 0\n",
    "        while(p < totalPersons):\n",
    "            person = persons[p]\n",
    "            if(not (hardhat['BoundingBoxCoordinates']['x2'] < person['BoundingBoxCoordinates']['x1']\n",
    "                or hardhat['BoundingBoxCoordinates']['x1'] > person['BoundingBoxCoordinates']['x2']\n",
    "                or hardhat['BoundingBoxCoordinates']['y4'] < person['BoundingBoxCoordinates']['y1']\n",
    "                    or hardhat['BoundingBoxCoordinates']['y1'] > person['BoundingBoxCoordinates']['y4']\n",
    "                )):\n",
    "\n",
    "                personsWithHats.append({'Person' : person, 'Hardhat' : hardhat})\n",
    "\n",
    "                del persons[p]\n",
    "                del hardhats[h - matched]\n",
    "\n",
    "                matched = matched + 1\n",
    "\n",
    "                break\n",
    "            p = p + 1\n",
    "        h = h + 1\n",
    "\n",
    "    return (personsWithHats, persons, hardhats)\n",
    "\n",
    "def getBoundingBoxCoordinates(boundingBox, imageWidth, imageHeight):\n",
    "    x1 = 0\n",
    "    y1 = 0\n",
    "    x2 = 0\n",
    "    y2 = 0\n",
    "    x3 = 0\n",
    "    y3 = 0\n",
    "    x4 = 0\n",
    "    y4 = 0\n",
    "\n",
    "    boxWidth = boundingBox['Width']*imageWidth\n",
    "    boxHeight = boundingBox['Height']*imageHeight\n",
    "\n",
    "    x1 = boundingBox['Left']*imageWidth\n",
    "    y1 = boundingBox['Top']*imageWidth\n",
    "\n",
    "    x2 = x1 + boxWidth\n",
    "    y2 = y1\n",
    "\n",
    "    x3 = x2\n",
    "    y3 = y1 + boxHeight\n",
    "\n",
    "    x4 = x1\n",
    "    y4 = y3\n",
    "\n",
    "    return({'x1': x1, 'y1' : y1, 'x2' : x2, 'y2' : y2, 'x3' : x3, 'y3' : y3, 'x4' : x4, 'y4' : y4})\n",
    "\n",
    "def getPersonsAndHardhats(labelsResponse, imageWidth, imageHeight):\n",
    "\n",
    "    persons = []\n",
    "    hardhats = []\n",
    "\n",
    "    for label in labelsResponse['Labels']:\n",
    "        if label['Name'] == 'Person' and 'Instances' in label:\n",
    "            for person in label['Instances']:\n",
    "                    persons.append({'BoundingBox' : person['BoundingBox'], 'BoundingBoxCoordinates' : getBoundingBoxCoordinates(person['BoundingBox'], imageWidth, imageHeight), 'Confidence' : person['Confidence']})\n",
    "        elif ((label['Name'] == 'Hardhat' or label['Name'] == 'Helmet') and 'Instances' in label):\n",
    "            for hardhat in label['Instances']:\n",
    "                hardhats.append({'BoundingBox' : hardhat['BoundingBox'], 'BoundingBoxCoordinates' : getBoundingBoxCoordinates(hardhat['BoundingBox'], imageWidth, imageHeight), 'Confidence' : hardhat['Confidence']})\n",
    "\n",
    "    return (persons, hardhats)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3Resource = boto3.resource('s3')\n",
    "bucket = s3Resource.Bucket(bucket)\n",
    "iojb = bucket.Object(imageName)\n",
    "response = iojb.get()\n",
    "file_stream = response['Body']\n",
    "im = Image.open(file_stream)\n",
    "imageWidth, imageHeight = im.size"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "persons, hardhats = getPersonsAndHardhats(detectLabelsResponse, imageWidth, imageHeight)\n",
    "\n",
    "personsWithHats, personsWithoutHats, hatsWihoutPerson = matchPersonsAndHats(persons, hardhats)\n",
    "\n",
    "personsWithHatsCount = len(personsWithHats)\n",
    "personsWithoutHatsCount = len(personsWithoutHats)\n",
    "hatsWihoutPersonCount = len(hatsWihoutPerson)\n",
    "\n",
    "outputMessage = \"Person(s): {}\".format(personsWithHatsCount+personsWithoutHatsCount)\n",
    "outputMessage = outputMessage + \"\\nPerson(s) With Safety Hat: {}\\nPerson(s) Without Safety Hat: {}\".format(personsWithHatsCount, personsWithoutHatsCount)\n",
    "print(outputMessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}