Individuazione di anomalie con unAWS Lambdafunzione

Accedere alla AWS Management Console e aprire la console di AWS Lambda all'indirizzo https://console.aws.amazon.com/lambda/.

Scegli Create function (Crea funzione). Per ulteriori informazioni, vedereCrea una funzione Lambda con la console.

Scegliete le seguenti opzioni.

ScegliCrea funzioneper creare ilAWS Lambdafunzione.

Nella pagina delle funzioni, scegli ilConfigurazionescheda.

SulVariabili d'ambienteriquadro, scegliModifica.

Aggiungere le seguenti variabili di ambiente. Per ogni variabile scegliAggiungi variabile di ambientequindi inserisci la chiave e il valore della variabile.

ScegliSalvaper salvare le variabili di ambiente.

SulAutorizzazionipannello, sottoNome del ruolo, scegli il ruolo di esecuzione per aprire il ruolo nella console IAM.

NelAutorizzazionischeda, scegliAggiungi autorizzazionie poiCrea una politica in linea.

ScegliJSONe sostituisci la politica esistente con la seguente politica.

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

Seleziona Successivo.

NelDettagli della politica, inserisci un nome per la politica, ad esempioDetectAnomalies-accesso.

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 politica. Sostituiscipercorso del bucket/cartellacon il percorso del bucket e della cartella Amazon S3 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 diverso per la politica, ad esempioAccesso al bucket S3.

Apri un prompt dei comandi e inserisci i seguenti comandi.

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

Nota il nome del file zip (boto3-layer.zip). Ne hai bisogno nella fase 6 di questa procedura.

Apri la console AWS Lambda all'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).

ScegliCarica un file.zipe scegliCarica.

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

Per tempi di esecuzione compatibili, scegliPython 3.9.

ScegliCreaper creare il livello.

Scegli l'icona del menu del pannello di navigazione.

Nel riquadro di navigazione, seleziona Funzioni.

Nell'elenco delle risorse, scegli la funzione che hai creato inFase 1: Creare unAWS Lambdafunzione (console).

Scegli la scheda Codice.

NelStratisezione, scegliAggiungi un livello.

ScegliLivelli personalizzati.

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

NelVersionescegli la versione del livello, che dovrebbe essere 1.

Scegli Add (Aggiungi).

Se non utilizzi la console Lambda, procedi come segue:

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

2. Apri la funzione Lambda che hai creato inFase 1: Creare unAWS Lambdafunzione (console).

Scegli la scheda Codice.

NelCodice sorgente, sostituisci il codice inlambda_function.pycon 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

ScegliDistribuisciper implementare la tua funzione Lambda.

Se non l'hai già fatto, procedi come segue:

1. Assicurati che l'utente che utilizza il codice client abbialambda:InvokeFunctionpermesso. Puoi usare le seguenti autorizzazioni.

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

   È possibile ottenere l'ARN per la funzione della funzione Lambda dalla panoramica delle funzioni nelConsolle 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 in IAM tramite un provider di identità:

     Crea un ruolo per la federazione delle identità. Segui le istruzioni riportate nella pagina Creating a role for a third-party identity provider (federation) (Creazione di un ruolo per un provider di identità di terze parti [federazione]) nella Guida per l'utente di IAM.

   • Utenti IAM:

     • Crea un ruolo che l'utente possa assumere. Per istruzioni, consulta la pagina Creating a role for an IAM user (Creazione di un ruolo per un utente IAM) nella Guida per l'utente di IAM.

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

2. Installa e configuraAWSSDK per Python. Per ulteriori informazioni, consulta Fase 4: Configurare il AWS CLI e AWS SDKs.

3. Avvia il modelloche hai specificato nel passaggio 7 diFase 1: Creare unAWS Lambdafunzione (console).

Salva il seguente codice in un file denominatoclient.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. Come argomento della riga di comando, fornisci il nome della funzione Lambda e il percorso di un'immagine locale che desideri analizzare. Ad esempio:

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

In caso di successo, l'output è una classificazione per le anomalie rilevate nell'immagine. Se non viene restituita una classificazione, valuta la possibilità di ridurre il valore di confidenza impostato nel passaggio 7 diFase 1: Creare unAWS Lambdafunzione (console).

Se hai finito con la funzione Lambda e il modello non è utilizzato da altre applicazioni,ferma il modello. Ricordati diavvia il modellola prossima volta che vuoi usare la funzione Lambda.