[AWS] Amazon Rekognition 실습

대학교 시절 프로젝트 성으로 잠깐 약간 맛만 봤었는데 이번 게임데이 준비하면서 좀 더 깊게 파볼 생각...

딥 러닝 기반 시각 분석 서비스 인데.. 신기하다 신기해! 약간 YOLO랑 비슷한 느낌?

 

참고: https://aws.amazon.com/ko/blogs/machine-learning/build-your-own-face-recognition-service-using-amazon-rekognition

---

• Indexing(blue flow): 나중에 분석하기 위해 얼굴 이미지를 컬렉션으로 가져오는 프로세스
• Analysis(black flow): 인덱스 내에서 일치하는 얼굴 컬렉션 쿼리하는 프로세스

---

• CloudFormation 실행(한번에 다 설치해줌...)

템플릿 다운

위의 이미지는 스택 생성 후 생성된 리소스들이다. 과정 하나씩 살펴보자.

• rekognition create-collection

aws rekognition create-collection --collection-id family-collection

위의 명령어를 AWS CLI에 입력해 collection을 생성하는 과정을 거친다.

 

• DynamoDB create-table

RekognitionId 를 파티션 키로 family-collection 테이블을 생성했다.

 

• IndexFaces 작업

S3 버킷 생성

 

IAM 역할 설정

1) S3 객체에 액세스할 수 있는 권한을 lambda에 부여

2) Rekognition의 IndexFaces 함수를 시작하고 FaceId 간의 매핑을 위해 DynamoDB key-value 저장소 내에 여러 항목 생성  가능

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*",
            "Effect": "Allow"
        },
        {
            "Action": [
                "s3:GetObject"
            ],
            "Resource": [
                "arn:aws:s3:::training-bucket-221004/*"
            ],
            "Effect": "Allow"
        },
        {
            "Action": [
                "rekognition:IndexFaces"
            ],
            "Resource": "*",
            "Effect": "Allow"
        },
        {
            "Action": [
                "dynamodb:PutItem"
            ],
            "Resource": "arn:aws:dynamodb:<Region>:<ACCOUNT-ID>:table/family-collection",
            "Effect": "Allow"
        }
    ]
}

해당 IAM 역할을 Lambda 함수에 붙여주자.

 

마지막으로 새 사진이 S3에 업로드될 때마다 트리거되는 Lambda 함수 생성(트리거 추가)

 

해당 lambda 함수에 코드 넣어주기.

1) Rekognition IndexFaces API를 사용하여 입력 이미지에서 얼굴을 감지하고 지정된 컬렉션에 추가

2) 성공하면 S3에 있는 객체의 메타데이터에서 사람의 전체 이름을 검색, 그런 다음 나중에 참조할 수 있도록 DynamoDB 테이블에 FaceId가 있는 key-value 튜플로 저장

import os
import boto3
from decimal import Decimal
import json
import urllib

print('Loading function')

dynamodb = boto3.client('dynamodb')
s3 = boto3.client('s3')
rekognition = boto3.client('rekognition')


# --------------- Helper Functions ------------------

def index_faces(bucket, key):
    CollectionId = os.getenv("CollectionName")

    response = rekognition.index_faces(
        Image={"S3Object": {"Bucket": bucket, "Name": key}},
        CollectionId=CollectionId,
    )
    return response


def update_index(tableName,faceId, fullName):
    response = dynamodb.put_item(
        TableName=tableName,
        Item={
        'RekognitionId': {'S': faceId},
        'FullName': {'S': fullName}
        }
    )

# --------------- Main handler ------------------

def lambda_handler(event, context):

    # Get the object from the event
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
              

    try:
        # Calls Amazon Rekognition IndexFaces API to detect faces in S3 object
        # to index faces into specified collection
        response = index_faces(bucket, key)

        # Commit faceId and full name object metadata to DynamoDB
        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
            faceId = response['FaceRecords'][0]['Face']['FaceId']

            ret = s3.head_object(Bucket=bucket,Key=key)
            personFullName = ret['Metadata']['fullname']

            update_index(os.getenv("DynamoDBTableName"), faceId, personFullName)

        # Print response to console.
        print(response)

        return response
    except Exception as e:
        print(e)
        print("Error processing {} from bucket {}. ".format(key, bucket)) 
        raise e

 

이후에 이미지를 S3에 업로드한 뒤에 해당 코드를 통해 Face collection을 시드할 수 있다.

(현재는 S3 버킷에 이미지가 없음....;; 따라서 코드 실행에 에러 나는게 당연하다..?)

 

해당 컬렉션이 잘 만들어졌는지 확인하기 위해선

aws rekognition list-faces --collection-id family-collection

 

위의 명령어를 사용하면 된다.

 

• 분석 작업

컬렉션이 채워지면 얼굴이 포함된 다른 이미지를 전달하여 쿼리가 가능하다.

SearchFacesByImageAPI를 사용하여 쿼리할 컬렉션 이름과 분석할 이미지에 대한 참조 두 가지 이상의 매개변수 제공

아래의 코드는 이미지를 bytestream으로 제출하는 방법을 보여준다.

(응답의 경우 일치 항목의 FaceId가 포함된 JSON 객체를 반환, (메타데이터 중 신뢰도 점수, 이미지 내 얼굴의 좌표))

import boto3
import io
from PIL import Image

rekognition = boto3.client('rekognition', region_name='eu-west-1')
dynamodb = boto3.client('dynamodb', region_name='eu-west-1')
    
image = Image.open("group1.jpeg")
stream = io.BytesIO()
image.save(stream,format="JPEG")
image_binary = stream.getvalue()


response = rekognition.search_faces_by_image(
        CollectionId='family_collection',
        Image={'Bytes':image_binary}                                       
        )
    
for match in response['FaceMatches']:
    print (match['Face']['FaceId'],match['Face']['Confidence'])
        
    face = dynamodb.get_item(
        TableName='family_collection',  
        Key={'RekognitionId': {'S': match['Face']['FaceId']}}
        )
    
    if 'Item' in face:
        print (face['Item']['FullName']['S'])
    else:
        print ('no match found in person lookup')

 

 

❗ 우씨 뭐가뭔지 모르겠다... Rekognition 구조부터 파악하고 해봐야할듯?! ❗