[AWS] Face Rekognition:search_faces_by_image 를 활용해서 출석부 활용하기

 

S3 버킷 권한 설정

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowLambdaAccess",
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::YOUR_ACCOUNT_ID:role/YOUR_LAMBDA_EXECUTION_ROLE_NAME"
            },
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::YOUR_BUCKET_NAME",
                "arn:aws:s3:::YOUR_BUCKET_NAME/*"
            ]
        }
    ]
}

 

BucketName은 AWS Bucket Name 정책에 맞게 해주셔야 합니다 ㅎㅎ

Dynamodb 설정

DynamoDB 메인화면

 

 

저는 테스트 용으로 만들었기 때문에, 간단하게 DB 테이블을 만들어봤습니다.

파티션 키는 Name, 정렬 키는 Date로 하여서 Table을 생성했습니다.

표 항목 탐색

항목 생성

 

항목 생성을 클릭해서, 속성 값을 추가해 줍니다. 

저는 similarity, Status, Timestamp를 추가했습니다.

속성 추가

성공적으로 속성 값이 들어간 것을 확인할 수 있습니다.

 

Lambda 함수 설정

import json
import boto3
import base64
import logging
from datetime import datetime
from decimal import Decimal
import os
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3 = boto3.client('s3')
rekognition = boto3.client('rekognition')
dynamodb = boto3.resource('dynamodb')
REKOGNITION_COLLECTION_ID = 'employees-faces'
DYNAMODB_TABLE = 'checkin_out'

 

모듈 임포트, 로깅 설정, AWS 클라이언트 생성[S3, Rekognition, DynamoDB],

Rekognition Collection ID, DynamoDB 테이블 이름을 변수에 저장합니다.

def lambda_handler(event, context):
    logger.info('Received event : ' + json.dumps(event))
    
    try:
        if 'body' not in event:
            raise ValueError("No body found in the event")
        
        body = json.loads(event['body']) if isinstance(event['body'], str) else event['body']
        
        if 'image' not in body:
            raise ValueError('Image not found in the request body')
            
        target_image = body['image']
        logger.info(f'Target image length: {len(target_image)}')
        
        target_image_bytes = base64.b64decode(target_image)

 

이벤트에 'body'가 포함되어있는지 확인하고, 또 'body' 안에는 'image'가 있는 지 확인합니다. 

base64로 인코딩 되어서 온 값을 target_image에 저장하고, 

다시 인코딩 된 이미지를 디코딩하여 바이너리 데이터[바이트 배열]로 변환 합니다.

즉, API Gateway를 통해 Base64 인코딩된 이미지 데이터를 받아서 

그 것을 디코딩하여 AWS Rekognition에서 사용할 수 있는 

바이너리 데이터로 변환하는 것.

 

Base64에 대해서 간단히 설명하자면, 바이너리 데이터를 텍스트 형식으로 인코딩하는 방법으로,

주로 이미지나 파일 데이터를 텍스트로 전송할 때 사용한다고 합니다. 

 

        response = rekognition.search_faces_by_image(
            CollectionId=REKOGNITION_COLLECTION_ID,
            Image={'Bytes': target_image_bytes},
            MaxFaces=10,
            FaceMatchThreshold=85
        )
        
        logger.info(f"Rekognition response: {json.dumps(response)}")

 

디코딩된 이미지 데이터를 Rekognition Collection에서 검색합니다. 

여기서 MaxFaces 파라미터가 의미하는 것은 이미지에서 검색할 얼굴의 최대 개수를 설정합니다.

이미지에 다수의 얼굴이 있을 경우, 이 파라미터를 사용하여 검색 결과의 크기를 제한하도록 하는 것 입니다.

예를 들어, 이미지에 100명의 얼굴이 있어도 MaxFaces를 10으로 설정하면 최대 10명의 얼굴만 반환하도록 합니다.

 

그러면 MaxFaces에 대해서 더 궁금한게,

"만약 10이라고 한다면 어떤 기준으로 10개를 가져오냐?" 

유사도가 높은 순서대로 최대 MaxFaces에 설정된 개수만큼의 매칭 결과를 반환

 

FaceMatchThreshold는 similarity의 임계값이라고 생각하면 됩니다. 

일정 similarity 이상의 결과만 가져오라는 의미 입니다. 

너무 낮게 주면 예상치 못한 결과가 나올 수 있고

반대로 너무 높게 주면 아예 안 나올 수도 있기 때문에

적당히 휴리스틱한 방법으로 찾아야 합니다.

 

        attendance_results = []
        table = dynamodb.Table(DYNAMODB_TABLE)
        now = datetime.now()
        date = now.strftime("%Y-%m-%d")
        timestamp = now.isoformat()
        
        for match in response['FaceMatches']:
            person_name = match['Face']['ExternalImageId']
            similarity = Decimal(str(match['Similarity']))
            
            try:
                response = table.get_item(
                    Key={
                        'Name': person_name,
                        'Date': date
                    }
                )
                
                if 'Item' in response:
                    attendance_status = "이미 출석했습니다"
                else:
                    table.put_item(
                        Item={
                            'Name': person_name,
                            'Date': date,
                            'Timestamp': timestamp,
                            'Similarity': similarity,
                            'Status': '출석'
                        }
                    )
                    attendance_status = "출석 완료"
                
                attendance_results.append({
                    "person": person_name,
                    "similarity": float(similarity),
                    "status": attendance_status
                })
            
            except ClientError as e:
                logger.error(e.response['Error']['Message'])
                attendance_results.append({
                    "person": person_name,
                    "error": "데이터베이스 오류"
                })

 

person_name에다가 Collection에 넣을 때 ExternalImageId는 사람의 이름으로 넣어서 저장했기에 

ExternalImageId를 가져오면 됩니다.

Similarity는 처음엔 그냥 가져왔었는데, 밑에 오류를 첨부했는데, DynamoDB가 python의 'float'을

직접적으로 지원하지 않기 때문에 발생합니다.

그래서 'Decimal' 타입으로 변환해야 합니다.

[ERROR] 2024-07-01T11:06:33.247Z 1fbd284c-4d25-4b16-a61f-9631ca22da38 Unexpected error: Float types are not supported. Use Decimal types instead.

Key : 조회할 항목의 primary key를 지정해서, 'Name'과 'Date'를 사용하여 특정 인물이

특정 날짜에 출석했는지 확인합니다.

만약 response 객체에 Item 키가 있으면, 이미 출석했다는 것이고, 

Item 키가 없으면, 해당 인물이 오늘 아직 출석하지 않았다는 것을 의미합니다. 

그러면 그 정보를 DynamoDB 테이블에 추가하는 매커니즘 입니다. 

 

attendance_results라는 리스트에 출석 결과를 추가해서 반환합니다.

[HTML 화면에 띄어주기 위에서 return으로 해당 딕셔너리를 보냅니다]

 

 

if attendance_results:
    return {
        "statusCode": 200,
        "body": json.dumps({
            "match": True,
            "results": attendance_results,
            "timestamp": timestamp
        }),
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        }
    }
else:
    return {
        "statusCode": 200,
        "body": json.dumps({
            "match": False,
            "message": "No matching faces found"
        }),
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        }
    }

 

attendance_results 리스트에 출석 결과가 있는 지를 확인하고, 

HTTP 200 상태 코드와 함께 매칭된 얼굴 정보와 타임스탬프를 반환합니다.

출석 결과가 없다면, HTTP 200 상태코드와 함께 매칭되지 않았다는 메시지를 반환합니다. 

 

except ValueError as ve:
    logger.error(f"ValueError: {str(ve)}")
    return {
        "statusCode": 400,
        "body": json.dumps({"error": str(ve)}),
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        }
    }
    
except ClientError as e:
    logger.error(f"ClientError: {e.response['Error']['Message']}")
    return {
        "statusCode": 500,
        "body": json.dumps({"error": "Database error"}),
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        }
    }
    
except Exception as e:
    logger.error(f"Unexpected error: {str(e)}")
    return {
        "statusCode": 500,
        "body": json.dumps({"error": "Internal server error"}),
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        }
    }

 

ValueError, ClientError, Exception을 넣어서 예외 처리를 합니다. 

ClientError의 경우 AWS SDK인 boto3에서 발생하는 예외로, 

AWS 서비스와의 상호작용 중에 클라이언트 측에서 오류가 발생했음을 나타냅니다.

저는 어제 저 오류가 계속 발생해서 어떤 문제인지 봤었는데,

제가 REKOGNITION_COLLECTION_ID를 잘못입력해서 발생했더라구요.

ClientError가 뜨면 AWS 서비스 쪽 어디 변수명에 내가 잘못 입력한 게 없나를 확인해

보시는 게 좋아요~

ClientError

 

S3에 있는 이미지 파일 Rekognition Collection에 올리는 법!!

https://mrkite.tistory.com/35

[AWS] Rekognition Collection에 S3에 있는 파일들을 저장해보자!