Detecting Custom Labels in videos

Using DetectCustomLabels with captured frames

1. If you haven't already done so, install and configure the AWS CLI and the AWS SDKs. For more information, see Step 4: Set up the AWS CLI and AWS SDKs.

2. Make sure you have rekognition:DetectCustomLabels and AmazonS3ReadOnlyAccess permissions. For more information, see Step 4: Set up the AWS CLI and AWS SDKs.

3. Use the following example code. Change the value of videoFile to the name of a video file. Change the value of projectVersionArn to the Amazon Resource Name (ARN) of your Amazon Rekognition Custom Labels model.

   # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
   # SPDX-License-Identifier: Apache-2.0
   
   """
   Purpose
   Shows how to analyze a local video with an Amazon Rekognition Custom Labels model.
   """
   import argparse
   import logging
   import json
   import math
   import cv2
   import boto3
   
   from botocore.exceptions import ClientError
   
   logger = logging.getLogger(__name__)
   
   
   def analyze_video(rek_client, project_version_arn, video_file):
       """
       Analyzes a local video file with an Amazon Rekognition Custom Labels model.
       Creates a results JSON file based on the name of the supplied video file.
       :param rek_client: A Boto3 Amazon Rekognition client.
       :param project_version_arn: The ARN of the Custom Labels model that you want to use.
       :param video_file: The video file that you want to analyze.
       """
       
       custom_labels = []
       cap = cv2.VideoCapture(video_file)
       frame_rate = cap.get(5)  # Frame rate.
       while cap.isOpened():
           frame_id = cap.get(1)  # Current frame number.
           print(f"Processing frame id: {frame_id}")
           ret, frame = cap.read()
           if ret is not True:
               break
           if frame_id % math.floor(frame_rate) == 0:
               has_frame, image_bytes = cv2.imencode(".jpg", frame)
   
               if has_frame:
                   response = rek_client.detect_custom_labels(
                       Image={
                           'Bytes': image_bytes.tobytes(),
                       },
                       ProjectVersionArn=project_version_arn
                   )
   
               for elabel in response["CustomLabels"]:
                   elabel["Timestamp"] = (frame_id/frame_rate)*1000
                   custom_labels.append(elabel)
   
       print(custom_labels)
   
       with open(video_file + ".json", "w", encoding="utf-8") as f:
           f.write(json.dumps(custom_labels))
   
       cap.release()
   
   
   def add_arguments(parser):
       """
       Adds command line arguments to the parser.
       :param parser: The command line parser.
       """
   
       parser.add_argument(
           "project_version_arn", help="The ARN of the model that you want to use."
       )
   
       parser.add_argument(
           "video_file", help="The local path to the video that you want to analyze."
       )
   
   
   def main():
   
       logging.basicConfig(level=logging.INFO,
                           format="%(levelname)s: %(message)s")
   
       try:
           # Get command line arguments.
           parser = argparse.ArgumentParser(usage=argparse.SUPPRESS)
           add_arguments(parser)
           args = parser.parse_args()
   
           session = boto3.Session(profile_name='custom-labels-access')
           rekognition_client = session.client("rekognition")
   
           analyze_video(rekognition_client,
                        args.project_version_arn, args.video_file)
   
       except ClientError as err:
           print(f"Couldn't analyze video: {err}")
   
   
   if __name__ == "__main__":
       main()