<jupyter_start><jupyter_text>Object and Scene Detection using Amazon Rekognition 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.<jupyter_code>import boto3
from IPython.display import HTML, display, Image as IImage
from PIL import Image, ImageDraw, ImageFont
import time
import os
import sagemaker
import boto3

sagemaker_session = sagemaker.Session()
role = sagemaker.get_execution_role()
bucket = sagemaker_session.default_bucket()
region = boto3.Session().region_name
rekognition = boto3.client('rekognition')
s3 = boto3.client('s3')
!mkdir -p ./tmp
temp_folder = 'tmp/'<jupyter_output><empty_output><jupyter_text>Detect Objects in Image<jupyter_code>imageName = 'content-moderation/media/cars.png'
display(IImage(url=s3.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': imageName})))<jupyter_output><empty_output><jupyter_text>Call Rekognition to Detect Objects in the Imagehttps://docs.aws.amazon.com/rekognition/latest/dg/API_DetectLabels.html<jupyter_code>detectLabelsResponse = rekognition.detect_labels(
    Image={
        'S3Object': {
            'Bucket': bucket,
            'Name': imageName,
        }
    }
)<jupyter_output><empty_output><jupyter_text>Review the Raw JSON Response from RekognitionShow JSON response returned by Rekognition Labels API (Object Detection).In the JSON response below, you will see Label, detected instances, confidence score and additional information.<jupyter_code>display(detectLabelsResponse)<jupyter_output><empty_output><jupyter_text>Show Bounding Boxes Around Recognized Objects<jupyter_code>def drawBoundingBoxes (sourceImage, boxes):
    # blue, green, red, grey
    colors = ((255,255,255),(255,255,255),(76,182,252),(52,194,123))
    
    # Download image locally
    imageLocation = temp_folder + os.path.basename(sourceImage)
    s3.download_file(bucket, sourceImage, imageLocation)

    # Draws BB on Image
    bbImage = Image.open(imageLocation)
    draw = ImageDraw.Draw(bbImage)
    width, height = bbImage.size
    col = 0
    maxcol = len(colors)
    line= 3
    for box in boxes:
        x1 = int(box[1]['Left'] * width)
        y1 = int(box[1]['Top'] * height)
        x2 = int(box[1]['Left'] * width + box[1]['Width'] * width)
        y2 = int(box[1]['Top'] * height + box[1]['Height']  * height)
        
        draw.text((x1,y1),box[0],colors[col])
        for l in range(line):
            draw.rectangle((x1-l,y1-l,x2+l,y2+l),outline=colors[col])
        col = (col+1)%maxcol
    
    imageFormat = "PNG"
    ext = sourceImage.lower()
    if(ext.endswith('jpg') or ext.endswith('jpeg')):
        imageFormat = 'JPEG'

    bbImage.save(imageLocation,format=imageFormat)

    display(bbImage)
boxes = []
objects = detectLabelsResponse['Labels']
for obj in objects:
    for einstance in obj["Instances"]:
        boxes.append ((obj['Name'], einstance['BoundingBox']))
    
drawBoundingBoxes(imageName, boxes)<jupyter_output><empty_output><jupyter_text>Display List of Detected Objects<jupyter_code>flaggedObjects = ["Car"]

for label in detectLabelsResponse["Labels"]:
    if(label["Name"] in flaggedObjects):
        print("Detected object:")
        print("- {} (Confidence: {})".format(label["Name"], label["Confidence"]))
        print("  - Parents: {}".format(label["Parents"]))<jupyter_output><empty_output><jupyter_text>Recognize Objects in Video Object recognition in video is an async operation. https://docs.aws.amazon.com/rekognition/latest/dg/API_StartLabelDetection.html. - First we start a label detection job which returns a Job Id.- We can then call `get_label_detection` to get the job status and after job is complete, we can get object metadata.- In production use cases, you would usually use StepFunction or SNS topic to get notified when job is complete.<jupyter_code>videoName = 'content-moderation/media/GrandTour720.mp4'

strDetail = 'Objects detected in video<br>=======================================<br>'
strOverall = 'Objects in the overall video:<br>=======================================<br>'
# Show video in a player

s3VideoUrl = s3.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': videoName})

videoTag = "<video controls='controls' autoplay width='640' height='360' name='Video' src='{0}'></video>".format(s3VideoUrl)

videoui = "<table><tr><td style='vertical-align: top'>{}</td></tr></table>".format(videoTag)

display(HTML(videoui))<jupyter_output><empty_output><jupyter_text>Call Rekognition to Start a Job for Object Detection Additional (Optional) Request AttributesClientRequestToken:https://docs.aws.amazon.com/rekognition/latest/dg/API_StartLabelDetection.htmlrekognition-StartLabelDetection-request-ClientRequestTokenJobTag:https://docs.aws.amazon.com/rekognition/latest/dg/API_StartLabelDetection.htmlrekognition-StartLabelDetection-request-JobTagMinConfidence:https://docs.aws.amazon.com/rekognition/latest/dg/API_StartLabelDetection.htmlrekognition-StartLabelDetection-request-MinConfidenceNotificationChannel:https://docs.aws.amazon.com/rekognition/latest/dg/API_StartLabelDetection.htmlrekognition-StartLabelDetection-request-NotificationChannel<jupyter_code># Start video label recognition job
startLabelDetection = rekognition.start_label_detection(
    Video={
        'S3Object': {
            'Bucket': bucket,
            'Name': videoName,
        }
    },
)

labelsJobId = startLabelDetection['JobId']
display("Job Id: {0}".format(labelsJobId))<jupyter_output><empty_output><jupyter_text>Wait for Object Detection Job to Complete<jupyter_code># Wait for object detection job to complete
# In production use cases, you would usually use StepFunction or SNS topic to get notified when job is complete.
getObjectDetection = rekognition.get_label_detection(
    JobId=labelsJobId,
    SortBy='TIMESTAMP'
)

while(getObjectDetection['JobStatus'] == 'IN_PROGRESS'):
    time.sleep(5)
    print('.', end='')
 
    getObjectDetection = rekognition.get_label_detection(
    JobId=labelsJobId,
    SortBy='TIMESTAMP')
    
display(getObjectDetection['JobStatus'])<jupyter_output><empty_output><jupyter_text>Review Raw JSON Response from Rekognition* Show JSON response returned by Rekognition Object Detection API.* In the JSON response below, you will see list of detected objects and activities.* For each detected object, you will see the `Timestamp` of the frame within the video.<jupyter_code>display(getObjectDetection)<jupyter_output><empty_output><jupyter_text>Display Recognized Objects in the VideoDisplay timestamps and objects detected at that time.<jupyter_code>flaggedObjectsInVideo = ["Car"]

theObjects = {}

# Objects detected in each frame
for obj in getObjectDetection['Labels']:
    ts = obj ["Timestamp"]
    cconfidence = obj['Label']["Confidence"]
    oname = obj['Label']["Name"]
    
    if(oname in flaggedObjectsInVideo):
        print("Found flagged object at {} ms: {} (Confidence: {})".format(ts, oname, round(cconfidence,2)))
    
    strDetail = strDetail + "At {} ms: {} (Confidence: {})<br>".format(ts, oname, round(cconfidence,2))
    if oname in theObjects:
        cojb = theObjects[oname]
        theObjects[oname] = {"Name" : oname, "Count": 1+cojb["Count"]}
    else:
        theObjects[oname] = {"Name" : oname, "Count": 1}

# Unique objects detected in video
for theObject in theObjects:
    strOverall = strOverall + "Name: {}, Count: {}<br>".format(theObject, theObjects[theObject]["Count"])

# Display results
display(HTML(strOverall))
listui = "<table><tr><td style='vertical-align: top'>{}</td></tr></table>".format(strDetail)
display(HTML(listui))<jupyter_output><empty_output><jupyter_text>Worker Safety with Amazon RekognitionYou 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.<jupyter_code>imageName = "content-moderation/media/hat-detection.png"
display(IImage(url=s3.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': imageName})))<jupyter_output><empty_output><jupyter_text>Call Amazon Rekognition to Detect Objects in the Image<jupyter_code>detectLabelsResponse = rekognition.detect_labels(
    Image={
        'S3Object': {
            'Bucket': bucket,
            'Name': imageName,
        }
    }
)<jupyter_output><empty_output><jupyter_text>Display Rekognition Response<jupyter_code>display(detectLabelsResponse)<jupyter_output><empty_output><jupyter_text>Show Bounding Boxes Around Recognized Objects<jupyter_code>def drawBoundingBoxes (sourceImage, boxes):
    # blue, green, red, grey
    colors = ((255,255,255),(255,255,255),(76,182,252),(52,194,123))
    
    # Download image locally
    imageLocation = temp_folder + os.path.basename(sourceImage)
    s3.download_file(bucket, sourceImage, imageLocation)

    # Draws BB on Image
    bbImage = Image.open(imageLocation)
    draw = ImageDraw.Draw(bbImage)
    width, height = bbImage.size
    col = 0
    maxcol = len(colors)
    line= 3
    for box in boxes:
        x1 = int(box[1]['Left'] * width)
        y1 = int(box[1]['Top'] * height)
        x2 = int(box[1]['Left'] * width + box[1]['Width'] * width)
        y2 = int(box[1]['Top'] * height + box[1]['Height']  * height)
        
        draw.text((x1,y1),box[0],colors[col])
        for l in range(line):
            draw.rectangle((x1-l,y1-l,x2+l,y2+l),outline=colors[col])
        col = (col+1)%maxcol
    
    imageFormat = "PNG"
    ext = sourceImage.lower()
    if(ext.endswith('jpg') or ext.endswith('jpeg')):
        imageFormat = 'JPEG'

    bbImage.save(imageLocation,format=imageFormat)

    display(bbImage)
boxes = []
objects = detectLabelsResponse['Labels']
for obj in objects:
    for einstance in obj["Instances"]:
        boxes.append ((obj['Name'], einstance['BoundingBox']))
    
drawBoundingBoxes(imageName, boxes)

def matchPersonsAndHats(personsList, hardhatsList):

    persons = []
    hardhats = []
    personsWithHats = []

    for person in personsList:
        persons.append(person)
    for hardhat in hardhatsList:
        hardhats.append(hardhat)

    h = 0
    matched = 0
    totalHats = len(hardhats)
    while(h < totalHats):
        hardhat = hardhats[h-matched]
        totalPersons = len(persons)
        p = 0
        while(p < totalPersons):
            person = persons[p]
            if(not (hardhat['BoundingBoxCoordinates']['x2'] < person['BoundingBoxCoordinates']['x1']
                or hardhat['BoundingBoxCoordinates']['x1'] > person['BoundingBoxCoordinates']['x2']
                or hardhat['BoundingBoxCoordinates']['y4'] < person['BoundingBoxCoordinates']['y1']
                    or hardhat['BoundingBoxCoordinates']['y1'] > person['BoundingBoxCoordinates']['y4']
                )):

                personsWithHats.append({'Person' : person, 'Hardhat' : hardhat})

                del persons[p]
                del hardhats[h - matched]

                matched = matched + 1

                break
            p = p + 1
        h = h + 1

    return (personsWithHats, persons, hardhats)

def getBoundingBoxCoordinates(boundingBox, imageWidth, imageHeight):
    x1 = 0
    y1 = 0
    x2 = 0
    y2 = 0
    x3 = 0
    y3 = 0
    x4 = 0
    y4 = 0

    boxWidth = boundingBox['Width']*imageWidth
    boxHeight = boundingBox['Height']*imageHeight

    x1 = boundingBox['Left']*imageWidth
    y1 = boundingBox['Top']*imageWidth

    x2 = x1 + boxWidth
    y2 = y1

    x3 = x2
    y3 = y1 + boxHeight

    x4 = x1
    y4 = y3

    return({'x1': x1, 'y1' : y1, 'x2' : x2, 'y2' : y2, 'x3' : x3, 'y3' : y3, 'x4' : x4, 'y4' : y4})

def getPersonsAndHardhats(labelsResponse, imageWidth, imageHeight):

    persons = []
    hardhats = []

    for label in labelsResponse['Labels']:
        if label['Name'] == 'Person' and 'Instances' in label:
            for person in label['Instances']:
                    persons.append({'BoundingBox' : person['BoundingBox'], 'BoundingBoxCoordinates' : getBoundingBoxCoordinates(person['BoundingBox'], imageWidth, imageHeight), 'Confidence' : person['Confidence']})
        elif ((label['Name'] == 'Hardhat' or label['Name'] == 'Helmet') and 'Instances' in label):
            for hardhat in label['Instances']:
                hardhats.append({'BoundingBox' : hardhat['BoundingBox'], 'BoundingBoxCoordinates' : getBoundingBoxCoordinates(hardhat['BoundingBox'], imageWidth, imageHeight), 'Confidence' : hardhat['Confidence']})

    return (persons, hardhats)
s3Resource = boto3.resource('s3')
bucket = s3Resource.Bucket(bucket)
iojb = bucket.Object(imageName)
response = iojb.get()
file_stream = response['Body']
im = Image.open(file_stream)
imageWidth, imageHeight = im.size
persons, hardhats = getPersonsAndHardhats(detectLabelsResponse, imageWidth, imageHeight)

personsWithHats, personsWithoutHats, hatsWihoutPerson = matchPersonsAndHats(persons, hardhats)

personsWithHatsCount = len(personsWithHats)
personsWithoutHatsCount = len(personsWithoutHats)
hatsWihoutPersonCount = len(hatsWihoutPerson)

outputMessage = "Person(s): {}".format(personsWithHatsCount+personsWithoutHatsCount)
outputMessage = outputMessage + "\nPerson(s) With Safety Hat: {}\nPerson(s) Without Safety Hat: {}".format(personsWithHatsCount, personsWithoutHatsCount)
print(outputMessage)<jupyter_output><empty_output><jupyter_text>Congratulations!You have successfully used Amazon Rekognition to identify specific objects in images and videos.  References- https://docs.aws.amazon.com/rekognition/latest/dg/API_DetectLabels.html- https://docs.aws.amazon.com/rekognition/latest/dg/API_StartLabelDetection.html- https://docs.aws.amazon.com/rekognition/latest/dg/API_GetLabelDetection.html  Release Resources<jupyter_code>%%html

<p><b>Shutting down your kernel for this notebook to release resources.</b></p>
<button class="sm-command-button" data-commandlinker-command="kernelmenu:shutdown" style="display:none;">Shutdown Kernel</button>
        
<script>
try {
    els = document.getElementsByClassName("sm-command-button");
    els[0].click();
}
catch(err) {
    // NoOp
}    
</script>
%%javascript

try {
    Jupyter.notebook.save_checkpoint();
    Jupyter.notebook.session.delete();
}
catch(err) {
    // NoOp
}<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup