realtime-extraction-analysis.yaml

AWSTemplateFormatVersion: '2010-09-09'
Parameters:
  RTSPURL:
    Type: String
    Description: RTSP URL for Lambda SnapshotGenerator 
    Default: 'rtsp://admin:DEVICE_VERIFICATION_CODE@YOUR_PUBLIC_IP:554/H.264'   
  S3BucketName:
    Type: String
    Description: S3 bucket name for Lambda SnapshotGenerator and ImageProcessor
    Default: 'snapshot-generator-visitor'   
  DynamoDBTableName:
    Type: String
    Description: DynamoDB table name
    Default: 'snapshot-generator-visitor'
  EventBridgeTime:
    Type: String
    Description: Scheduled time for Lambda ReportSender in CRON format 
    Default: cron(0 02 ? * * *)
  SNSTopicName:
    Type: String
    Description: SNS topic name for Lambda ReportSender
    Default: 'snapshot-generator-visitor'
  APISecret:
    Type: String
    Description: API Secret to Invoke Lambda Function URL for SnapshotGenerator
    Default: 'mysupersecretstring'
  EmailAddress:
    Type: String
    Description: Email address for SNS subscription
    Default: 'youremailaddressfornotifications@gmail.com'

Resources:

  S3Bucket:
    Type: AWS::S3::Bucket
    DependsOn:
      - ImageProcessor
    DeletionPolicy: Retain
    Properties:
      BucketName: !Ref S3BucketName
      NotificationConfiguration:
        LambdaConfigurations:
          - Event: s3:ObjectCreated:*
            Function: !GetAtt ImageProcessor.Arn
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: visitors/

  SnapshotGenerator:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: SnapshotGenerator
      Runtime: python3.9
      Handler: index.lambda_handler
      Code:
        ZipFile: |
          from datetime import datetime, timedelta
          import boto3
          import json
          import subprocess
          import os

          def convert_to_est(utc_time):
              est_offset = timedelta(hours=-4)  # Eastern Standard Time (EST) offset is -5 hours
              est_time = utc_time + est_offset
              return est_time
              
          def generate_snapshot(url):
              print(f'Trying to connect to {url}')
              FFMPEG_BINARY_PATH = "/opt/bin/ffmpeg"
              S3_BUCKET = os.environ.get("S3_BUCKET")

              # Convert the UTC time to Eastern Standard Time (EST)
              now_est = convert_to_est(datetime.now())

              # Format the timestamp as desired
              #timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
              timestamp = now_est.strftime("%Y%m%d-%H%M%S")

              filename = f"image-{timestamp}.jpg"

              command = [
                      FFMPEG_BINARY_PATH,
                      "-i", url,
                      "-vf", "fps=1/10,scale=640:480",
                      "-q:v", "2",
                      "-vsync", "0",
                      "-update", "1",
                      "-frames", "1",
                      "/tmp/" + filename
                      ]

              subprocess.run(command)

              s3_key = f"visitors/{now_est.strftime('%Y%m%d')}/{filename}"

              s3_client = boto3.client("s3")
              s3_client.upload_file("/tmp/" + filename, S3_BUCKET, s3_key)

              # Generate the pre-signed URL with a validity period of 1 hour
              expiration = datetime.now() + timedelta(hours=1)
              #presigned_url = s3_client.generate_presigned_url(
              #    "get_object",
              #    Params={"Bucket": S3_BUCKET, "Key": s3_key},
              #    ExpiresIn=3600,
              #)

              presigned_url ='s3://' + S3_BUCKET + '/'+ s3_key

              return presigned_url, expiration


          def lambda_handler(event, context):

              try:
                  secret = json.loads(event['body'])['secret']
              except (KeyError, TypeError):
                  try:
                      secret = event['secret']
                  except (KeyError, TypeError):
                      secret = None
                      
              expected_secret = os.environ.get("SECRET")  # Retrieve expected secret from environment variable

              # Check if the provided secret matches the expected secret
              if secret != expected_secret:
                  return {
                      "statusCode": 401,
                      "body": json.dumps({"message": f"Secret does not match"}),
                  }

              RTSP_URL = os.environ.get("RTSP_URL")

              try:
                  snapshot_url, expiration = generate_snapshot(RTSP_URL)
              except Exception as error:
                  return {
                      "statusCode": 500,
                      "body": json.dumps({"message": "Failed to create snapshot"}),
                  }

              return {
                  "statusCode": 200,
                  "body": json.dumps({
                      "message": "Streaming snapshot created",
                      "snapshot_url": snapshot_url,
                      "expiration": expiration.isoformat()
                  }),
              }
      Environment:
        Variables:
          RTSP_URL: !Ref RTSPURL
          S3_BUCKET: !Ref S3BucketName
          SECRET: !Ref APISecret
      Timeout: 60
      MemorySize: 300
      Role: !GetAtt ExecutionRole.Arn
      Layers:
        - !ImportValue ffmpegLayerArn

  SnapshotGeneratorPermissions:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref SnapshotGenerator
      Action: lambda:InvokeFunctionUrl
      Principal: "*"
      FunctionUrlAuthType: NONE

  SnapshotGeneratorFunctionUrl:
    Type: AWS::Lambda::Url
    Properties:
      TargetFunctionArn: !Ref SnapshotGenerator
      AuthType: NONE

  ImageProcessor:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: ImageProcessor
      Runtime: python3.9
      Handler: index.lambda_handler
      Code:
        ZipFile: |
          from datetime import datetime, timedelta
          import json
          import boto3
          from decimal import Decimal
          import os

          def convert_to_est(utc_time):
              est_offset = timedelta(hours=-4)  # Eastern Standard Time (EST) offset is -5 hours
              est_time = utc_time + est_offset
              return est_time

          def generate_presigned_url(bucket, key, expiration=86400):
              s3_client = boto3.client('s3')
              presigned_url = s3_client.generate_presigned_url(
                  'get_object',
                  Params={'Bucket': bucket, 'Key': key},
                  ExpiresIn=expiration
              )
              return presigned_url 

          def lambda_handler(event, context):
              # Extract the bucket and key from the S3 event
              s3_event = event['Records'][0]['s3']
              bucket = s3_event['bucket']['name']
              key = s3_event['object']['key']
              
              # Detect labels using Amazon Rekognition
              rekognition_client = boto3.client('rekognition')
              response = rekognition_client.detect_labels(
                  Image={'S3Object': {'Bucket': bucket, 'Name': key}},
                  MaxLabels=10,
                  MinConfidence=99
              )
              
              # Extract labels from the Rekognition response
              labels = response['Labels']
              
              # Generate a pre-signed URL for the image
              presigned_url = generate_presigned_url(bucket, key)
              
              # Create a list to hold the labels and confidence as merged items
              merged_labels = []
              
              for label in labels:
                  merged_label = f"{label['Name']} (Confidence: {label['Confidence']})"
                  merged_labels.append(merged_label)
              
              # Save the merged labels to DynamoDB
              dynamodb = boto3.resource('dynamodb')
              table_name =  os.environ.get("DYNAMODB_TABLE")
              table = dynamodb.Table(table_name)
              
              # Set TTL to expire after a week (7 days)
              expiration_time = datetime.now() + timedelta(days=7)
              
              item = {
                  'timestamp': int(convert_to_est(datetime.now()).timestamp()),
                  'image_name': key.split('/')[-1],
                  'labels': merged_labels,
                  'person_found': any(label['Name'] == 'Person' for label in labels),
                  'image_link': presigned_url,
                  'ttl': int(expiration_time.timestamp())
              }
              
              table.put_item(Item=item)
              
              return {
                  'statusCode': 200,
                  'body': json.dumps({
                      'message': 'Labels processed and saved to DynamoDB',
                      'labels': merged_labels,
                      'presigned_url': presigned_url
                  })
              }

      Environment:
        Variables:
          DYNAMODB_TABLE: !Ref DynamoDBTable
      Timeout: 60
      Role: !GetAtt ExecutionRole.Arn

  ImageProcessorPermission:
    Type: AWS::Lambda::Permission
    Properties:
      Action: 'lambda:InvokeFunction'
      FunctionName: !Ref ImageProcessor
      Principal: s3.amazonaws.com
      SourceArn: !Sub 'arn:aws:s3:::${S3BucketName}'
      SourceAccount: !Ref AWS::AccountId

  ReportSender:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: ReportSender
      Runtime: python3.9
      Handler: index.lambda_handler
      Code:
        ZipFile: |
          import boto3
          from datetime import datetime, timedelta
          from decimal import Decimal
          import json
          import os

          def convert_to_est(utc_time):
              est_offset = timedelta(hours=-4)  # Eastern Standard Time (EST) offset is -5 hours
              est_time = utc_time + est_offset
              return est_time

          def lambda_handler(event, context):
              # Create DynamoDB resource and specify table name
              dynamodb = boto3.resource('dynamodb')
              table_name = os.environ['DYNAMODB_TABLE']
              table = dynamodb.Table(table_name)

              # Convert the UTC time to Eastern Standard Time (EST)
              now_est = convert_to_est(datetime.now())
              today = now_est.date()

              # Calculate the start and end timestamps for the current day
              start_time = int(datetime.combine(today, datetime.min.time()).timestamp())
              end_time = int(datetime.combine(today, datetime.max.time()).timestamp())

              # Set the filter expression to retrieve items for the current day
              filter_expression = '#ts BETWEEN :start_time AND :end_time'

              # Set the expression attribute names and values
              expression_attribute_names = {
                  '#ts': 'timestamp',
              }
              expression_attribute_values = {
                  ':start_time': Decimal(str(start_time)),
                  ':end_time': Decimal(str(end_time)),
              }

              # Perform the scan operation
              response = table.scan(
                  FilterExpression=filter_expression,
                  ExpressionAttributeNames=expression_attribute_names,
                  ExpressionAttributeValues=expression_attribute_values
              )

              # Get the items from the response
              items = response.get('Items', [])

              # Sort the items based on person_found (True comes first) and timestamp (ascending order)
              items = sorted(items, key=lambda x: (not x['person_found'], x['timestamp']))

              # Prepare the report message
              report_message = f"Visitors Report - {today}\n\n"

              for item in items:
                  timestamp = datetime.fromtimestamp(int(item['timestamp'])).strftime('%Y-%m-%d %H:%M:%S')
                  image_name = item['image_name']
                  image_link = item['image_link']
                  labels = item['labels']
                  person_found = item['person_found']

                  report_message += f"Timestamp: {timestamp}\n"
                  report_message += f"Image Name: {image_name}\n"
                  report_message += f"Person Found: {person_found}\n"
                  report_message += f"Labels: {', '.join(labels)}\n"
                  report_message += f"Image Link: {image_link}\n"
                  report_message += "\n"

              # Publish the report message to an SNS topic
              sns_topic_arn = os.environ['SNS_TOPIC_ARN']
              sns_client = boto3.client('sns')
              sns_client.publish(
                  TopicArn=sns_topic_arn,
                  Message=report_message,
                  Subject=f"Visitor Report - {today}"
              )

              return {
                  'statusCode': 200,
                  'body': json.dumps({
                      'message': 'Visitor report sent successfully'
                  })
              }
      Environment:
        Variables:
          SNS_TOPIC_ARN: !Ref SNSTopic
          DYNAMODB_TABLE: !Ref DynamoDBTable
      Timeout: 60
      Role: !GetAtt ExecutionRole.Arn

  ExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: LambdaExecutionRole
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
            #- arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
            - arn:aws:iam::aws:policy/AmazonS3FullAccess
            - arn:aws:iam::aws:policy/AmazonSNSFullAccess
            - arn:aws:iam::aws:policy/AmazonRekognitionFullAccess
            - arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess

  DynamoDBTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: !Ref DynamoDBTableName
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: timestamp
          AttributeType: N
      KeySchema:
        - AttributeName: timestamp
          KeyType: HASH
      TimeToLiveSpecification:
        AttributeName: ttl
        Enabled: true

  SNSTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: !Ref SNSTopicName

  SNSTopicSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      Endpoint: !Ref EmailAddress
      Protocol: email
      TopicArn: !Ref SNSTopic

  EventBridgeRule:
    Type: AWS::Events::Rule
    Properties:
      Name: ReportSenderEventRule
      Description: EventBridge rule for Lambda ReportSender
      ScheduleExpression: !Ref EventBridgeTime
      State: ENABLED
      Targets:
        - Arn: !GetAtt ReportSender.Arn
          Id: TargetFunction

  EventPermission:
    Type: AWS::Lambda::Permission
    Properties:
      Action: lambda:InvokeFunction
      FunctionName: !Ref ReportSender
      Principal: events.amazonaws.com
      SourceArn: !GetAtt EventBridgeRule.Arn

Outputs:
  SnapshotGeneratorFunctionUrl:
    Description: URL of the SnapshotGenerator Lambda function
    Value: !GetAtt SnapshotGeneratorFunctionUrl.FunctionUrl