Encontrando anomalias com uma função AWS Lambda

Faça login no AWS Management Console e abra o AWS Lambda console em https://console.aws.amazon.com/lambda/.

Escolha a opção Criar função. Para obter mais informações, consulte Criar uma função do Lambda no console.

Escolha as seguintes opções:

Escolha Criar função para criar a função do AWS Lambda .

Em sua página da função, escolha a guia Configuração.

No painel Variáveis de ambiente, escolha Editar.

Adicione as seguintes variáveis de ambiente: Para cada variável, escolha Adicionar variável de ambiente, e insira a chave e o valor da variável.

Escolha Salvar para salvar as variáveis de ambiente.

No painel Permissões, em Nome da função, escolha a função de execução para abrir a função no IAM console.

Na guia Permissões, escolha Adicionar permissões e Criar política em linha.

Escolha JSONe substitua a política existente pela política a seguir.

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

Escolha Próximo.

Em Detalhes da política, insira um nome para a política, como DetectAnomalies-access.

Escolha Criar política.

Se estiver armazenando imagens para análise em um bucket do Amazon S3, repita as etapas 10 a 14.

1. Para a etapa 11, use a política a seguir. Substituir bucket/folder path com o bucket do Amazon S3 e o caminho da pasta para as imagens que você deseja analisar.

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "S3Access",
               "Effect": "Allow",
               "Action": "s3:GetObject",
               "Resource": "arn:aws:s3:::bucket/folder path/*"
           }
       ]
   }

2. Para a etapa 13, escolha um nome de política diferente, como S3Bucket-access.

Abra um prompt de comando e execute os comandos a seguir.

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

Observe o nome do arquivo zip (boto3-layer.zip). Ele será necessário na etapa 6 deste procedimento.

Abra o AWS Lambda console em https://console.aws.amazon.com/lambda/.

No painel de navegação, escolha Camadas.

Escolha Criar camada.

Insira valores para Nome e Descrição.

Escolha Fazer upload de um arquivo .zip e escolha Fazer upload.

Na caixa de diálogo, escolha o arquivo de arquivos.zip (boto3-layer.zip) criada na etapa 1 desse procedimento.

Para runtimes compatíveis, escolha Python 3.9.

Escolha Criar para criar a camada.

Escolha o ícone do menu do painel de navegação.

Selecione Funções no painel de navegação.

Na lista de recursos, escolha a função que você criou em Etapa 1: criar uma AWS Lambda função (console).

Escolha a guia Código.

Na seção Camadas, escolha Adicionar uma camada.

Escolha camadas personalizadas.

Em Camadas personalizadas, escolha o nome da camada que você inseriu na etapa 6.

Em Versão, escolha a versão da camada, que deve ser 1.

Escolha Adicionar.

Se não estiver no console do Lambda, faça o seguinte:

1. Abra o AWS Lambda console em https://console.aws.amazon.com/lambda/.

2. Abra a função do Lambda que você criou em Etapa 1: criar uma AWS Lambda função (console).

Escolha a guia Código.

Em Código-fonte, substitua o código em lambda_function.py pelo seguinte:

# 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

Escolha Implantar para implantar sua função do Lambda.

Se ainda não tiver feito isso, faça o seguinte:

1. O usuário que está usando o código do cliente deve ter a permissão lambda:InvokeFunction. Você pode usar as seguintes permissões.

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

   Você pode obter a função ARN para sua função Lambda na visão geral da função no console Lambda.

   Para conceder acesso, adicione as permissões aos seus usuários, grupos ou perfis:

   • Usuários e grupos em AWS IAM Identity Center:

     Crie um conjunto de permissões. Siga as instruções em Criação de um conjunto de permissões no Guia do usuário do AWS IAM Identity Center .

   • Usuários gerenciados IAM por meio de um provedor de identidade:

     Crie um perfil para a federação de identidades. Siga as instruções em Criar uma função para um provedor de identidade terceirizado (federação) no Guia IAM do usuário.

   • IAMusuários:

     • Crie um perfil que seu usuário possa assumir. Siga as instruções em Criar uma função para um IAM usuário no Guia do IAM usuário.

     • (Não recomendado) Vincule uma política diretamente a um usuário ou adicione um usuário a um grupo de usuários. Siga as instruções em Adicionar permissões a um usuário (console) no Guia do IAM usuário.

2. Instale e configure AWS SDK para Python. Para obter mais informações, consulte Etapa 4: configurar o AWS CLI e AWS SDKs.

3. Inicie o modelo que você especificou na etapa 7 de Etapa 1: criar uma AWS Lambda função (console).

Salve o código a seguir em um arquivo chamado 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()

Execute o código. No argumento da linha de comando, forneça o nome da função do Lambda e o caminho para uma imagem local que você deseja analisar. Por exemplo:

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

Se for bem-sucedida, a saída será uma classificação das anomalias encontradas na imagem. Se uma classificação não for retornada, considere reduzir o valor de confiança que você definiu na etapa 7 de Etapa 1: criar uma AWS Lambda função (console).

Se você tiver concluído a função do Lambda e o modelo não for usado por outras aplicações, interrompa o modelo. Inicie o modelo na próxima vez que quiser usar a função do Lambda.