Real-Time Face Search with Raspberry Pi, Kinesis Video Streams, and Amazon Rekognition

1
AWSTemplateFormatVersion: 2010-09-09
2
Transform: AWS::Serverless-2016-10-31
3
Description: face-detector-using-kinesis-video-streams
4

5
Resources:
6
  Function:
7
    Type: AWS::Serverless::Function
8
    Properties:
9
      FunctionName: face-detector-function
10
      CodeUri: src/
11
      Handler: app.lambda_handler
12
      Runtime: python3.11
13
      Architectures:
14
        - arm64
15
      Timeout: 3
16
      MemorySize: 128
17
      Role: !GetAtt FunctionIAMRole.Arn
18
      Events:
19
        KinesisEvent:
20
          Type: Kinesis
21
          Properties:
22
            Stream: !GetAtt KinesisStream.Arn
23
            MaximumBatchingWindowInSeconds: 10
24
            MaximumRetryAttempts: 3
25
            StartingPosition: LATEST
26

27
  FunctionIAMRole:
28
    Type: AWS::IAM::Role
29
    Properties:
30
      RoleName: face-detector-function-role
31
      AssumeRolePolicyDocument:
32
        Version: 2012-10-17
33
        Statement:
34
          - Effect: Allow
35
            Principal:
36
              Service: lambda.amazonaws.com
37
            Action: sts:AssumeRole
38
      ManagedPolicyArns:
39
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
40
        - arn:aws:iam::aws:policy/service-role/AWSLambdaKinesisExecutionRole
41
      Policies:
42
        - PolicyName: policy
43
          PolicyDocument:
44
            Version: 2012-10-17
45
            Statement:
46
              - Effect: Allow
47
                Action:
48
                  - kinesisvideo:GetHLSStreamingSessionURL
49
                  - kinesisvideo:GetDataEndpoint
50
                Resource: !GetAtt KinesisVideoStream.Arn
51

52
  KinesisVideoStream:
53
    Type: AWS::KinesisVideo::Stream
54
    Properties:
55
      Name: face-detector-kinesis-video-stream
56
      DataRetentionInHours: 24
57

58
  RekognitionCollection:
59
    Type: AWS::Rekognition::Collection
60
    Properties:
61
      CollectionId: FaceCollection
62

63
  RekognitionStreamProcessor:
64
    Type: AWS::Rekognition::StreamProcessor
65
    Properties:
66
      Name: face-detector-rekognition-stream-processor
67
      KinesisVideoStream:
68
        Arn: !GetAtt KinesisVideoStream.Arn
69
      KinesisDataStream:
70
        Arn: !GetAtt KinesisStream.Arn
71
      RoleArn: !GetAtt RekognitionStreamProcessorIAMRole.Arn
72
      FaceSearchSettings:
73
        CollectionId: !Ref RekognitionCollection
74
        FaceMatchThreshold: 80
75
      DataSharingPreference:
76
        OptIn: false
77

78
  KinesisStream:
79
    Type: AWS::Kinesis::Stream
80
    Properties:
81
      Name: face-detector-kinesis-stream
82
      StreamModeDetails:
83
        StreamMode: ON_DEMAND
84

85
  RekognitionStreamProcessorIAMRole:
86
    Type: AWS::IAM::Role
87
    Properties:
88
      RoleName: face-detector-rekognition-stream-processor-role
89
      AssumeRolePolicyDocument:
90
        Version: 2012-10-17
91
        Statement:
92
          - Effect: Allow
93
            Principal:
94
              Service: rekognition.amazonaws.com
95
            Action: sts:AssumeRole
96
      ManagedPolicyArns:
97
        - arn:aws:iam::aws:policy/service-role/AmazonRekognitionServiceRole
98
      Policies:
99
        - PolicyName: policy
100
          PolicyDocument:
101
            Version: 2012-10-17
102
            Statement:
103
              - Effect: Allow
104
                Action:
105
                  - kinesis:PutRecord
106
                  - kinesis:PutRecords
107
                Resource:
108
                  - !GetAtt KinesisStream.Arn

1
import base64
2
import json
3
import logging
4
from datetime import datetime, timedelta, timezone
5
from functools import cache
6

7
import boto3
8

9
JST = timezone(timedelta(hours=9))
10
kvs_client = boto3.client('kinesisvideo')
11
logger = logging.getLogger(__name__)
12
logger.setLevel(logging.INFO)
13

14

15
def lambda_handler(event: dict, context: dict) -> dict:
16
  for record in event['Records']:
17
    base64_data = record['kinesis']['data']
18
    stream_processor_event = json.loads(base64.b64decode(base64_data).decode())
19
    # Refer to https://docs.aws.amazon.com/rekognition/latest/dg/streaming-video-kinesis-output.html for details on the structure.
20

21
    if not stream_processor_event['FaceSearchResponse']:
22
      continue
23

24
    logger.info(stream_processor_event)
25
    url = get_hls_streaming_session_url(stream_processor_event)
26
    logger.info(url)
27

28
  return {
29
    'statusCode': 200,
30
  }
31

32

33
@cache
34
def get_kvs_am_client(api_name: str, stream_arn: str):
35
  # Retrieves the data endpoint for the stream.
36
  # See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesisvideo/client/get_data_endpoint.html
37
  endpoint = kvs_client.get_data_endpoint(
38
    APIName=api_name.upper(),
39
    StreamARN=stream_arn
40
  )['DataEndpoint']
41
  return boto3.client('kinesis-video-archived-media', endpoint_url=endpoint)
42

43

44
def get_hls_streaming_session_url(stream_processor_event: dict) -> str:
45
  # Generates an HLS streaming URL for the video stream.
46
  # See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesis-video-archived-media/client/get_hls_streaming_session_url.html
47

48
  kinesis_video = stream_processor_event['InputInformation']['KinesisVideo']
49
  stream_arn = kinesis_video['StreamArn']
50
  kvs_am_client = get_kvs_am_client('get_hls_streaming_session_url', stream_arn)
51
  start_timestamp = datetime.fromtimestamp(kinesis_video['ServerTimestamp'], JST)
52
  end_timestamp = datetime.fromtimestamp(kinesis_video['ServerTimestamp'], JST) + timedelta(minutes=1)
53

54
  return kvs_am_client.get_hls_streaming_session_url(
55
    StreamARN=stream_arn,
56
    PlaybackMode='ON_DEMAND',
57
    HLSFragmentSelector={
58
      'FragmentSelectorType': 'SERVER_TIMESTAMP',
59
      'TimestampRange': {
60
        'StartTimestamp': start_timestamp,
61
        'EndTimestamp': end_timestamp,
62
      },
63
    },
64
    ContainerFormat='FRAGMENTED_MP4',
65
    Expires=300,
66
  )['HLSStreamingSessionURL']