{ "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 }