使用 Lambda 和 Python 检测图像中的标签 - Amazon Rekognition

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

使用 Lambda 和 Python 检测图像中的标签

AWS Lambda 是一项计算服务,无需预置或管理服务器即可使用它来运行代码。您可以从 Lambda 函数中调用 Rekognition API 操作。以下说明显示如何在 Python 中创建用来调用 DetectLabels 的 Lambda 函数。

Lambda 函数会调用 DetectLabels 并返回图像中检测到的一组标签和检测标签时所依据的置信度级别。

这些说明包括示例 Python 代码,该代码向您展示如何调用 Lambda 函数并向其提供来自 Amazon S3 存储桶或本地计算机的图像。

确保您选择的图像符合 Rekognition 的限制。有关图像文件类型和大小限制的信息,请参阅 Rekognition 中的指南和DetectLabels 配额以及 API 参考。

创建 Lambda 函数(控制台)

在此步骤中,您将创建一个空的 Lambda 函数和一个允许您的 Lambda 函数调用 DetectLabels 操作的 IAM 执行角色。在后续步骤中,您将添加源代码,也可以选择向 Lambda 函数添加层。

如果您使用的是存储在 Amazon S3 存储桶中的文档,则此步骤还演示了如何授予对存储您文档的存储桶的访问权限。

创建 AWS Lambda 函数(控制台)
  1. 登录 AWS Management Console 并打开 AWS Lambda 控制台,网址为https://console.aws.amazon.com/lambda/

  2. 选择 Create function (创建函数)。有关更多信息,请参阅使用控制台创建 Lambda 函数

  3. 选择以下选项:

    • 选择从头开始创作

    • 函数名称输入一个值。

    • 对于运行时系统,请选择最新版本的 Python。

    • 对于架构,选择 x86_64

  4. 选择创建函数以创建 AWS Lambda 函数。

  5. 在函数页面上,选择配置选项卡。

  6. 权限窗格的执行角色下,选择角色名称以在 IAM 控制台中打开该角色。

  7. 权限选项卡中,依次选择添加权限创建内联策略

  8. 选择 JSON 选项卡,并将该策略替换为以下策略:

    { "Version": "2012-10-17", "Statement": [ { "Action": "rekognition:DetectLabels", "Resource": "*", "Effect": "Allow", "Sid": "DetectLabels" } ] }
  9. 选择查看策略

  10. 输入策略的名称,例如 DetectLabels-access

  11. 选择创建策略

  12. 如果您要在 Amazon S3 存储桶中存储用于分析的文档,则必须添加 Amazon S3 访问策略。为此,请在 AWS Lambda 控制台中重复步骤 7 到 11 并执行以下更改。

    1. 对于步骤 8,请使用以下策略。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. 在步骤 10 中,选择不同的策略名称,例如 S3Bucket-access

(可选)创建层(控制台)

您无需执行此步骤即可使用 Lambda 函数和调用 DetectLabels

DetectLabels操作作为适用于 Python 的 AWS SDK (Boto3) 的一部分包含在默认 Lambda Python 环境中。

如果您的 Lambda 函数的其他部分需要最新的 AWS 服务更新,而这些更新不在默认 Lambda Python 环境中,则可以执行此步骤将最新版本的 Boto3 SDK 作为层添加到您的函数中。

要将 SDK 添加为层,请先创建一个包含 Boto3 SDK 的 zip 文件存档。然后,创建一个层并将 zip 文件存档添加到该层。有关更多信息,请参阅组合使用层与 Lambda 函数

创建并添加层(控制台)
  1. 打开命令提示符并输入以下命令,使用最新版本的 AWS SDK 创建部署包。

    pip install boto3 --target python/. zip boto3-layer.zip -r python/
  2. 记下您在本过程的步骤 8 中使用的压缩文件 (boto3-layer.zip) 的名称。

  3. 打开 AWS Lambda 控制台,网址为https://console.aws.amazon.com/lambda/

  4. 在导航窗格中,选择

  5. 选择创建层

  6. 名称描述输入值。

  7. 对于代码输入种类,选择上传 .zip 文件并选择上传

  8. 在对话框中,选择您在本过程步骤 1 中创建的 zip 文件存档 (boto3-layer.zip)。

  9. 对于兼容运行时系统,请选择最新版本的 Python。

  10. 选择创建以创建层。

  11. 选择导航窗格菜单图标。

  12. 在导航窗格中,选择函数

  13. 在资源列表中,选择您之前在中创建的函数。

  14. 选择节点选项卡。

  15. 部分,选择添加层

  16. 选择自定义层

  17. 自定义层中,选择您在步骤 6 中输入的层名称。

  18. 版本中,选择层版本,该版本应为 1。

  19. 选择 添加

添加 Python 代码(控制台)

在此步骤中,您将使用 Lambda 控制台代码编辑器,向您的Lambda 函数添加 Python 代码。该代码使用 DetectLabels 操作检测图像中的标签。它返回在图像中检测到的一组标签,以及检测标签所依据的置信度级别。

您为 DetectLabels 操作提供的文档可以位于 Amazon S3 存储桶或本地计算机中。

添加 Python 代码(控制台)
  1. 导航至代码选项卡。

  2. 在代码编辑器中,将 lambda_function.py 中的代码替换为以下代码:

    import boto3 import logging from botocore.exceptions import ClientError import json import base64 # Instantiate logger logger = logging.getLogger(__name__) # connect to the Rekognition client rekognition = boto3.client('rekognition') def lambda_handler(event, context): try: image = None if 'S3Bucket' in event and 'S3Object' in event: s3 = boto3.resource('s3') s3_object = s3.Object(event['S3Bucket'], event['S3Object']) image = s3_object.get()['Body'].read() elif 'image' in event: image_bytes = event['image'].encode('utf-8') img_b64decoded = base64.b64decode(image_bytes) image = img_b64decoded elif image is None: raise ValueError('Missing image, check image or bucket path.') else: raise ValueError("Only base 64 encoded image bytes or S3Object are supported.") response = rekognition.detect_labels(Image={'Bytes': image}) lambda_response = { "statusCode": 200, "body": json.dumps(response) } labels = [label['Name'] for label in response['Labels']] print("Labels found:") print(labels) except ClientError as client_err: error_message = "Couldn't analyze image: " + client_err.response['Error']['Message'] lambda_response = { 'statusCode': 400, 'body': { "Error": client_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
  3. 选择部署以部署您的 Lambda 函数。

添加 Python 代码(控制台)

现在您已创建 Lambda 函数,可以调用它来检测图像中的标签。

在此步骤中,您在计算机上运行 Python 代码,该代码会将本地图像或 Amazon S3 存储桶中的图像传递给您的 Lambda 函数。

确保在创建 Lambda 函数的同一 AWS 区域运行代码。您可以在 Lambda 控制台的函数详情页面的导航栏中查看 Lambda 函数的 AWS 区域。

如果 Lambda 函数返回超时错误,请延长 Lambda 函数的超时时间。有关更多信息,请参阅配置函数超时(控制台)

有关从您的代码调用 Lambda 函数的更多信息,请参阅调用 AWS Lambda 函数

试用您的 Lambda 函数
  1. 执行以下操作(如果尚未这样做):

    1. 确保用户拥有 lambda:InvokeFunction 权限。您可以使用以下策略:

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

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

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

    2. 安装和配置 AWS 适用于 Python 的开发工具包。有关更多信息,请参阅 步骤 2:设置 AWS CLI 和 AWS SDKs

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

    import boto3 import json import base64 import pprint # Replace with the name of your S3 bucket and image object key bucket_name = "name of bucket" object_key = "name of file in s3 bucket" # If using a local file, supply the file name as the value of image_path below image_path = "" # Create session and establish connection to client[' session = boto3.Session(profile_name='developer-role') s3 = session.client('s3', region_name="us-east-1") lambda_client = session.client('lambda', region_name="us-east-1") # Replace with the name of your Lambda function function_name = 'RekDetectLabels' def analyze_image_local(img_path): print("Analyzing local image:") with open(img_path, 'rb') as image_file: image_bytes = image_file.read() data = base64.b64encode(image_bytes).decode("utf8") lambda_payload = {"image": data} # Invoke the Lambda function with the event payload response = lambda_client.invoke( FunctionName=function_name, Payload=(json.dumps(lambda_payload)) ) decoded = json.loads(response['Payload'].read().decode()) pprint.pprint(decoded) def analyze_image_s3(bucket_name, object_key): print("Analyzing image in S3 bucket:") # Load the image data from S3 into memory response = s3.get_object(Bucket=bucket_name, Key=object_key) image_data = response['Body'].read() image_data = base64.b64encode(image_data).decode("utf8") # Create the Lambda event payload event = { 'S3Bucket': bucket_name, 'S3Object': object_key, 'ImageBytes': image_data } # Invoke the Lambda function with the event payload response = lambda_client.invoke( FunctionName=function_name, InvocationType='RequestResponse', Payload=json.dumps(event), ) decoded = json.loads(response['Payload'].read().decode()) pprint.pprint(decoded) def main(path_to_image, name_s3_bucket, obj_key): if str(path_to_image) != "": analyze_image_local(path_to_image) else: analyze_image_s3(name_s3_bucket, obj_key) if __name__ == "__main__": main(image_path, bucket_name, object_key)
  3. 运行该代码。如果文档在 Amazon S3 存储桶中,请确保它与您之前在 步骤 12 中指定的存储桶相同。

    如果成功,代码会针对文档中检测到的每个块类型返回部分 JSON 响应。