{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Explicit, suggestive and violent content detection using Amazon Rekognition"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"***\n",
"This notebook provides a walkthrough of [content moderation APIs](https://docs.aws.amazon.com/rekognition/latest/dg/moderation.html) in Amazon Rekognition. You can quickly identify inappropriate content in your video and image libraries.\n",
"***"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Initialize stuff"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialise Notebook\n",
"import boto3\n",
"from IPython.display import Image as IImage, display\n",
"from IPython.display import HTML, display\n",
"from PIL import Image, ImageDraw, ImageFont\n",
"import time\n",
"import os"
]
},
{
"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-workshops-\" + awsRegion"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Content moderation in Images\n",
"***"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"imageName = \"content-moderation/media/weapon.png\""
]
},
{
"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 unsafe content in the image\n",
"# https://docs.aws.amazon.com/rekognition/latest/dg/moderation.html\n",
"\n",
"detectModerationLabelsResponse = rekognition.detect_moderation_labels(\n",
" Image={\n",
" 'S3Object': {\n",
" 'Bucket': bucketName,\n",
" 'Name': imageName,\n",
" }\n",
" }\n",
")"
]
},
{
"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": [
"#### Review the raw JSON reponse from Rekognition"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Show JSON response returned by Rekognition Moderation API\n",
"# In the JSON response below, you will see Moderation Labels, confidence score and additional information.\n",
"\n",
"display(detectModerationLabelsResponse)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Display list of detected moderation labels"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"for label in detectModerationLabelsResponse[\"ModerationLabels\"]:\n",
" print(\"- {} (Confidence: {})\".format(label[\"Name\"], label[\"Confidence\"]))\n",
" print(\" - Parent: {}\".format(label[\"ParentName\"]))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Content moderation in videos\n",
"\n",
"Content Moderation in video is an async operation. \n",
"https://docs.aws.amazon.com/rekognition/latest/dg/API_StartContentModeration.html\n",
" - We first start content moderation job which returns a Job Id.\n",
" - We can then call `get_content_moderation` to get the job status and after job is complete, we can get moderation results.\n",
" - In production use cases, you would usually use StepFucntion or SNS topic to get notified when job is complete.\n",
"***"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"videoName = \"content-moderation/media/weapon.mov\""
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Call Rekognition to start a job for content moderation"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Start content moderation job\n",
"startModerationLabelDetection = rekognition.start_content_moderation(\n",
" Video={\n",
" 'S3Object': {\n",
" 'Bucket': bucketName,\n",
" 'Name': videoName,\n",
" }\n",
" },\n",
")\n",
"\n",
"moderationJobId = startModerationLabelDetection['JobId']\n",
"display(\"Job Id: {0}\".format(moderationJobId))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Additional (Optional) Request Attributes\n",
"\n",
"ClientRequestTokenL\n",
"https://docs.aws.amazon.com/rekognition/latest/dg/API_StartContentModeration.html#rekognition-StartContentModeration-request-ClientRequestToken\n",
"\n",
"JobTag:\n",
"https://docs.aws.amazon.com/rekognition/latest/dg/API_StartContentModeration.html#rekognition-StartContentModeration-request-JobTag\n",
"\n",
"MinConfidence:\n",
"https://docs.aws.amazon.com/rekognition/latest/dg/API_StartContentModeration.html#rekognition-StartContentModeration-request-MinConfidence\n",
"\n",
"NotificationChannel:\n",
"https://docs.aws.amazon.com/rekognition/latest/dg/API_StartContentModeration.html#rekognition-StartContentModeration-request-NotificationChannel\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Wait for content moderation job to complete"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Wait for content moderation job to complete\n",
"# In production use cases, you would usually use StepFucntion or SNS topic to get notified when job is complete.\n",
"getContentModeration = rekognition.get_content_moderation(\n",
" JobId=moderationJobId,\n",
" SortBy='TIMESTAMP'\n",
")\n",
"\n",
"while(getContentModeration['JobStatus'] == 'IN_PROGRESS'):\n",
" time.sleep(5)\n",
" print('.', end='')\n",
" \n",
" getContentModeration = rekognition.get_content_moderation(\n",
" JobId=moderationJobId,\n",
" SortBy='TIMESTAMP')\n",
" \n",
"display(getContentModeration['JobStatus'])"
]
},
{
"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 Content Moderation API\n",
"# In the JSON response below, you will see list of detected content\n",
"# For each detected object, you will see information like Timestamp\n",
"\n",
"display(getContentModeration)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Display list of potentially unsafe content in the video"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"theObjects = {}\n",
"\n",
"strDetail = \"Moderation labels in video
=======================================
\"\n",
"strOverall = \"Moderation labels in the overall video:
=======================================
\"\n",
"\n",
"# Potentially unsafe detected in each frame\n",
"for obj in getContentModeration['ModerationLabels']:\n",
" ts = obj [\"Timestamp\"]\n",
" cconfidence = obj['ModerationLabel'][\"Confidence\"]\n",
" oname = obj['ModerationLabel'][\"Name\"]\n",
" strDetail = strDetail + \"At {} ms: {} (Confidence: {})
\".format(ts, oname, round(cconfidence,2))\n",
" if oname in theObjects:\n",
" cojb = theObjects[oname]\n",
" theObjects[oname] = {\"Name\" : oname, \"Count\": 1+cojb[\"Count\"]}\n",
" else:\n",
" theObjects[oname] = {\"Name\" : oname, \"Count\": 1}\n",
"\n",
"# Unique objects detected in video\n",
"for theObject in theObjects:\n",
" strOverall = strOverall + \"Name: {}, Count: {}
\".format(theObject, theObjects[theObject][\"Count\"])\n",
"\n",
"# Display results\n",
"display(HTML(strOverall))\n",
"#display(HTML(strDetail))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### Show video in the player"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"scrolled": true
},
"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": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"listui = \"\".format(strDetail)\n",
"display(HTML(listui))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"***\n",
"### References\n",
"- https://docs.aws.amazon.com/rekognition/latest/dg/API_DetectModerationLabels.html\n",
"- https://docs.aws.amazon.com/rekognition/latest/dg/API_StartContentModeration.html\n",
"- https://docs.aws.amazon.com/rekognition/latest/dg/API_GetContentModeration.html\n",
"\n",
"***"
]
}
],
"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.5"
}
},
"nbformat": 4,
"nbformat_minor": 2
}