Búsqueda de anomalías con una función AWS Lambda

Inicie sesión en AWS Management Console y abra la AWS Lambda consola en https://console.aws.amazon.com/lambda/.

Seleccione Crear función. Para obtener más información, consulte Crear una función de Lambda con la consola.

Elija las siguientes opciones.

Elija Crear función para crear la función AWS Lambda .

En la página de la función, seleccione la pestaña Configuración.

En el panel Variables de entorno, elija Editar.

Añada las siguientes variables de entorno. En cada variable, elija Añadir variable de entorno y luego introduzca la clave y el valor de la variable.

Elija Guardar para guardar las variables de entorno.

En el panel de permisos, en Nombre del rol, elija el rol de ejecución para abrir el rol en la IAM consola.

En la pestaña Permisos, elija Agregar permisos y después Crear política insertada.

Elija JSONy sustituya la política existente por la siguiente política.

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

Elija Next (Siguiente).

En Detalles de la política, ingresa un nombre para la política, como DetectAnomalies-access.

Elija Crear política.

Si va a almacenar imágenes para analizarlas en un bucket de Amazon S3, repita los pasos del 10 al 14.

1. En el paso 11, utilice la siguiente política. Reemplazar bucket/folder path con la ruta del depósito y la carpeta de Amazon S3 a las imágenes que desee analizar.

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

2. En el paso 13, elija un nombre de política diferente, como S3Bucket-access.

Abra el símbolo del sistema y escriba el comando siguiente.

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

Apunte el nombre del archivo zip (boto3-layer.zip). Lo necesitará más tarde en el paso 6 de este mismo procedimiento.

Abre la consola en AWS Lambda . https://console.aws.amazon.com/lambda/

En el panel de navegación, elija Capas.

Elija Crear capa.

Introduzca los valores correspondientes en Nombre y Descripción.

Elija Cargar archivo .zip y luego Cargar.

En el cuadro de diálogo, elija el archivo .zip (boto3-layer.zip) que creó en el paso 1 de este procedimiento.

Con tiempos de ejecución compatibles, elija Python 3.9.

Elija Crear para crear la capa.

Elija el icono del menú del panel de navegación.

Seleccione Funciones en el panel de navegación.

En la lista de recursos, elija la función que haya creado en Paso 1: Crear una AWS Lambda función (consola).

Elija la pestaña Código.

En la sección Capas, elija Agregar una capa.

Elija Capas personalizadas.

En Capas personalizadas, elija el nombre de la capa que escogió en el paso 6.

En Versión, elija la versión de la capa, que debe ser 1.

Elija Agregar.

Si no se encuentra en la consola de Lambda, haga lo siguiente:

1. Abra la AWS Lambda consola en https://console.aws.amazon.com/lambda/.

2. Abra la función de Lambda que creó en Paso 1: Crear una AWS Lambda función (consola).

Elija la pestaña Código.

En Código fuente, sustituya el código en lambda_function.py por lo siguiente:

# 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

Elija Implementar para implementar la función de Lambda.

Si aún no lo ha hecho, haga lo siguiente:

1. Asegúrese de que el usuario que usa el código de cliente tiene permiso lambda:InvokeFunction. Puede usar los siguientes permisos.

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

   Puede obtener la función ARN para su función Lambda en la descripción general de la función en la consola de Lambda.

   Para dar acceso, agregue permisos a los usuarios, grupos o roles:

   • Usuarios y grupos en: AWS IAM Identity Center

     Cree un conjunto de permisos. Siga las instrucciones de Creación de un conjunto de permisos en la Guía del usuario de AWS IAM Identity Center .

   • Usuarios gestionados IAM a través de un proveedor de identidad:

     Cree un rol para la federación de identidades. Siga las instrucciones de la Guía del IAM usuario sobre cómo crear un rol para un proveedor de identidades externo (federación).

   • IAMusuarios:

     • Cree un rol que el usuario pueda aceptar. Siga las instrucciones de la Guía del IAMusuario sobre cómo crear un rol para un IAM usuario.

     • (No recomendado) Adjunte una política directamente a un usuario o añada un usuario a un grupo de usuarios. Siga las instrucciones de Añadir permisos a un usuario (consola) de la Guía del IAM usuario.

2. Instalar y configurar AWS SDK para Python. Para obtener más información, consulte Paso 4: Configura el AWS CLI y AWS SDKs.

3. Inicie el modelo que eligió en el paso 7 de Paso 1: Crear una AWS Lambda función (consola).

Guarde el siguiente código en un archivo denominado 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()

Ejecute el código. En el argumento de la línea de comandos, indique el nombre de la función de Lambda y la ruta a la imagen local que desee analizar. Por ejemplo:

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

Si se ejecuta correctamente, el resultado es una clasificación de las anomalías encontradas en la imagen. Si no devuelve ninguna clasificación, considere la posibilidad de reducir el valor de confianza que eligió en el paso 7 de Paso 1: Crear una AWS Lambda función (consola).

Si ha terminado con la función de Lambda y otras aplicaciones no utilizan el modelo, detenga el modelo. Recuerde iniciar el modelo la próxima vez que desee utilizar la función de Lambda.