使用尋找異常AWS Lambda功能 - Amazon Lookout for Vision
步驟 1：建立AWS Lambda功能（控制台）步驟 2：（可選）創建一個層（控制台）第 3 步：添加 Python 代碼（控制台）步驟 4：嘗試您的 Lambda 函數

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

使用尋找異常AWS Lambda功能

AWS Lambda 是一項運算服務，可讓您執行程式碼，無需佈建或管理伺服器。例如，您可以分析從行動應用程式提交的影像，而不必建立伺服器來託管應用程式程式碼。以下說明顯示瞭如何在 Python 中創建一個調用 Lambda 函數DetectAnomalies。此函數會分析提供的影像，並傳回該影像中存在異常的分類。這些指示包括範例 Python 程式碼，示範如何使用 Amazon S3 儲存貯體中的映像呼叫 Lambda 函數，或從本機電腦提供的映像呼叫 Lambda 函數。

步驟 1：建立AWS Lambda功能（控制台）

在此步驟中，您將創建一個空的AWS函數和 IAM 執行角色，可讓您的函數呼叫DetectAnomalies操作。它還授予對存放映像以進行分析的 Amazon S3 儲存貯體的存取權。您也可以為下列項目指定環境變數：

  • 您希望 Lambda 函數使用的亞馬遜觀察視覺專案和模型版本。

  • 您希望模型使用的可信度限制。

稍後，您可以將原始程式碼和選擇性地新增至 Lambda 函數的圖層。

若要建立AWS Lambda功能（控制台）

  1. 請登入 AWS Management Console，並開啟位於 https://console.aws.amazon.com/lambda/ 的 AWS Lambda 主控台。

  2. 選擇 建立函數 。如需詳細資訊，請參閱〈使用主控台建立 Lambda 函數。

  3. 選擇下列選項。

    • 選擇 從頭開始撰寫。

    • 輸入以下項目的值函數名稱。

    • 對於运行时選擇蟒蛇。

  4. 選擇建立函數以建立AWS Lambda功能。

  5. 在功能頁面上，選擇配置標籤。

  6. 在「」環境變量」窗格中，選擇編輯。

  7. 新增下列環境變數。對於每個變量選擇加入環境變數然後輸入變量鍵和值。

    索引鍵 值

    專案名稱

    包含您要使用的模型的觀察視覺專案。

    模型版本

    您要使用的模型版本。

    信心

    模型對於預測是異常的可信度的最小值 (0-100)。如果信心較低，則分類被認為是正常的。

  8. 選擇儲存以儲存環境變數。

  9. 在「」權限窗格，「下」角色名稱」下方，選擇執行角色以在 IAM 主控台中開啟角色。

  10. 在權限」頁籤上，選擇新增權限然後建立內嵌政策。

  11. 選擇JSON並以下列原則取代現有原則。

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

  12. 選擇 下一步。

  13. 在政策詳情，輸入策略的名稱，例如DetectAnomalies-訪問。

  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，選擇不同的策略名稱，例如S3 儲存貯體存取。

步驟 2：（可選）創建一個層（控制台）

若要執行此範例，您不需要執行此步驟。該DetectAnomalies作業包含在預設的 Lambda Python 環境中，AWS開發套件為蟒蛇 (肉毒 3). 如果您的 Lambda 函數的其他部分需要最近AWS不在預設 Lambda Python 環境中的服務更新，請執行此步驟，將最新的 Boto3 SDK 版本新增為函數的一個層。

首先，您要建立一個包含 Boto3 SDK 的 .zip 檔案歸檔。然後建立圖層，並將 .zip 檔案封存新增至圖層。如需詳細資訊，請參閱〈搭配 Lambda 函數使用圖層。

建立和新增圖層 (控制台) 的步驟

  1. 打開命令提示符，然後輸入以下命令。

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

  2. 請記下壓縮檔案的名稱 (boto3-layer.zip)。您在此程序的步驟 6 中需要它。

  3. 在 https://console.aws.amazon.com/lambda/ 開啟 AWS Lambda 主控台。

  4. 在導覽窗格中，選擇 Layers (層)。

  5. 選擇 Create layer (建立 Layer)。

  6. 輸入 Name (名稱) 與 Description (描述) 的值。

  7. 選擇上傳一個 .zip 文件並選擇上传。

  8. 在對話方塊中，選擇您在此程序的步驟 1 中建立的 .zip 檔案封存 (boto3-layer.zip)。

  9. 如需相容的執行階段，請選擇蟒蛇 3.9。

  10. 選擇創建以建立圖層。

  11. 選擇導覽窗格功能表圖示。

  12. 在導覽窗格中，選擇函數。

  13. 在資源清單中，選擇您在其中建立的函數步驟 1：建立AWS Lambda功能（控制台）。

  14. 選取程式碼索引標籤。

  15. 在圖層區段中，選擇新增圖層。

  16. 選擇自訂圖層。

  17. 在自訂圖層，選擇您在步驟 6 中輸入的圖層名稱。

  18. 在版本選擇圖層版本，應為 1。

  19. 選擇 Add (新增)。

第 3 步：添加 Python 代碼（控制台）

在此步驟中，您可以使用 Lambda 主控台程式碼編輯器，將 Python 程式碼新增至 Lambda 函數。該代碼分析提供的圖像DetectAnomalies並返回一個分類（如果圖像異常為 true，如果圖像是正常的則為 false）。提供的映像檔可以位於 Amazon S3 儲存貯體中，或以 byte64 編碼的影像位元組提供。

若要新增 Python 程式碼 (主控台)

  1. 如果您不在 Lambda 主控台中，請執行下列動作：

    1. 在 https://console.aws.amazon.com/lambda/ 開啟 AWS Lambda 主控台。

    2. 開啟您在其中建立的 Lambda 函數步驟 1：建立AWS 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 儲存貯體的相關資訊，請參閱上載物件。

確保你運行相同的代碼AWS您在其中建立 Lambda 函數的區域。您可以檢視AWS在函數詳細資訊頁面的導覽列中，您 Lambda 函數的區域主控台。

如果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 函數函數的 ARN主控台。

      若要提供存取權，請新增許可到您的使用者、群組或角色：

      • AWS IAM Identity Center 中的使用者和群組：

        建立許可集合。請遵循《AWS IAM Identity Center 使用者指南》的建立許可集合中的指示。

      • 透過身分提供者在 IAM 中管理的使用者：

        建立聯合身分的角色。請遵循《IAM 使用者指南》的為第三方身分提供者 (聯合) 建立角色中的指示。

      • IAM 使用者：

    2. 安裝和配置AWS開發套件的 Python。如需詳細資訊，請參閱步驟 4：設定 AWS CLI 和 AWS 軟體開發套件。

    3. 啟動模型您在步驟 7 中指定的步驟 1：建立AWS Lambda功能（控制台）。

  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

    如果成功，則輸出為在影像中找到的異常分類。如果未傳回分類，請考慮降低您在步驟 7 中設定的信賴度值步驟 1：建立AWS Lambda功能（控制台）。

  4. 如果您已完成 Lambda 函數，且該模型未被其他應用程式使用，停止模型。請記住啟動模型下次您想要使用 Lambda 函數時。