{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "64907692",
   "metadata": {},
   "source": [
    "# 3.2 Fun with Rekognition DetectFaces and SearchFacesByImage\n",
    "----\n",
    "This is fun but optional lab. This lab shows how you can index faces into a collection then use detect faces and search faces by image to identify several faces found in a single image. To do this we'll create a collection, index several face images into the collection. Then we'll search the faces found in a single image against the collection of known faces in order to identify the faces in the image. \n",
    "\n",
    "## Steps \n",
    "\n",
    "1. Load packages \n",
    "2. View existing collections \n",
    "3. Create a new collection \n",
    "4. Index faces into the collection \n",
    "5. Search the collection to find and present faces found in an image\n",
    "6. Clean up\n",
    "\n",
    "This  notebook will guide you through on how to compare all faces detected in an image against your Amazon Rekognition Face Collection. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ec7fb7d7",
   "metadata": {},
   "source": [
    "## Step 1. Load Libraries "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ed871982",
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3, os, io\n",
    "client=boto3.client('rekognition')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ce603810",
   "metadata": {},
   "source": [
    "## Step 2. View your existing collections"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ca7475f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "def list_collections():\n",
    "\n",
    "    max_results=10\n",
    "    \n",
    "    print('Displaying collections...')\n",
    "    response=client.list_collections(MaxResults=max_results)\n",
    "    collection_count=0\n",
    "    done=False\n",
    "    \n",
    "    while not done:\n",
    "        collections=response['CollectionIds']\n",
    "\n",
    "        for collection in collections:\n",
    "            print (collection)\n",
    "            collection_count+=1\n",
    "        if 'NextToken' in response:\n",
    "            nextToken=response['NextToken']\n",
    "            response=client.list_collections(NextToken=nextToken,MaxResults=max_results)\n",
    "            \n",
    "        else:\n",
    "            done=True\n",
    "\n",
    "    return collection_count   \n",
    "\n",
    "collection_count=list_collections()\n",
    "\n",
    "print(\"There are: {} collections in your account \".format(collection_count))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fce788a7",
   "metadata": {},
   "source": [
    "## Step 3. Create a new collection\n",
    "-----\n",
    "\n",
    "Remember you must use a unique name if you are creating a new collection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8c27a89b",
   "metadata": {},
   "outputs": [],
   "source": [
    "collection_id=' ' # name your collection "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "25560fe4",
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_collection(collection_id):\n",
    "    #Create a collection\n",
    "    print('Creating collection:' + collection_id)\n",
    "    try:\n",
    "        response=client.create_collection(CollectionId=collection_id)\n",
    "    except:\n",
    "        client.delete_collection(CollectionId=collection_id)\n",
    "        response=client.create_collection(CollectionId=collection_id)\n",
    "    print('Collection ARN: ' + response['CollectionArn'])\n",
    "    print('Status code: ' + str(response['StatusCode']))\n",
    "    print('Done.')\n",
    "    \n",
    "create_collection(collection_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1a22c079",
   "metadata": {},
   "source": [
    "### Step 3a. Confirm your collection creation. \n",
    "-----\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2b5e0bab",
   "metadata": {},
   "outputs": [],
   "source": [
    "collection_count=list_collections()\n",
    "print(\"collections: \" + str(collection_count))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "42199e56",
   "metadata": {},
   "source": [
    "## Step 4. Index faces (add faces to a collection) \n",
    "-----\n",
    "Here we are going to iterate over the files in the populate folder and index their faces. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "403168c7",
   "metadata": {},
   "outputs": [],
   "source": [
    "directory = 'media/populate'\n",
    " \n",
    "# iterate over files the populate directory\n",
    "for filename in os.listdir(directory):\n",
    "    f = os.path.join(directory, filename)\n",
    "    # checking if it is a file\n",
    "    if os.path.isfile(f):\n",
    "        print(f)\n",
    "        file = open(f, \"rb\") # opening for [r]eading as [b]inary\n",
    "        data = file.read() \n",
    "        response=client.index_faces(CollectionId=collection_id,\n",
    "                            Image={'Bytes':data},\n",
    "                            ExternalImageId=f.split(\"/\")[2],\n",
    "                            MaxFaces=1,\n",
    "                            QualityFilter=\"AUTO\",\n",
    "                            DetectionAttributes=['ALL'])\n",
    "        print ('Results for ' + f.split(\"/\")[2])\n",
    "        print('Faces indexed:')\n",
    "        for faceRecord in response['FaceRecords']:\n",
    "            print('  Face ID : {}'.format( faceRecord['Face']['FaceId']))\n",
    "            print('  Location: {}'.format(faceRecord['Face']['BoundingBox']))\n",
    "        \n",
    "        if len(response['UnindexedFaces']) > 0:\n",
    "            print('Faces not indexed:')\n",
    "            for unindexedFace in response['UnindexedFaces']:\n",
    "                print(' Location: {}'.format(unindexedFace['FaceDetail']['BoundingBox']))\n",
    "                print(' Reasons :')\n",
    "                for reason in unindexedFace['Reasons']:\n",
    "                    print('   ' + reason)\n",
    "        file.close()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2a5924ea",
   "metadata": {},
   "source": [
    "### 4a. List faces in the collection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "45c9428c",
   "metadata": {},
   "outputs": [],
   "source": [
    "def list_faces_in_collection(collection_id):\n",
    "\n",
    "    maxResults=20\n",
    "    faces_count=0\n",
    "    tokens=True\n",
    "\n",
    "    response=client.list_faces(CollectionId=collection_id,\n",
    "                               MaxResults=maxResults)\n",
    "\n",
    "    print('Faces in collection: {}'.format( collection_id))\n",
    " \n",
    "    while tokens:\n",
    "\n",
    "        faces=response['Faces']\n",
    "\n",
    "        for face in faces:\n",
    "            print(\"   FaceID: {}, ExternalImageId: {}\".format(face[\"FaceId\"],face[\"ExternalImageId\"].split('.')[0]))\n",
    "            faces_count+=1\n",
    "        if 'NextToken' in response:\n",
    "            nextToken=response['NextToken']\n",
    "            response=client.list_faces(CollectionId=collection_id,\n",
    "                                       NextToken=nextToken,MaxResults=maxResults)\n",
    "        else:\n",
    "            tokens=False\n",
    "    return faces_count   \n",
    "\n",
    "faces_count=list_faces_in_collection(collection_id)\n",
    "print(\"Number of faces in collection: {}\".format(faces_count))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "12a91336",
   "metadata": {},
   "source": [
    "## Step 5. Find faces in photo\n",
    "\n",
    "----\n",
    "Here we create a few functions that will be useful for transforming, detecting and extracting faces "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "40b4fdd9",
   "metadata": {},
   "outputs": [],
   "source": [
    "def transform_bounding(img, box):\n",
    "    imgWidth, imgHeight = img.size\n",
    "    l = (imgWidth * box['Left'])-5\n",
    "    t = (imgHeight * box['Top'])-5\n",
    "    w = (imgWidth * box['Width'])+10\n",
    "    h = (imgHeight * box['Height'])+10\n",
    "    return l,t,w,h"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f7b18bf3",
   "metadata": {},
   "outputs": [],
   "source": [
    "def detect_faces(file):\n",
    "    faces = []\n",
    "    f = open(file, \"rb\") # opening for [r]eading as [b]inary\n",
    "    data = f.read() \n",
    "    response = client.detect_faces(Image={'Bytes':data})\n",
    "    for face in response[\"FaceDetails\"]:\n",
    "        faces.append(face['BoundingBox'])\n",
    "    print(\"Faces detected: \" + str(len(response['FaceDetails'])))   \n",
    "    return faces"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ece418cd",
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "directory = 'media/test'\n",
    "\n",
    "from PIL import Image               # to load images\n",
    "from IPython.display import display # to display images\n",
    "\n",
    "for filename in os.listdir(directory):\n",
    "    f = os.path.join(directory, filename)\n",
    "    # checking if it is a file\n",
    "    if os.path.isfile(f):\n",
    "        img = Image.open(f)\n",
    "        display(img)\n",
    "        \n",
    "        faces = detect_faces(f)        \n",
    "        for face in faces:\n",
    "            l,t,w,h = transform_bounding(img,face)\n",
    "            cropped = img.crop((l,t,l+w,t+h))    \n",
    "\n",
    "            stream = io.BytesIO()\n",
    "            cropped.save(stream, format='PNG')\n",
    "            bin_img = stream.getvalue()\n",
    "\n",
    "            response0 = client.detect_faces(\n",
    "                Image={'Bytes': bin_img},\n",
    "            )\n",
    "\n",
    "            if len(response0['FaceDetails']) > 0:\n",
    "                print(\"face found\")\n",
    "                display(cropped)\n",
    "                response1=client.search_faces_by_image(CollectionId=collection_id,\n",
    "                                                  Image={'Bytes': bin_img},\n",
    "                                                  FaceMatchThreshold=50)\n",
    "                faceMatches=response1['FaceMatches']\n",
    "                if(len(faceMatches) > 0):\n",
    "                    for match in faceMatches:\n",
    "                        print ('Person     : ' + match['Face']['ExternalImageId'].split('.')[0])\n",
    "                        print ('Similarity : ' + \"{:.2f}\".format(match['Similarity']) + \"%\")\n",
    "                else:\n",
    "                    print(\"but no match found\")\n",
    "            else:\n",
    "                print(\"face not found in the following crop\")\n",
    "                cropped.show()\n",
    "\n",
    "            print(\"------------------------------\")  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "36915da3",
   "metadata": {},
   "source": [
    "## Clean up the resources"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "86cf8c7e",
   "metadata": {},
   "source": [
    "Delete your face collection, this will delete the collection and the face vectors stored in the collection."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c008ce84",
   "metadata": {},
   "outputs": [],
   "source": [
    "def delete_collection(collection_id):\n",
    "\n",
    "    print('Attempting to delete collection ' + collection_id)\n",
    "    status_code=0\n",
    "    try:\n",
    "        response=client.delete_collection(CollectionId=collection_id)\n",
    "        status_code=response['StatusCode']\n",
    "        \n",
    "    except ClientError as e:\n",
    "        if e.response['Error']['Code'] == 'ResourceNotFoundException':\n",
    "            print ('The collection ' + collection_id + ' was not found ')\n",
    "        else:\n",
    "            print ('Error other than Not Found occurred: ' + e.response['Error']['Message'])\n",
    "        status_code=e.response['ResponseMetadata']['HTTPStatusCode']\n",
    "    return(status_code)\n",
    "\n",
    "\n",
    "status_code=delete_collection(collection_id)\n",
    "print('Status code: ' + str(status_code))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e755e2f0",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}