AWS Lambda 함수를 사용하여 이상 항목 찾기 - Amazon Lookout for Vision
1단계: AWS Lambda 함수 생성(콘솔)2단계: (선택 사항) 계층 생성 (콘솔)3단계: Python 코드 추가 (콘솔)4단계: Lambda 함수 테스트

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

AWS Lambda 함수를 사용하여 이상 항목 찾기

AWS Lambda은 서버를 프로비저닝하거나 관리하지 않고도 코드를 실행할 수 있게 해주는 컴퓨팅 서비스입니다. 예를 들어 애플리케이션 코드를 호스팅할 서버를 만들지 않고도 모바일 애플리케이션에서 제출된 이미지를 분석할 수 있습니다. 다음 지침은 DetectCustomLabels를 호출하는 Lambda 함수를 Python에서 생성하는 방법을 보여줍니다. 이 함수는 제공된 이미지를 분석하고 해당 이미지에 이상이 있는지 분류하여 반환합니다. 지침에는 Amazon S3 버킷의 이미지 또는 로컬 컴퓨터에서 제공된 이미지를 사용하여 Lambda 함수를 호출하는 방법을 보여주는 예제 Python 코드가 포함되어 있습니다.

1단계: AWS Lambda 함수 생성(콘솔)

이 단계에서는 빈 AWS 함수와 함수가 DetectAnomalies 작업을 호출하게 해주는 IAM 실행 역할을 생성합니다. 또한 분석용 이미지를 저장하는 Amazon S3 버킷에 대한 액세스 권한도 부여합니다. 또한 다음에 대한 환경 변수를 지정합니다.

  • Lambda 함수에서 사용하려는 Amazon Lookout for Vision 프로젝트 및 모델 버전.

  • 모델에서 사용하려는 신뢰 한도.

나중에 Lambda 함수에 소스 코드와 계층(선택 사항)을 추가합니다.

AWS Lambda 함수를 생성하려면(콘솔)

  1. AWS Management Console에 로그인하고 AWS Lambdahttps://console.aws.amazon.com/lambda/에서 콘솔을 엽니다.

  2. 함수 생성(Create function)을 선택합니다. 자세한 내용은 콘솔로 Lambda 함수 생성을 참조하세요.

  3. 다음과 같은 옵션을 선택하세요.

    • 새로 작성을 선택합니다.

    • 함수 이름의 값을 입력합니다.

    • 런타임에는 Python 3.10을 선택합니다.

  4. 함수 생성을 선택하여 AWS Lambda 함수를 생성합니다.

  5. 함수 페이지에서 구성 탭을 선택합니다.

  6. 환경 변수 창에서 편집을 선택합니다.

  7. 다음 환경 변수를 추가합니다. 각 변수에 대해 환경 변수 추가를 선택한 다음 변수 키와 값을 입력합니다.

    키 값

    PROJECT_NAME

    사용하려는 모델이 들어 있는 Lookout for Vision 프로젝트입니다.

    MODEL_VERSION

    사용하려는 모델의 버전입니다.

    CONFIDENCE

    예측이 비정상적이라는 모델 신뢰도의 최소값 (0-100) 입니다. 신뢰도가 낮으면 분류가 정상으로 간주됩니다.

  8. 저장을 선택하여 환경 변수를 저장합니다.

  9. 권한 창의 실행 역할에서 IAM 콘솔의 역할을 열 실행 역할을 선택합니다.

  10. 권한 탭에서 권한 추가, 그다음 인라인 정책 생성을 선택합니다.

  11. JSON을 선택하고 기존 정책을 다음 정책으로 바꿉니다.

    {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": "lookoutvision:DetectAnomalies",
            "Resource": "*",
            "Effect": "Allow",
            "Sid": "DetectAnomaliesAccess"
        }
    ]
}

  12. 다음(Next)을 선택합니다.

  13. 정책 세부 정보에 정책 이름 (예: DetectAnomalies-access)을 입력합니다.

  14. 정책 생성을 선택합니다.

  15. Amazon S3 버킷에 분석용 이미지를 저장하는 경우 10~14단계를 반복합니다.

    1. 11단계에서는 다음 정책을 사용하십시오. 버킷/폴더 경로를 Amazon S3 버킷 및 분석하려는 이미지의 폴더 경로로 바꿉니다.

      {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3Access",
            "Effect": "Allow",
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::bucket/folder path/*"
        }
    ]
}

    2. 13단계에서는 S3Bucket-access와 같은 다른 정책 이름을 선택합니다.

2단계: (선택 사항) 계층 생성 (콘솔)

이 단계는 예제를 실행하는 데 필수 사항이 아닙니다. DetectAnomalies 작업은 Python용 AWS SDK(Boto3)의 일부로 기본 Lambda Python 환경에 포함됩니다. Lambda 함수의 다른 부분에서 기본 Lambda Python 환경에 없는 최신 AWS 서비스 업데이트가 필요한 경우, 이 단계를 수행하여 최신 Boto3 SDK 릴리스를 함수에 계층으로 추가하세요.

먼저 Boto3 SDK가 포함된 .zip 파일 아카이브를 생성합니다. 그런 다음 계층을 생성하고 .zip 파일 아카이브를 계층에 추가합니다. 자세한 내용은 Lambda 함수에서 계층 사용을 참조하세요.

계층을 생성하고 추가하려면(콘솔)

  1. 명령 프롬프트를 열고 다음 명령을 입력합니다.

    pip install boto3 --target python/.
zip boto3-layer.zip -r python/

  2. zip 파일(boto3-layer.zip)의 이름을 기록해 두세요. 이 절차의 6단계에서 필요합니다.

  3. https://console.aws.amazon.com/lambda/에서 AWS Lambda 콘솔을 엽니다.

  4. 탐색 창에서 계층을 선택합니다.

  5. 계층 생성을 선택합니다.

  6. Name(이름)과 Description(설명)을 입력합니다.

  7. .zip 파일 업로드를 선택한 후 업로드를 선택합니다.

  8. 대화 상자에서 이 절차의 1단계에서 만든.zip 파일 아카이브 (boto3-layer.zip)를 선택합니다.

  9. 호환되는 런타임은 Python 3.9를 선택하세요.

  10. 생성을 선택하여 계층을 생성합니다.

  11. 탐색 창 메뉴 아이콘을 선택합니다.

  12. 탐색 창에서 함수를 선택합니다.

  13. 리소스 목록에서 1단계: AWS Lambda 함수 생성(콘솔) 항목에서 생성한 함수를 선택합니다.

  14. 코드 탭을 선택합니다.

  15. 계층 영역에서 계층 추가를 선택합니다.

  16. 사용자 지정 계층을 선택합니다.

  17. 사용자 지정 계층에서 6단계에서 입력한 계층 이름을 선택합니다.

  18. 버전에서 계층 버전을 선택합니다. 계층 버전은 1이어야 합니다.

  19. 추가를 선택합니다.

3단계: Python 코드 추가 (콘솔)

이 단계에서는 Lambda 콘솔 코드 편집기를 사용하여 Lambda 함수에 Python 코드를 추가합니다. 코드는 제공된 이미지를 DetectAnomalies로 분석하여 분류를 반환합니다 (이미지가 비정상이면 true, 이미지가 정상이면 false). 제공된 이미지는 Amazon S3 버킷에 있거나 byte64로 인코딩된 이미지 바이트로 제공될 수 있습니다.

Python 코드를 추가하려면 (콘솔)

  1. Lambda 콘솔을 사용하지 않는 경우 다음을 수행합니다.

    1. https://console.aws.amazon.com/lambda/에서 AWS Lambda 콘솔을 엽니다.

    2. 1단계: AWS Lambda 함수 생성(콘솔)에서 생성한 Lambda 함수를 엽니다.

  2. 코드 탭을 선택합니다.

  3. 코드 소스에서 lambda_function.py의 코드를 다음과 같이 바꾸세요.

    # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0

"""
Purpose
An AWS lambda function that analyzes images with an Amazon Lookout for Vision model.
"""
import base64
import imghdr
from os import environ
from io import BytesIO
import logging
import boto3

from botocore.exceptions import ClientError


logger = logging.getLogger(__name__)


# Get the model and confidence.
project_name = environ['PROJECT_NAME']
model_version = environ['MODEL_VERSION']
min_confidence = int(environ.get('CONFIDENCE', 50))

lookoutvision_client = boto3.client('lookoutvision')


def lambda_handler(event, context):
    """
    Lambda handler function
    param: event: The event object for the Lambda function.
    param: context: The context object for the lambda function.
    return: The labels found in the image passed in the event
    object.
    """

    try:
        
        file_name = ""

        # Determine image source.
        if 'image' in event:
            # Decode the encoded image
            image_bytes = event['image'].encode('utf-8')
            img_b64decoded = base64.b64decode(image_bytes)
            image_type = get_image_type(img_b64decoded)
            image = BytesIO(img_b64decoded)
            file_name = event['filename']

        elif 'S3Object' in event:
            bucket = boto3.resource('s3').Bucket(event['S3Object']['Bucket'])
            image_object = bucket.Object(event['S3Object']['Name'])
            image = image_object.get().get('Body').read()
            image_type = get_image_type(image)
            file_name = f"s3://{event['S3Object']['Bucket']}/{event['S3Object']['Name']}"

        else:
            raise ValueError(
                'Invalid image source. Only base 64 encoded image bytes or images in S3 buckets are supported.')

        # Analyze the image.
        response = lookoutvision_client.detect_anomalies(
            ProjectName=project_name,
            ContentType=image_type,
            Body=image,
            ModelVersion=model_version)

        reject = reject_on_classification(
            response['DetectAnomalyResult'], confidence_limit=float(environ['CONFIDENCE'])/100)

        status = "anomalous" if reject else "normal"

        lambda_response = {
            "statusCode": 200,
            "body": {
                "Reject": reject,
                "RejectMessage": f"Image {file_name} is {status}."
            }
        }

    except ClientError as err:
        error_message = f"Couldn't analyze {file_name}. " + \
            err.response['Error']['Message']

        lambda_response = {
            'statusCode': 400,
            'body': {
                "Error": err.response['Error']['Code'],
                "ErrorMessage": error_message,
                "Image": file_name
            }
        }
        logger.error("Error function %s: %s",
                     context.invoked_function_arn, error_message)

    except ValueError as val_error:

        lambda_response = {
            'statusCode': 400,
            'body': {
                "Error": "ValueError",
                "ErrorMessage": format(val_error),
                "Image": event['filename']
            }
        }
        logger.error("Error function %s: %s",
                     context.invoked_function_arn, format(val_error))

    return lambda_response

def get_image_type(image):
    """
    Gets the format of the image. Raises an error
    if the type is not PNG or JPEG.
    :param image: The image that you want to check.
    :return The type of the image.

    """
    image_type = imghdr.what(None, image)

    if image_type == "jpeg":
        content_type = "image/jpeg"
    elif image_type == "png":
        content_type = "image/png"
    else:
        logger.info("Invalid image type")
        raise ValueError(
            "Invalid file format. Supply a jpeg or png format file.")
    return content_type


def reject_on_classification(prediction, confidence_limit):
    """
    Returns True if the anomaly confidence is greater than or equal to
    the supplied confidence limit.
    :param image: The name of the image file that was analyzed.
    :param prediction: The DetectAnomalyResult object returned from DetectAnomalies
    :param confidence_limit: The minimum acceptable confidence. Float value between 0 and 1.
    :return: True if the error condition indicates an anomaly, otherwise False.
    """

    reject = False

    if prediction['IsAnomalous'] and prediction['Confidence'] >= confidence_limit:
        reject = True
        reject_info = (f"Rejected: Anomaly confidence ({prediction['Confidence']:.2%}) is greater"
                       f" than limit ({confidence_limit:.2%})")
        logger.info("%s", reject_info)

    if not reject:
        logger.info("No anomalies found.")
    return reject

  4. Lambda 함수를 배포하려면 배포를 선택합니다.

4단계: Lambda 함수 테스트

이 단계에서는 컴퓨터의 Python 코드를 사용하여 로컬 이미지 또는 Amazon S3 버킷의 이미지를 Lambda 함수에 전달합니다. 로컬 컴퓨터에서 전달된 이미지는 6291456바이트보다 작아야 합니다. 이미지가 더 크면 Amazon S3 버킷에 이미지를 업로드하고 이미지에 대한 Amazon S3 경로를 사용하여 스크립트를 호출합니다. Amazon S3 버킷에 이미지 파일을 업로드하는 방법에 대한 자세한 내용은 객체 업로드를 참조하세요.

Lambda 함수를 생성한 AWS 리전과 동일한 리전에서 코드를 실행하는지 확인합니다. Lambda 콘솔에 있는 함수 세부 정보 페이지의 탐색 모음에서 Lambda 함수의 AWS 리전을 볼 수 있습니다.

AWS Lambda 함수가 제한 시간 오류를 반환하는 경우 Lambda 함수 제한 시간 기간을 연장하세요. 자세한 내용은 함수 제한 시간 구성(콘솔)을 참조하세요.

코드에서 Lambda 함수를 간접 호출하는 방법에 대한 자세한 내용은 AWS Lambda 함수 간접 호출을 참조하세요.

Lambda 함수를 테스트하는 방법

  1. 아직 수행하지 않은 경우 다음 작업을 수행하십시오.

    1. 클라이언트 코드를 사용하는 사용자에게 lambda:InvokeFunction 권한이 있는지 확인하십시오. 다음 권한을 사용할 수 있습니다.

      {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "LambdaPermission",
            "Effect": "Allow",
            "Action": "lambda:InvokeFunction",
            "Resource": "ARN for lambda function"
        }
    ]
}

      Lambda 콘솔의 함수 개요에서 Lambda 함수 함수에 대한 ARN을 가져올 수 있습니다.

      액세스 권한을 제공하려면 사용자, 그룹 또는 역할에 권한을 추가합니다.

      • AWS IAM Identity Center의 사용자 및 그룹:

        권한 세트를 생성합니다. AWS IAM Identity Center 사용 설명서의 권한 세트 생성의 지침을 따르세요.

      • 자격 증명 공급자를 통해 IAM에서 관리되는 사용자:

        아이덴티티 페더레이션을 위한 역할을 생성합니다. IAM 사용 설명서의 서드 파티 자격 증명 공급자의 역할 만들기(연합)의 지침을 따르세요.

      • IAM 사용자:

        • 사용자가 맡을 수 있는 역할을 생성합니다. IAM 사용 설명서에서 IAM 사용자의 역할 생성의 지침을 따르세요.

        • (권장되지 않음) 정책을 사용자에게 직접 연결하거나 사용자를 사용자 그룹에 추가합니다. IAM 사용 설명서에서 사용자(콘솔)에 권한 추가의 지침을 따르세요.

    2. Python용 AWS SDK를 설치하고 구성합니다. 자세한 내용은 4단계: AWS CLI 및 AWS SDK 설정 섹션을 참조하세요.

    3. 1단계: AWS Lambda 함수 생성(콘솔)의 7단계에서 지정한 모델을 시작합니다.

  2. 다음 코드를 client.py 이름의 파일에 저장합니다.

    # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0

"""
Purpose: Shows how to call the anomaly detection
AWS Lambda function.
"""
from botocore.exceptions import ClientError

import argparse
import logging
import base64
import json
import boto3
from os import environ

logger = logging.getLogger(__name__)

def analyze_image(function_name, image):
    """
    Analyzes an image with an AWS Lambda function.
    :param image: The image that you want to analyze.
    :return The status and classification result for
    the image analysis.
    """

    lambda_client = boto3.client('lambda')

    lambda_payload = {}

    if image.startswith('s3://'):
        logger.info("Analyzing image from S3 bucket: %s", image)
        bucket, key = image.replace("s3://", "").split("/", 1)
        s3_object = {
            'Bucket': bucket,
            'Name': key
        }
        lambda_payload = {"S3Object": s3_object}

    # Call the lambda function with the image.
    else:
        with open(image, 'rb') as image_file:
            logger.info("Analyzing local image image: %s ", image)
            image_bytes = image_file.read()
            data = base64.b64encode(image_bytes).decode("utf8")
            lambda_payload = {"image": data, "filename": image}

    response = lambda_client.invoke(FunctionName=function_name,
                                    Payload=json.dumps(lambda_payload))
    return json.loads(response['Payload'].read().decode())


def add_arguments(parser):
    """
    Adds command line arguments to the parser.
    :param parser: The command line parser.
    """

    parser.add_argument(
        "function", help="The name of the AWS Lambda function "
        "that you want to use to analyze the image.")
    parser.add_argument(
        "image", help="The local image that you want to analyze.")


def main():
    """
    Entrypoint for script.
    """
    try:
        logging.basicConfig(level=logging.INFO,
                            format="%(levelname)s: %(message)s")

        # Get command line arguments.
        parser = argparse.ArgumentParser(usage=argparse.SUPPRESS)
        add_arguments(parser)
        args = parser.parse_args()

        # Analyze image and display results.

        result = analyze_image(args.function, args.image)

        status = result['statusCode']

        if status == 200:
            classification = result['body']
            print(f"classification: {classification['Reject']}")
            print(f"Message: {classification['RejectMessage']}")
        else:
            print(f"Error: {result['statusCode']}")
            print(f"Message: {result['body']}")

    except ClientError as error:
        logging.error(error)
        print(error)


if __name__ == "__main__":
    main()

  3. 코드를 실행합니다. 명령줄 인수의 경우 Lambda 함수 이름과 분석하려는 로컬 이미지의 경로를 제공하십시오. 예:

    python client.py function_name /bucket/path/image.jpg

    성공하면 이미지에서 발견된 예외 항목에 대한 분류가 출력됩니다. 분류가 반환되지 않는 경우 1단계: AWS Lambda 함수 생성(콘솔)의 7단계에서 설정한 신뢰도를 낮추는 것을 고려해 보십시오.

  4. Lambda 함수 사용을 마쳤지만 다른 애플리케이션에서 해당 모델을 사용하지 않는 경우 모델을 중지하세요. 다음에 Lambda 함수를 사용하려는 경우 모델을 시작해야 한다는 점을 기억하십시오.