Automate detection of broken utility poles using the Amazon Rekognition Custom Labels SDK

Domestic infrastructure issues are a pain for everyone involved. Not only does it negatively affect customer satisfaction, it also has a cascading effect on businesses and their bottom line in terms of financials. Electric utility poles, for example, are an example of a cumbersome infrastructure issue to resolve. Normally, the standard wooden distribution pole is expected to last roughly 50 years. However, occasionally these poles need to be replaced earlier due to unexpected incidents such as accidents, severe weather disasters, or even power line relocation. The industry standard right now is to use drone or street cameras to generate images of these broken poles. The poles in the images are then manually inspected to ensure they’re in good condition and don’t require repair or replacement. As you can imagine, the process of determining whether or not these poles need replacement is a time-consuming and manual task that is susceptible to human error or neglect.

To address this, we propose a solution using Amazon Rekognition Custom Labels. You can feed images of utility poles, taken from street cameras or from drones, into a machine computer vision model trained on Amazon Rekognition Custom Labels to automatically detect whether a utility pole is in good condition or damaged.

Amazon Rekognition is a computer vision service within the AWS AI/ML stack. It allows for the automation of image and videos analysis. With Amazon Rekognition, you can identify objects, people, text, scenes, inappropriate content, and activities in images and videos.

We use Amazon Rekognition Custom Labels for our solution, which enables us to create custom machine learning (ML) models to analyze images. With Amazon Rekognition Custom Labels, you can train a robust, deployable model with a few images as opposed to thousands of images.

For our use case, we use images of electric poles. The following is an example of a normal pole.

The following image is an example of a damaged pole.

If your images are already labeled, Amazon Rekognition Custom Labels can begin training in just a few clicks on the Amazon Rekognition console. If not, you can label them directly within the Amazon Rekognition Custom Labels labeling user interface, or use another service such as Amazon SageMaker Ground Truth to label them. After you train your image set with Amazon Rekognition Custom Labels, it can produce a custom image computer vision model for you in just a few hours. After this custom model is trained, you can use it as an endpoint to make inferences on new images.

In this post, we use the Amazon Rekognition Custom Labels API and the AWS SDK to show how easily you can integrate this technology into your applications.

Prepare dataset bucket with images

As with all ML models, we begin with some data—for this post, images of broken and not broken utility poles. This dataset is fully labeled and stored in Amazon Simple Storage Service (Amazon S3). The location of this dataset is fed into the SDK to train the model.

Train the model

Now that our labeled data is in our S3 bucket, let’s create the model for our data.

Import the necessary libraries:
```
import boto3
```
Because Amazon Rekognition Custom Labels requires you to create a project for a given use case, we use the following function to create a project:

def create_project(project_name):

    client=boto3.client('rekognition')

    #Create a project
    print('Creating project:' + project_name)
    response=client.create_project(ProjectName=project_name)
    arn = response['ProjectArn']
    print('project ARN: ' + response['ProjectArn'])
    return arn

The function returns the project ARN, which you should write down or store in a variable because you use it later to train the model.

We next define a method to train a model using Amazon Rekognition Custom Labels. This method requires the project_arn as an input (the project ARN variable that you saved earlier), a unique version_name for this version of the model, out_config to inform where to store training results, as well as locations of the training and test data manifest files.

Run the following train_model() method, and make a note of the project version ARN to use later:

# Train model helper method
import json

def train_model(project_arn, version_name, output_config, training_dataset,testing_dataset):

    client=boto3.client('rekognition')
    
    print('Starting training of: ' + version_name)
      
    try:
        response=client.create_project_version(ProjectArn=project_arn, 
            VersionName=version_name,
            OutputConfig=output_config,
            TrainingData=training_dataset,
            TestingData=testing_dataset)

        # Wait for the project version training to complete
        project_version_training_completed_waiter = client.get_waiter('project_version_training_completed')
        project_version_training_completed_waiter.wait(ProjectArn=project_arn,
        VersionNames=[version_name])
    
        #Get the completion status
        describe_response=client.describe_project_versions(ProjectArn=project_arn,
            VersionNames=[version_name])
        for model in describe_response['ProjectVersionDescriptions']:
            print('Project Version ARN: ' + model['ProjectVersionArn'])
            print("Status: " + model['Status'])
            print("Message: " + model['StatusMessage']) 
    except Exception as e:
        print(e)

    print('Done...')

Next, call the train_model() method, as shown in the following code:

project_arn='arn:aws:rekognition:us-east-1:xxxxxxxxxxxx:project/project_utility_pole_blog/yyyyyyyyyyy'

version_name='v1'

output_config = json.loads('{"S3Bucket":"my-bucket", "S3KeyPrefix":"blog/REK-CL/utilitypoles/"}')

training_dataset= json.loads('{"Assets": [{ "GroundTruthManifest": { "S3Object": { "Bucket": "my-bucket", "Name": "datasets/cables-ds/manifests/output/output.manifest" } } } ] }')

testing_dataset= json.loads('{"AutoCreate":true}')

train_model(project_arn, version_name, output_config, training_dataset, testing_dataset)

We set "AutoCreate":true for testing_dataset because we’re using Amazon Rekognition Custom Labels to split the training data randomly into an 80/20 split. Alternatively, you can specify the testing_dataset as a manifest file, just as is done for the training_dataset in the preceding code, if you have a separate test dataset.

Training can take a couple of hours to complete.

You can get the current status by calling DescribeProjectVersions and, when it’s complete, calling DescribeProjectVersions to get the training results and evaluate the model:

def describe_model(project_arn, version_name):

    client=boto3.client('rekognition')
    
    response=client.describe_project_versions(ProjectArn=project_arn,
        VersionNames=[version_name])

    for model in response['ProjectVersionDescriptions']:
        print(json.dumps(model,indent=4,default=str))

Model results and enhancement

We train two models in this post. The first model takes 80 images of good and broken utility poles, which are equally split between the good and broken. These images are fed into Amazon Rekognition Custom Labels and the model metrics are evaluated.

For the second model, instead of feeding the raw images as they are, we do some data augmentation on these images, which is common in computer vision problems. Amazon Rekognition Custom Labels doesn’t do data augmentation by itself because it doesn’t know your images too well. Therefore, we recommend explicitly doing image augmentation for scenarios where you want to further improve your model metrics.

We then compare how the model metrics such as accuracy and AUC score compare for the original model and the enhanced model.

Image augmentation is accomplished using the following code:

import random
import numpy as np
from scipy import ndarray
import skimage as sk
from skimage import transform
from skimage import util

def random_rotation(image_array: ndarray):
    # pick a random degree of rotation between 25% on the left and 25% on the right
    random_degree = random.uniform(-25, 25)
    return sk.transform.rotate(image_array, random_degree, preserve_range=True)

def horizontal_flip(image_array: ndarray):
    # horizontal flip doesn't need skimage, it's easy as flipping the image array of pixels !
    return image_array[:, ::-1]

The following is our original image.

Random rotation of that image produces the following rotated image.

Our original data of 80 images is randomly split into 60 images of training data and 20 images of test data. The original model is built using Amazon Rekognition Custom Labels on 60 images of the training data.

For building our enhanced model, we augmented the 60 training data images by running them through the preceding code, which involves rotation and horizontal flip. That increases the training data size from 60 to 180. We build a new model using this dataset.

After we build these two models, we run test data these models to see how the results compare. We use the following code to obtain the results:

# Helper method that makes a inference on a Custom Labels trained model for a binary classification problem
import boto3
import io

from botocore.exceptions import ClientError

def getLabelAndConfidenceBinary(bucket, image, model):
    """
    :param bucket: The name of the S3 bucket that contains the image that you want to analyze.
    :param image: The name of the image that you want to analyze.
    :param model: The ARN of the Amazon Rekognition Custom Labels model that you want to use.
    """
    
    rek_client=boto3.client('rekognition')
    s3_connection = boto3.resource('s3')
    
    # set minimum confidence to 0 as we are interested in results for all condidence levels
    min_confidence = 0
    
    try:
        #Get image from S3 bucket.
        s3_object = s3_connection.Object(bucket, image)
        s3_response = s3_object.get()

        #Call DetectCustomLabels 
        response = rek_client.detect_custom_labels(Image={'S3Object': {'Bucket': bucket, 'Name': image}},
            MinConfidence=min_confidence,
            ProjectVersionArn=model)
    
    except ClientError as err:
        print("Exception in get_custom_labels()")
        raise
        
    if response['CustomLabels'][0]["Confidence"] >= response['CustomLabels'][1]["Confidence"]:
        return response['CustomLabels'][0]["Name"], response['CustomLabels'][0]["Confidence"]
    else:
        return response['CustomLabels'][1]["Name"], response['CustomLabels'][1]["Confidence"]


# Helper method that iterates through an S3 bucket and retrieves image labels
def getImageNames(bucket, prefix):
    import boto3
    client = boto3.client('s3')
    paginator = client.get_paginator('list_objects_v2')
    result = paginator.paginate(Bucket=bucket, Prefix=prefix)
    images = []
    for page in result:
        if "Contents" in page:
            for key in page[ "Contents" ]:
                keyString = key[ "Key" ]
                name = keyString.split("/")[-1]
                iclass = name.split("-")[0]
                if len(iclass)>0:
                    images.append([keyString, iclass])
    return images

# This code loops through all images and creates a y_true list objects based on image labels
image_folder = "blog/REK-CL/utilitypoles/new_test_images/"
y_true = []
names = []
test_images = getImageNames('my-bucket', image_folder)
for image in test_images:
    iclass = image[1]
    name = image[0]
    names.append(name)
    if iclass=="bad":
        y_true.append(1)
    else:
        y_true.append(0)

# The helper method below makes predictions on a deploy Custom Labels model and returns predictions
# Custom Labels associates confidence level to the label it predicts
# In the code below, we are turning that to a [0 to 1] probability scale, where 1 indicates a broken pole 
def getPredictions(model, bucket, images):
    y_pred = []
    y_prob = []
    for image in images:
        labelconf = getLabelAndConfidenceBinary(bucket, image[0], model)
        if labelconf[0]=="broken":
            y_pred.append(1)
            prob = labelconf[1]/100.0
            y_prob.append(prob)
        if labelconf[0]=="good":
            y_pred.append(0)
            prob = 1.0 - labelconf[1]/100.0
            y_prob.append(prob)
        if labelconf[0]=="":
            raise Exception("Invalid label")
    return y_pred, y_prob

bucket = "my-bucket"
# Assign Project Version ARN returned by train_model() below for both the models
original_model = "arn:aws:rekognition:us-east-1:xxxxxxxxxxxx:project/upole-new-aug14/version/upole-new-aug14.2021-08-14T17.20.06/1628976006624"
enhanced_model = "arn:aws:rekognition:us-east-1: xxxxxxxxxxxx:project/upole-new-aug14/version/upole-new-aug14.2021-08-14T18.36.45/1628980605880"
y_pred_original, y_prob_original = getPredictions(original_model, bucket, test_images)
y_pred_enhanced, y_prob_enhanced = getPredictions(enhanced_model, bucket, test_images)


from sklearn.metrics import accuracy_score
print("Original model accuracy = ", round(accuracy_score(y_true, y_pred_original), 2))
print("Enhanced model accuracy = ", round(accuracy_score(y_true, y_pred_enhanced), 2))

Original model accuracy =  0.79
Enhanced model accuracy =  0.89

import numpy as np
from sklearn import metrics
def calculateAUC(y_true, y_prob, pos_label):
    fpr, tpr, thresholds = metrics.roc_curve(y_true, np.array(y_prob), pos_label=pos_label)
    return metrics.auc(fpr, tpr)

print("Original model AUC = ", round(calculateAUC(y_true, y_prob_original, 1),2))
print("Enhanced model AUC = ", round(calculateAUC(y_true, y_prob_enhanced, 1),2))

Original model AUC =  0.92
Enhanced model AUC =  0.96

Performance review

As we can observe from the model results, doing image augmentation has helped improve model accuracy significantly, from 0.79 to 0.89, and model AUC from 0.92 to 0.96.

Although the original model gave good results as is, doing image augmentation has further improved the results. You don’t necessarily need to augment your images all the time, but you can employ this technique to see if it can further improve your model.

Clean up

After you finish using this solution, it’s important that you stop your model to stop accruing charges. To delete your project, simply run the following function:

def delete_project(project_arn):

	client = boto3.client(‘rekognition’)
	
	print(‘Deleting project: ’ + project_arn)
	response = client.delete_project(ProjectArn=project_arn)
	print(‘Status: ’ + response[‘Status’])
	print(‘Done…’)

Conclusion

In this post, we successfully trained, evaluated, and inferred on a custom ML model for detecting broken and damaged utility poles. This once time-consuming and manual task was normally very susceptible to human error or neglect, but by using Amazon Rekognition Custom Labels, we were able to make this process quicker while maintaining accuracy.

For more information about using custom labels, see What Is Amazon Rekognition Custom Labels?

About the Authors:

Winston Nwanne is an AWS Solutions Architect working with public sector partners. He specializes in AI/ML and has helped customers and partners expand their capabilities within the AWS Cloud. Apart from supporting customers, he likes to read, make YouTube videos about financial literacy, play basketball, and spend time with his family.

Raju Penmatcha is a Senior AI/ML Specialist Solutions Architect at AWS. He works with education, government, and nonprofit customers on machine learning and artificial intelligence related projects, helping them build solutions using AWS. When not helping customers, he likes traveling to new places.