AWS Lambda 함수에서 이상 찾기

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

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

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

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

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

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

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

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

권한 창의 역할 이름 아래에서 실행 역할을 선택하여 IAM 콘솔에서 역할을 엽니다.

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

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

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

Next(다음)를 선택합니다.

정책 세부 정보 에 DetectAnomalies-access 와 같은 정책의 이름을 입력합니다.

정책 생성을 선택합니다.

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

1. 11단계에서는 다음 정책을 사용하십시오. Replace bucket/folder path 분석하려는 이미지에 대한 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와 같은 다른 정책 이름을 선택합니다.

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

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

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

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

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

계층 생성을 선택합니다.

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

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

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

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

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

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

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

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

코드 탭을 선택합니다.

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

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

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

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

추가를 선택합니다.

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

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

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

코드 탭을 선택합니다.

코드 소스에서 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

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

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

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

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

   Lambda 콘솔 의 함수 개요에서 Lambda 함수에 ARN 대한 를 가져올 수 있습니다. https://console.aws.amazon.com/lambda/

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

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

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

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

     ID 페더레이션을 위한 역할을 생성합니다. IAM 사용 설명서의 타사 자격 증명 공급자(페더레이션)에 대한 역할 생성의 지침을 따릅니다.

   • IAM 사용자:

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

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

2. Python용 를 설치하고 구성합니다 AWS SDK. 자세한 내용은 4단계: AWS CLI 및 설정 AWS SDKs 단원을 참조하십시오.

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

다음 코드를 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()

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

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

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

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