Détecter des anomalies à l'aide d'unAWS Lambdafonction

Connectez-vous à la AWS Management Console et ouvrez la console AWS Lambda à l'adresse https://console.aws.amazon.com/lambda/.

Sélectionnez Créer une fonction. Pour plus d'informations, voirCréation d'une fonction Lambda avec la console.

Choisissez les options suivantes.

ChoisissezCréer une fonctionpour créer leAWS Lambdafonction.

Sur la page des fonctions, choisissezConfigurationonglet.

Sur leVariables d'environnementvolet, choisissezModifier.

Ajoutez les variables d'environnement suivantes. Pour chaque variable, choisissezAjouter une variable d'environnementpuis entrez la clé et la valeur de la variable.

ChoisissezEnregistrerpour enregistrer les variables d'environnement.

Sur leAutorisationsvolet, En dessousNom du rôle, choisissez le rôle d'exécution pour ouvrir le rôle dans la console IAM.

Dans leAutorisationsonglet, choisissezAjouter des autorisationspuisCréer une politique en ligne.

ChoisissezJASONet remplacez la politique existante par la politique suivante.

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

Choisissez Suivant.

DansDétails de la politique, entrez un nom pour la politique, tel queDetectAnomalies-accès.

Choisissez Create Policy (Créer une politique).

Si vous stockez des images à des fins d'analyse dans un compartiment Amazon S3, répétez les étapes 10 à 14.

1. Pour l'étape 11, appliquez la politique suivante. Remplacerchemin d'accès au bucket ou au dossieravec le chemin du compartiment et du dossier Amazon S3 vers les images que vous souhaitez analyser.

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

2. Pour l'étape 13, choisissez un autre nom de politique, tel queAccès au compartiment S3.

Ouvrez une invite de commandes et entrez les commandes suivantes.

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

Notez le nom du fichier zip (boto3-layer.zip). Vous en aurez besoin à l'étape 6 de cette procédure.

Ouvrez la console AWS Lambda à l'adresse https://console.aws.amazon.com/lambda/.

Choisissez Layers dans le volet de navigation.

Sélectionnez Créer un calque.

Saisissez un Name (Nom) et une Description pour la règle.

ChoisissezCharger un fichier .zipet choisissezUploader.

Dans la boîte de dialogue, choisissez l'archive de fichiers .zip (boto3-layer.zip) que vous avez créée à l'étape 1 de cette procédure.

Pour des environnements d'exécution compatibles, choisissezPython 3.9.

ChoisissezCréezpour créer la couche.

Choisissez l'icône du menu du volet de navigation.

Dans le volet de navigation, choisissez Fonctions.

Dans la liste des ressources, choisissez la fonction que vous avez créée dansÉtape 1 : Création d'unAWS Lambdafonction (console).

Cliquez sur l'onglet Code.

Dans leCouchessection, choisissezAjouter une couche.

ChoisissezCouches personnalisées.

DansCouches personnalisées, choisissez le nom de la couche que vous avez saisi à l'étape 6.

DansVersionchoisissez la version de la couche, qui doit être 1.

Choisissez Add (Ajouter).

Si vous n'êtes pas dans la console Lambda, procédez comme suit :

1. Ouvrez la console AWS Lambda à l'adresse https://console.aws.amazon.com/lambda/.

2. Ouvrez la fonction Lambda que vous avez créée dansÉtape 1 : Création d'unAWS Lambdafonction (console).

Cliquez sur l'onglet Code.

DansSource du code, remplacez le code danslambda_function.pyavec les éléments suivants :

# 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

ChoisissezDéployerpour déployer votre fonction Lambda.

Si ce n'est pas déjà fait, procédez comme suit :

1. Assurez-vous que l'utilisateur qui utilise le code client possèdelambda:InvokeFunctionautorisation. Vous pouvez utiliser les autorisations suivantes.

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

   Vous pouvez obtenir l'ARN de votre fonction Lambda à partir de la vue d'ensemble des fonctions dansConsole Lambda.

   Pour activer l'accès, ajoutez des autorisations à vos utilisateurs, groupes ou rôles :

   • Utilisateurs et groupes dans AWS IAM Identity Center :

     Créez un jeu d'autorisations. Suivez les instructions de la rubrique Création d'un jeu d'autorisations du Guide de l'utilisateur AWS IAM Identity Center.

   • Utilisateurs gérés dans IAM par un fournisseur d'identité :

     Créez un rôle pour la fédération d'identité. Pour plus d'informations, voir la rubrique Création d'un rôle pour un fournisseur d'identité tiers (fédération) du Guide de l'utilisateur IAM.

   • Utilisateurs IAM :

     • Créez un rôle que votre utilisateur peut assumer. Suivez les instructions de la rubrique Création d'un rôle pour un utilisateur IAM du Guide de l'utilisateur IAM.

     • (Non recommandé) Attachez une politique directement à un utilisateur ou ajoutez un utilisateur à un groupe d'utilisateurs. Suivez les instructions de la rubrique Ajout d'autorisations à un utilisateur (console) du Guide de l'utilisateur IAM.

2. Installation et configurationAWSSDK pour Python. Pour plus d'informations, veuillez consulter Étape 4 : Configurez le AWS CLI and AWS SDKs.

3. Démarrez le modèleque vous avez spécifié à l'étape 7 deÉtape 1 : Création d'unAWS Lambdafonction (console).

Enregistrez le code suivant dans un fichier nommé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()

Exécutez le code. Pour l'argument de ligne de commande, indiquez le nom de la fonction Lambda et le chemin d'accès à l'image locale que vous souhaitez analyser. Par exemple :

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

En cas de succès, le résultat est une classification des anomalies détectées dans l'image. Si aucune classification n'est renvoyée, envisagez de réduire la valeur de confiance que vous avez définie à l'étape 7 deÉtape 1 : Création d'unAWS Lambdafonction (console).

Si vous avez terminé d'utiliser la fonction Lambda et que le modèle n'est pas utilisé par d'autres applications,arrêter le modèle. N'oubliez pas dedémarrer le modèlela prochaine fois que vous voudrez utiliser la fonction Lambda.