Individuazione di anomalie con una funzione AWS Lambda

Accedi a AWS Management Console e apri la AWS Lambda console all'indirizzo https://console.aws.amazon.com/lambda/.

Scegli Crea funzione. Per ulteriori informazioni, consulta Creare una funzione Lambda con la console.

Scegli le opzioni seguenti.

Scegli Crea funzione per creare la funzione AWS Lambda .

Nella pagina della funzione, scegli la scheda Configurazione.

Nella sezione Variabili di ambiente, scegli Modifica.

Aggiungi le seguenti variabili di ambiente. Per ogni variabile, scegli Aggiungi variabile di ambiente, quindi immetti la chiave e il valore della variabile.

Scegli Salva per salvare le variabili di ambiente.

Nel riquadro Autorizzazioni, in Nome ruolo, scegli il ruolo di esecuzione per aprire il ruolo nella IAM console.

Nella scheda Autorizzazioni, scegli Aggiungi autorizzazioni, poi Crea policy inline.

Scegli JSONe sostituisci la politica esistente con la seguente politica.

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

Scegli Next (Successivo).

Nei dettagli della politica, inserisci un nome per la politica, ad esempio DetectAnomalies-access.

Scegli Create Policy (Crea policy).

Se stai archiviando immagini per l'analisi in un bucket Amazon S3, ripeti i passaggi da 10 a 14.

1. Per il passaggio 11, utilizza la seguente policy. Replace (Sostituisci) bucket/folder path con il bucket Amazon S3 e il percorso della cartella verso le immagini che desideri analizzare.

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

2. Per il passaggio 13, scegli un nome di policy diverso, ad esempio S3Bucket-access.

Apri un prompt di comandi e inserisci il comando seguente.

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

Annota il nome del file .zip (boto3-layer.zip). Questo valore sarà necessario nel passaggio 6 di questa procedura.

Apri la console all' AWS Lambda indirizzo. https://console.aws.amazon.com/lambda/

Nel riquadro di navigazione scegli Layers (Livelli).

Scegli Create layer (Crea livello).

Immetti i valori per Nome (Nome) e Description (Descrizione).

Seleziona Carica file .zip, quindi scegli Carica.

Nella finestra di dialogo, scegli l'archivio di file .zip (boto3-layer.zip) creato nel passaggio 1 di questa procedura.

Per runtime compatibili, scegli Python 3.9.

Scegli Crea per creare lo strato.

Scegli l'icona del menu del pannello di navigazione.

Nel riquadro di navigazione, seleziona Funzioni.

Nell'elenco delle risorse, scegli la funzione in cui è stata creata Fase 1: Creare una AWS Lambda funzione (console).

Scegli la scheda Codice.

Nella sezione Livelli, scegli Aggiungi un livello.

Scegliete Livelli personalizzati.

In Livelli personalizzati, scegli il nome del livello che hai inserito nel passaggio 6.

In Versione scegli la versione del livello, che dovrebbe essere 1.

Scegli Aggiungi.

Se non ti trovi nella console Lambda, esegui quanto segue:

1. Apri la console all' AWS Lambda indirizzo. https://console.aws.amazon.com/lambda/

2. Apri la funzione Lambda che hai creato in Fase 1: Creare una AWS Lambda funzione (console).

Scegli la scheda Codice.

In Codice sorgente, sostituisci il codice in lambda_function.py con quanto segue:

# 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

Per distribuire il codice della funzione Lambda, scegli Distribuisci.

Se non lo hai ancora fatto, esegui queste operazioni:

1. Assicurati che l'utente che utilizza il codice client disponga dell'autorizzazione. lambda:InvokeFunction Puoi utilizzare le seguenti autorizzazioni.

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

   Puoi ottenere la funzione ARN per la tua funzione Lambda dalla panoramica delle funzioni nella console Lambda.

   Per fornire l'accesso, aggiungi autorizzazioni ai tuoi utenti, gruppi o ruoli:

   • Utenti e gruppi in: AWS IAM Identity Center

     Crea un set di autorizzazioni. Segui le istruzioni riportate nella pagina Create a permission set (Creazione di un set di autorizzazioni) nella Guida per l'utente di AWS IAM Identity Center .

   • Utenti gestiti IAM tramite un provider di identità:

     Crea un ruolo per la federazione delle identità. Segui le istruzioni riportate in Creare un ruolo per un provider di identità di terze parti (federazione) nella Guida per l'IAMutente.

   • IAMutenti:

     • Crea un ruolo che l'utente possa assumere. Segui le istruzioni riportate in Creare un ruolo per un IAM utente nella Guida per l'IAMutente.

     • (Non consigliato) Collega una policy direttamente a un utente o aggiungi un utente a un gruppo di utenti. Segui le istruzioni riportate in Aggiungere autorizzazioni a un utente (console) nella Guida per l'IAMutente.

2. Installa e configura AWS SDK per Python. Per ulteriori informazioni, consulta Fase 4: Configurare e AWS CLIAWS SDKs.

3. Avvia il modello specificato nel passaggio 7 di Fase 1: Creare una AWS Lambda funzione (console).

Salva il seguente codice in un file denominato 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()

Eseguire il codice. Per l'argomento della riga di comando, fornite il nome della funzione Lambda e il percorso di un'immagine locale da analizzare. Per esempio:

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

In caso di successo, l'output è una classificazione delle anomalie rilevate nell'immagine. Se non viene restituita una classificazione, valuta la possibilità di ridurre il valore di confidenza impostato nel passaggio 7 di. Fase 1: Creare una AWS Lambda funzione (console)

Se hai completato il processo con la funzione Lambda e il modello non viene utilizzato da altre applicazioni, interrompi il modello. Ricordati di avviare il modello la prossima volta che desideri utilizzare la funzione Lambda.