AWS Lambda 関数による異常の検出

AWS Management Console にサインインして AWS Lambda コンソール (https://console.aws.amazon.com/lambda/) を開きます。

[関数を作成] を選択します。詳細については、「コンソールで Lambda 関数を作成する」を参照してください。

次のオプションを選択します。

[関数を作成] を選択し AWS Lambda 関数を作成します。

「関数」ページで、[構成] タブを選択します。

[環境変数] ペインで、[編集] を選択します。

次の環境変数を追加します。変数ごとに [環境変数を追加] を選択し、変数のキーと値を入力します。

[保存] を選択して環境変数を保存します。

[権限] ペインの [ロール名] で、実行ロールを選択して IAM コンソールで開きます。

[権限] タブで、[権限を追加]、[インラインポリシーを作成] をクリックします。

[JSON] を選択し、既存の JSON を次のポリシーに置き換えます。

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

[次へ] をクリックします。

「ポリシー詳細」に、「DetectAnomalies-access」などのポリシーの名前を入力します。

[ポリシーを作成] を選択します。

分析用の画像を 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 などの別のポリシー名を選択します。

コマンドプロンプトを開き、次のコマンドを入力します。

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

zip ファイル (boto3-layer.zip) の名前をメモします。それは、この手順のステップ 6 で必要になります。

AWS Lambda コンソールを https://console.aws.amazon.com/lambda/ で開きます。

ナビゲーションペインで [レイヤー] を選択します。

[レイヤーを作成] を選択します。

[名前] と [説明] の値を入力します。

[.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 関数の ARN は、Lambda コンソールの関数概要から取得できます。

   アクセスを提供するには、ユーザー、グループ、またはロールにアクセス許可を追加します。

   • AWS IAM Identity Center のユーザーとグループ:

     アクセス許可セットを作成します。「AWS IAM Identity Center ユーザーガイド」の「シークレットの作成と管理」の手順に従ってください。

   • ID プロバイダーを通じて IAM で管理されているユーザー:

     ID フェデレーションのロールを作成する。詳細については、「IAM ユーザーガイド」の「サードパーティー ID プロバイダー (フェデレーション) 用のロールの作成」を参照してください。

   • 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 関数を使用するときには、モデルを開始することを忘れないでください。