使用 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": "rekognition:DetectCustomLabels",
            "Resource": "*",
            "Effect": "Allow",
            "Sid": "DetectCustomLabels"
        }
    ]
}

选择下一步。

在策略详细信息中，输入策略的名称，例如 DetectCustomLabels-acces s。

选择创建策略。

如果供分析的图像存储在 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/
                

记下压缩文件的名称 (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 the Amazon Rekognition
Custom Labels model.
"""
import json
import base64
from os import environ
import logging
import boto3

from botocore.exceptions import ClientError

# Set up logging.
logger = logging.getLogger(__name__)

# Get the model ARN and confidence.
model_arn = environ['MODEL_ARN']
min_confidence = int(environ.get('CONFIDENCE', 50))

# Get the boto3 client.
rek_client = boto3.client('rekognition')


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:

        # Determine image source.
        if 'image' in event:
            # Decode the image
            image_bytes = event['image'].encode('utf-8')
            img_b64decoded = base64.b64decode(image_bytes)
            image = {'Bytes': img_b64decoded}


        elif 'S3Object' in event:
            image = {'S3Object':
                     {'Bucket':  event['S3Object']['Bucket'],
                      'Name': event['S3Object']['Name']}
                     }

        else:
            raise ValueError(
                'Invalid source. Only image base 64 encoded image bytes or S3Object are supported.')


        # Analyze the image.
        response = rek_client.detect_custom_labels(Image=image,
            MinConfidence=min_confidence,
            ProjectVersionArn=model_arn)

        # Get the custom labels
        labels = response['CustomLabels']

        lambda_response = {
            "statusCode": 200,
            "body": json.dumps(labels)
        }

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

        lambda_response = {
            'statusCode': 400,
            'body': {
                "Error": err.response['Error']['Code'],
                "ErrorMessage": error_message
            }
        }
        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)
            }
        }
        logger.error("Error function %s: %s",
            context.invoked_function_arn, format(val_error))

    return lambda_response

选择部署以部署您的 Lambda 函数。

确保您具有 lambda:InvokeFunction 权限。您可以使用以下策略。

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

您可以从 Lambda 控制台的函数概述中获取ARN适用于您的 Lambda 函数的函数。

要提供访问权限，请为您的用户、组或角色添加权限：

为 Python 安装和配置 AWS SDK。有关更多信息，请参阅 步骤 4：设置 AWS CLI 和 AWS 软件开发工具包。

启动模型，即您在 步骤 1：创建 AWS Lambda 函数（控制台） 的步骤 7 中指定的模型。

将以下代码保存到名为 client.py 的文件中。

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

"""
Purpose
Test code for running the Amazon Rekognition Custom Labels Lambda
function example code.
"""

import argparse
import logging
import base64
import json
import boto3

from botocore.exceptions import ClientError


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}

    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()

        # Get analysis results.
        result = analyze_image(args.function, args.image)
        status = result['statusCode']

        if status == 200:
            labels = result['body']
            labels = json.loads(labels)
            print(f"There are {len(labels)} labels in the image.")
            for custom_label in labels:
                confidence = int(round(custom_label['Confidence'], 0))
                print(
                    f"Label: {custom_label['Name']}: Confidence: {confidence}%")
        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 函数名称和要分析的图像。您可以提供指向本地图像的路径，或存储在 Amazon S3 存储桶中的图像的 S3 路径。例如：

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

如果图像位于 Amazon S3 存储桶中，请确保该存储桶与您在步骤 1：创建 AWS Lambda 函数（控制台）的步骤 15 中指定的存储桶相同。

如果成功，则会输出在图像中找到的标签列表。如果未返回标签，请考虑降低您在步骤 1：创建 AWS Lambda 函数（控制台）的步骤 7 中设置的置信度值。

如果已结束使用 Lambda 函数并且模型未被其他应用程序使用，请停止模型。下次要使用 Lambda 函数时，请记得启动模型。