Finding anomalies with an AWS Lambda function

Sign in to the AWS Management Console and open the AWS Lambda console at https://console.aws.amazon.com/lambda/.

Choose Create function. For more information, see Create a Lambda Function with the Console.

Choose the following options.

Choose Create function to create the AWS Lambda function.

On the function page, Choose the Configuration tab.

On the Environment variables pane, choose Edit.

Add the following environment variables. For each variable choose Add enviroment variable and then enter the variable key and value.

Choose Save to save the environment variables.

On the Permissions pane, Under Role name, choose the execution role to open the role in the IAM console.

In the Permissions tab, choose Add permissions and then Create inline policy.

Choose JSON and replace the existing policy with the following policy.

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

Choose Next.

In Policy details, enter a name for the policy, such as DetectAnomalies-access.

Choose Create policy.

If you are storing images for analysis in an Amazon S3 bucket, repeat steps 10–14.

1. For step 11, use the following policy. Replace bucket/folder path with the Amazon S3 bucket and folder path to the images that you want to analyze.

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

2. For step 13, choose a different policy name, such as S3Bucket-access.

Open a command prompt and enter the following commands.

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

Note the name of the zip file (boto3-layer.zip). You need it in step 6 of this procedure.

Open the AWS Lambda console at https://console.aws.amazon.com/lambda/.

In the navigation pane, choose Layers.

Choose Create layer.

Enter values for Name and Description.

Choose Upload a .zip file and choose Upload.

In the dialog box, choose the .zip file archive (boto3-layer.zip) that you created in step 1 of this procedure.

For compatible runtimes, choose Python 3.9.

Choose Create to create the layer.

Choose the navigation pane menu icon.

In the navigation pane, choose Functions.

In the resources list, choose the function that you created in Step 1: Create an AWS Lambda function (console).

Choose the Code tab.

In the Layers section, choose Add a layer.

Choose Custom layers.

In Custom layers, choose the layer name that you entered in step 6.

In Version choose the layer version, which should be 1.

Choose Add.

If you're not in the Lambda console, do the following:

1. Open the AWS Lambda console at https://console.aws.amazon.com/lambda/.

2. Open the Lambda function you created in Step 1: Create an AWS Lambda function (console).

Choose the Code tab.

In Code source, replace the code in lambda_function.py with the following:

# 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

Choose Deploy to deploy your Lambda function.

If you haven't already done so, do the following:

1. Make sure the user using the client code has lambda:InvokeFunction permission. You can use the following permissions.

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

   You can get the ARN for your Lambda function function from the function overview in the Lambda console.

   To provide access, add permissions to your users, groups, or roles:

   • Users and groups in AWS IAM Identity Center:

     Create a permission set. Follow the instructions in Create a permission set in the AWS IAM Identity Center User Guide.

   • Users managed in IAM through an identity provider:

     Create a role for identity federation. Follow the instructions in Create a role for a third-party identity provider (federation) in the IAM User Guide.

   • IAM users:

     • Create a role that your user can assume. Follow the instructions in Create a role for an IAM user in the IAM User Guide.

     • (Not recommended) Attach a policy directly to a user or add a user to a user group. Follow the instructions in Adding permissions to a user (console) in the IAM User Guide.

2. Install and configure AWS SDK for Python. For more information, see Step 4: Set up the AWS CLI and AWS SDKs.

3. Start the model that you specified in step 7 of Step 1: Create an AWS Lambda function (console) .

Save the following code to a file named 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()

Run the code. For the command line argument, supply the Lambda function name and the path to a local image that you want to analyze. For example:

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

If successful, the output is a classification for the anomalies found in the image. If a classification isn't returned, consider lowering the confidence value that you set in step 7 of Step 1: Create an AWS Lambda function (console).

If you have finished with the Lambda function and the model isn't used by other applications, stop the model. Remember to start the model the next time you want use the Lambda function.